/**
 * Starter core. Load the core configuration file, and initialize the startup sequence through {@link ModuleManager}.
 */
// 由server-starter和server-starter-es7调用server-bootstrap
// server-starter和server-starter-es7的区别在于maven中引入的存储模块Module不同
// server-starter      storage-elasticsearch-plugin
// server-starter-es7  storage-elasticsearch7-plugin
@Slf4j
public class OAPServerBootstrap {
    public static void start() {
        // // 初始化mode为init或者no-init,表示是否初始化例如:底层存储组件等
        String mode = System.getProperty("mode");
        RunningMode.setMode(mode);

        // 初始化ApplicationConfigurationd的加载器
        ApplicationConfigLoader configLoader = new ApplicationConfigLoader();
        // 初始化Module的加载管理器
        ModuleManager manager = new ModuleManager();
        try {
            // 加载yml生成ApplicationConfiguration配置
            ApplicationConfiguration applicationConfiguration = configLoader.load();
            // 初始化模块 通过spi获取所有Module实现,基于yml配置加载spi中存在的相关实现
            manager.init(applicationConfiguration);

            // 对 telemetry 模块进行特殊处理, 添加 uptime 的 Gauge, 用于标识服务开始运行
            manager.find(TelemetryModule.NAME)
                   .provider()
                   .getService(MetricsCreator.class)
                   .createGauge("uptime", "oap server start up time", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE)
                   // Set uptime to second
                   .setValue(System.currentTimeMillis() / 1000d);
            // 如果是 Init 模式, 现在就直接退出, 否则就继续运行
            if (RunningMode.isInitMode()) {
                log.info("OAP starts up in init mode successfully, exit now...");
                System.exit(0);
            }
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
            System.exit(1);
        }
    }
}

/**
 * Initialize collector settings with following sources. Use application.yml as primary setting, and fix missing setting
 * by default settings in application-default.yml.
 * <p>
 * At last, override setting by system.properties and system.envs if the key matches moduleName.provideName.settingKey.
 */
// application.yml加载类, 三层结构:模块定义名.模块提供名.属性key
// ApplicationConfiguration 的辅助类，将 application.yml 配置文件加载到内存， 设置 selector 对应的 Provider 的配置信息
@Slf4j
public class ApplicationConfigLoader implements ConfigLoader<ApplicationConfiguration> {
    // 当不配置模块提供者时，使用"-"
    private static final String DISABLE_SELECTOR = "-";
    // 该字段选择模块提供者
    private static final String SELECTOR = "selector";

    private final Yaml yaml = new Yaml();

    @Override
    public ApplicationConfiguration load() throws ConfigFileNotFoundException {
        // 初始化ApplicationConfiguration对象
        ApplicationConfiguration configuration = new ApplicationConfiguration();
        this.loadConfig(configuration);
        // 读取系统属性对值进行覆盖处理, 使用的是
        this.overrideConfigBySystemEnv(configuration);
        return configuration;
    }

