code: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   *
   */
  
  package org.apache.skywalking.oap.server.configuration.nacos;
  
  import com.alibaba.nacos.api.NacosFactory;
  import com.alibaba.nacos.api.PropertyKeyConst;
  import com.alibaba.nacos.api.config.ConfigService;
  import com.alibaba.nacos.api.config.listener.Listener;
  import com.alibaba.nacos.api.exception.NacosException;
  import java.util.Arrays;
  import java.util.HashSet;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Properties;
  import java.util.Set;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.Executor;
  
  import lombok.extern.slf4j.Slf4j;
  import org.apache.skywalking.oap.server.library.util.StringUtil;
  import org.apache.skywalking.oap.server.configuration.api.ConfigTable;
  import org.apache.skywalking.oap.server.configuration.api.ConfigWatcherRegister;
  import org.apache.skywalking.oap.server.configuration.api.GroupConfigTable;
  
  @Slf4j
  public class NacosConfigWatcherRegister extends ConfigWatcherRegister {
      private final NacosServerSettings settings;
      private final ConfigService configService;
      private final Map<String, Optional<String>> configItemKeyedByName;
      private final Map<String, Listener> listenersByKey;
  
      public NacosConfigWatcherRegister(NacosServerSettings settings) throws NacosException {
          super(settings.getPeriod());
  
          this.settings = settings;
          this.configItemKeyedByName = new ConcurrentHashMap<>();
          this.listenersByKey = new ConcurrentHashMap<>();
  
          final int port = this.settings.getPort();
          final String serverAddr = this.settings.getServerAddr();
  
          final Properties properties = new Properties();
          properties.put(PropertyKeyConst.SERVER_ADDR, serverAddr + ":" + port);
          properties.put(PropertyKeyConst.NAMESPACE, settings.getNamespace());
          if (StringUtil.isNotEmpty(settings.getUsername())) {
              properties.put(PropertyKeyConst.USERNAME, settings.getUsername());
              properties.put(PropertyKeyConst.PASSWORD, settings.getPassword());
          } else if (StringUtil.isNotEmpty(settings.getAccessKey())) {
              properties.put(PropertyKeyConst.ACCESS_KEY, settings.getAccessKey());
              properties.put(PropertyKeyConst.SECRET_KEY, settings.getSecretKey());
          }
          this.configService = NacosFactory.createConfigService(properties);
      }
  
      @Override
      public Optional<ConfigTable> readConfig(Set<String> keys) {
          removeUninterestedKeys(keys);
          registerKeyListeners(keys);
  
          final ConfigTable table = new ConfigTable();
  
          for (Map.Entry<String, Optional<String>> entry : configItemKeyedByName.entrySet()) {
              final String key = entry.getKey();
              final Optional<String> value = entry.getValue();
  
              if (value.isPresent()) {
                  table.add(new ConfigTable.ConfigItem(key, value.get()));
              } else {
                  table.add(new ConfigTable.ConfigItem(key, null));
              }
          }
  
          return Optional.of(table);
      }
  
      @Override
      public Optional<GroupConfigTable> readGroupConfig(final Set<String> keys) {
          GroupConfigTable groupConfigTable = new GroupConfigTable();
          keys.forEach(key -> {
              GroupConfigTable.GroupConfigItems groupConfigItems = new GroupConfigTable.GroupConfigItems(key);
              groupConfigTable.addGroupConfigItems(groupConfigItems);
              String config = null;
              try {
                  config = configService.getConfig(key, settings.getGroup(), 1000);
                  if (StringUtil.isNotEmpty(config)) {
                      String[] itemNames = config.split("\\n|\\r\\n");
                      Arrays.stream(itemNames).map(String::trim).forEach(itemName -> {
                          String itemValue = null;
                          try {
                              itemValue = configService.getConfig(itemName, settings.getGroup(), 1000);
                          } catch (NacosException e) {
                              log.error("Failed to register Nacos listener for dataId: {}", itemName, e);
                          }
                          groupConfigItems.add(
                              new ConfigTable.ConfigItem(itemName, itemValue));
                      });
                  }
              } catch (NacosException e) {
                  log.error("Failed to register Nacos listener for dataId: {}", key, e);
              }
          });
  
          return Optional.of(groupConfigTable);
      }
  
      private void registerKeyListeners(final Set<String> keys) {
          final String group = settings.getGroup();
  
          for (final String dataId : keys) {
              if (listenersByKey.containsKey(dataId)) {
                  continue;
              }
              try {
                  listenersByKey.putIfAbsent(dataId, new Listener() {
                      @Override
                      public Executor getExecutor() {
                          return null;
                      }
  
                      @Override
                      public void receiveConfigInfo(String configInfo) {
                          onDataIdValueChanged(dataId, configInfo);
                      }
                  });
                  configService.addListener(dataId, group, listenersByKey.get(dataId));
  
                  // the key is newly added, read the config for the first time
                  final String config = configService.getConfig(dataId, group, 1000);
                  onDataIdValueChanged(dataId, config);
              } catch (NacosException e) {
                  log.warn("Failed to register Nacos listener for dataId: {}", dataId);
              }
          }
      }
  
      private void removeUninterestedKeys(final Set<String> interestedKeys) {
          final String group = settings.getGroup();
  
          final Set<String> uninterestedKeys = new HashSet<>(listenersByKey.keySet());
          uninterestedKeys.removeAll(interestedKeys);
  
          uninterestedKeys.forEach(k -> {
              final Listener listener = listenersByKey.remove(k);
              if (listener != null) {
                  configService.removeListener(k, group, listener);
              }
          });
      }
  
      void onDataIdValueChanged(String dataId, String configInfo) {
          if (log.isInfoEnabled()) {
              log.info("Nacos config changed: {}: {}", dataId, configInfo);
          }
  
          configItemKeyedByName.put(dataId, Optional.ofNullable(configInfo));
      }
  }
  