    @SuppressWarnings("unchecked")
    private void loadConfig(ApplicationConfiguration configuration) throws ConfigFileNotFoundException {
        try {
            // 读取application.yml
            Reader applicationReader = ResourceUtils.read("application.yml");
            // 将 application.yml 转换为 ModuleName, Map<String, Object> 的结构
            Map<String, Map<String, Object>> moduleConfig = yaml.loadAs(applicationReader, Map.class);

            if (CollectionUtils.isNotEmpty(moduleConfig)) {
                // 读取每个ModuleName下的 Map, 先读取到selector属性, 如果不存在就跳过此端配置
                selectConfig(moduleConfig);

                moduleConfig.forEach((moduleName, providerConfig) -> {
                    if (providerConfig.size() > 0) {
                        log.info("Get a module define from application.yml, module name: {}", moduleName);
                        ApplicationConfiguration.ModuleConfiguration moduleConfiguration = configuration.addModule(
                            moduleName);
                        providerConfig.forEach((providerName, config) -> {
                            log.info(
                                "Get a provider define belong to {} module, provider name: {}", moduleName,
                                providerName
                            );
                            final Map<String, ?> propertiesConfig = (Map<String, ?>) config;
                            final Properties properties = new Properties();
                            if (propertiesConfig != null) {
                                propertiesConfig.forEach((propertyName, propertyValue) -> {
                                    if (propertyValue instanceof Map) {
                                        Properties subProperties = new Properties();
                                        ((Map) propertyValue).forEach((key, value) -> {
                                            subProperties.put(key, value);
                                            replacePropertyAndLog(key, value, subProperties, providerName);
                                        });
                                        properties.put(propertyName, subProperties);
                                    } else {
                                        properties.put(propertyName, propertyValue);
                                        replacePropertyAndLog(propertyName, propertyValue, properties, providerName);
                                    }
                                });
                            }
                            moduleConfiguration.addProviderConfiguration(providerName, properties);
                        });
                    } else {
                        log.warn(
                            "Get a module define from application.yml, but no provider define, use default, module name: {}",
                            moduleName
                        );
                    }
                });
            }
        } catch (FileNotFoundException e) {
            throw new ConfigFileNotFoundException(e.getMessage(), e);
        }
    }
    // ... 

// OAP应用配置类
public class ApplicationConfiguration {
    // 模块定义配置map
    private HashMap<String, ModuleConfiguration> modules = new HashMap<>();
    // 模块配置名列表
    public String[] moduleList() {
        return modules.keySet().toArray(new String[0]);
    }
    // 添加模块定义配置
    public ModuleConfiguration addModule(String moduleName) {
        ModuleConfiguration newModule = new ModuleConfiguration();
        modules.put(moduleName, newModule);
        return newModule;
    }
    // 判断指定模块名是否存在模块定义配置map中
    public boolean has(String moduleName) {
        return modules.containsKey(moduleName);
    }
    // 获取模块定义配置
    public ModuleConfiguration getModuleConfiguration(String name) {
        return modules.get(name);
    }

    /**
     * The configurations about a certain module.
     */
    // 模块定义配置类
    public static class ModuleConfiguration {

        // 模块提供对象map
        private HashMap<String, ProviderConfiguration> providers = new HashMap<>();

        private ModuleConfiguration() {
        }
        // 获取模块提供配置
        public Properties getProviderConfiguration(String name) {
            return providers.get(name).getProperties();
        }
        // 是否存在模块提供配置
        public boolean has(String name) {
            return providers.containsKey(name);
        }
        // 添加模块提供配置
        public ModuleConfiguration addProviderConfiguration(String name, Properties properties) {
            ProviderConfiguration newProvider = new ProviderConfiguration(properties);
            providers.put(name, newProvider);
            return this;
        }
    }

    /**
     * The configuration about a certain provider of a module.
     */
    // 模块提供配置类
    public static class ProviderConfiguration {
        // 模块提供属性
        private Properties properties;

        ProviderConfiguration(Properties properties) {
            this.properties = properties;
        }

        private Properties getProperties() {
            return properties;
        }
    }
}

// 模块管理类，管理模块的生命周期
// Module 是 Skywalking 在 OAP 提供的一种管理功能特性的机制。通过 Module 机制，可以方便的定义模块，并且可以提供多种实现，在配置文件中任意选择实现。
public class ModuleManager implements ModuleDefineHolder {
    // 所有模块是否已经通过准备阶段
    private boolean isInPrepareStage = true;
    // 所有被加载的模块定义对象map
    private final Map<String, ModuleDefine> loadedModules = new HashMap<>();

    /**
     * Init the given modules
     */
    // 初始化所有配置的模块
    public void init(
        ApplicationConfiguration applicationConfiguration) throws ModuleNotFoundException, ProviderNotFoundException, ServiceNotProvidedException, CycleDependencyException, ModuleConfigException, ModuleStartException {
        // 获取配置类中的模块名
        String[] moduleNames = applicationConfiguration.moduleList();
        // SPI加载所有模块定义对象
        ServiceLoader<ModuleDefine> moduleServiceLoader = ServiceLoader.load(ModuleDefine.class);
        // SPI加载所有模块提供对象
        ServiceLoader<ModuleProvider> moduleProviderLoader = ServiceLoader.load(ModuleProvider.class);
        // 所有配置类中定义的模块，进行准备阶段
        HashSet<String> moduleSet = new HashSet<>(Arrays.asList(moduleNames));
        for (ModuleDefine module : moduleServiceLoader) {
            // 判断 ModuleDefine.name 是否在 ApplicationConfiguration 配置中存在, 不存在跳过
            if (moduleSet.contains(module.name())) {
                // 存在时, 分别调用每个 ModuleDefine#prepare 进行初始化, 此处主要工作为配置初始化, 完成后放入 loadedModules 中
                module.prepare(this, applicationConfiguration.getModuleConfiguration(module.name()), moduleProviderLoader);
                loadedModules.put(module.name(), module);
                moduleSet.remove(module.name());
            }
        }
        // Finish prepare stage
        // 准备阶段结束
        isInPrepareStage = false;

        if (moduleSet.size() > 0) {
            throw new ModuleNotFoundException(moduleSet.toString() + " missing.");
        }
        // 使用loadedModules 对 BootstrapFlow 进行初始化, BootstrapFlow 会对模块的依赖关系进行排序
        BootstrapFlow bootstrapFlow = new BootstrapFlow(loadedModules);
        // 调用每个模块的启动阶段
        bootstrapFlow.start(this);
        // 所有模块进入完成后通知阶段 所有模块都已经加载完成
        bootstrapFlow.notifyAfterCompleted();
    }

    // 判断是否有该模块
    @Override
    public boolean has(String moduleName) {
        return loadedModules.get(moduleName) != null;
    }

    // 通过模块名获取模块定义对象
    @Override
    public ModuleProviderHolder find(String moduleName) throws ModuleNotFoundRuntimeException {
        assertPreparedStage();
        ModuleDefine module = loadedModules.get(moduleName);
        if (module != null)
            return module;
        throw new ModuleNotFoundRuntimeException(moduleName + " missing.");
    }
    // 断言是否还在准备阶段，如果还在准备阶段，则抛出异常
    private void assertPreparedStage() {
        if (isInPrepareStage) {
            throw new AssertionError("Still in preparing stage.");
        }
    }
}

    private String configPath = "meter-analyzer-config";
    private String topicNameOfMetrics = "skywalking-metrics";
    private String topicNameOfProfiling = "skywalking-profilings";
    private String topicNameOfTracingSegments = "skywalking-segments";
    private String topicNameOfManagements = "skywalking-managements";
    private String topicNameOfMeters = "skywalking-meters";
    private String topicNameOfLogs = "skywalking-logs";
    private boolean createTopicIfNotExist = true;

    private boolean createTopicIfNotExist = true; 
    private int partitions = 3;
    private int replicationFactor = 2;

        if (!missedTopics.isEmpty()) {
            log.info("Topics" + missedTopics.toString() + " not exist.");
            List<NewTopic> newTopicList = missedTopics.stream()
                                                      .map(topic -> new NewTopic(
                                                          topic,
                                                          config.getPartitions(),
                                                          (short) config.getReplicationFactor()
                                                      )).collect(Collectors.toList());

            try {
                adminClient.createTopics(newTopicList).all().get();
            } catch (Exception e) {
                throw new ModuleStartException("Failed to create Kafka Topics" + missedTopics + ".", e);
            }
        }

       executor = new ThreadPoolExecutor(threadPoolSize, threadPoolSize,
                                          60, TimeUnit.SECONDS,
                                          new ArrayBlockingQueue(threadPoolQueueSize),
                                          new CustomThreadFactory("KafkaConsumer"),
                                          new ThreadPoolExecutor.CallerRunsPolicy()
        );

@Override
public void start() throws ServiceNotProvidedException, ModuleStartException {
    handlerRegister.register(new JVMMetricsHandler(getManager(), config));
    handlerRegister.register(new ServiceManagementHandler(getManager(), config));
    handlerRegister.register(new TraceSegmentHandler(getManager(), config));
    handlerRegister.register(new ProfileTaskHandler(getManager(), config));
    handlerRegister.register(new MeterServiceHandler(getManager(), config));
 
    if (config.isEnableNativeProtoLog()) {
        handlerRegister.register(new LogHandler(getManager(), config));
    }
    if (config.isEnableNativeJsonLog()) {
        handlerRegister.register(new JsonLogHandler(getManager(), config));
    }
 
    handlerRegister.start();
}

@Override
public void run() {
    while (true) {
        try {
            ConsumerRecords<String, Bytes> consumerRecords = consumer.poll(Duration.ofMillis(500L));
            if (!consumerRecords.isEmpty()) {
                for (final ConsumerRecord<String, Bytes> record : consumerRecords) {
                    executor.submit(() -> handlerMap.get(record.topic()).handle(record));
                }
                if (!enableKafkaMessageAutoCommit) {
                    consumer.commitAsync();
                }
            }
        } catch (Exception e) {
            log.error("Kafka handle message error.", e);
        }
    }
}