refactor: refactor rpc and config settings

Author: kevinten10

capa-spi-aws-config/pom.xml

@@ -39,17 +39,18 @@
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>appconfig</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+
         <!-- unit test -->
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-engine</artifactId>
             <scope>test</scope>
         </dependency>
-
-        <dependency>
-            <groupId>com.google.guava</groupId>
-            <artifactId>guava</artifactId>
-        </dependency>
     </dependencies>
 
     <build>

…spi/aws/config/AwsCapaConfiguration.java …a/spi/aws/config/AwsCapaConfigStore.javacapa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfiguration.java renamed to capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfigStore.java capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfiguration.java renamed to capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfigStore.java

@@ -21,9 +21,9 @@
 import group.rxcloud.capa.component.configstore.StoreConfig;
 import group.rxcloud.capa.component.configstore.SubscribeResp;
 import group.rxcloud.capa.infrastructure.serializer.CapaObjectSerializer;
-import group.rxcloud.capa.spi.aws.config.common.serializer.SerializerProcessor;
 import group.rxcloud.capa.spi.aws.config.entity.Configuration;
 import group.rxcloud.capa.spi.aws.config.scheduler.AwsCapaConfigurationScheduler;
+import group.rxcloud.capa.spi.aws.config.serializer.SerializerProcessor;
 import group.rxcloud.capa.spi.configstore.CapaConfigStoreSpi;
 import group.rxcloud.cloudruntimes.utils.TypeRef;
 import org.slf4j.Logger;
@@ -46,19 +46,13 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
-import static group.rxcloud.capa.spi.aws.config.common.constant.CapaAWSConstants.AWS_APP_CONFIG_NAME;
-import static group.rxcloud.capa.spi.aws.config.common.constant.CapaAWSConstants.DEFAULT_ENV;
 
 /**
  * @author Reckless Xu
  */
-public class AwsCapaConfiguration extends CapaConfigStoreSpi {
+public class AwsCapaConfigStore extends CapaConfigStoreSpi {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(AwsCapaConfiguration.class);
-
-    private AppConfigAsyncClient appConfigAsyncClient;
-
-    private SerializerProcessor serializerProcessor;
+    private static final Logger LOGGER = LoggerFactory.getLogger(AwsCapaConfigStore.class);
 
     /**
      * key of versionMap--applicationName,format:appid_ENV,e.g:"12345_FAT"
@@ -69,17 +63,31 @@ public class AwsCapaConfiguration extends CapaConfigStoreSpi {
      * <p>
      * ps:currentHashMap may not be necessary, as update in synchronized method
      */
-    private static Map<String, ConcurrentHashMap<String, Configuration<?>>> versionMap;
+    private static final Map<String, ConcurrentHashMap<String, Configuration<?>>> versionMap;
+
+    static {
+        versionMap = new ConcurrentHashMap<>();
+    }
+
+    private final CapaObjectSerializer objectSerializer;
+
+    private SerializerProcessor serializerProcessor;
+
+    private AppConfigAsyncClient appConfigAsyncClient;
 
     /**
      * Instantiates a new Capa configuration.
      *
      * @param objectSerializer Serializer for transient request/response objects.
      */
-    public AwsCapaConfiguration(CapaObjectSerializer objectSerializer) {
+    public AwsCapaConfigStore(CapaObjectSerializer objectSerializer) {
         super(objectSerializer);
+        this.objectSerializer = objectSerializer;
+    }
+
+    @Override
+    protected void doInit(StoreConfig storeConfig) {
         appConfigAsyncClient = AppConfigAsyncClient.create();
-        versionMap = new ConcurrentHashMap<>();
         serializerProcessor = new SerializerProcessor(objectSerializer);
     }
 
@@ -91,12 +99,7 @@ public String stopSubscribe() {
 
     @Override
     public void close() {
-        //no need
-    }
-
-    @Override
-    protected void doInit(StoreConfig storeConfig) {
-        //no need
+        // no need
     }
 
     @Override
@@ -105,7 +108,7 @@ protected <T> Mono<List<ConfigurationItem<T>>> doGet(String appId, String group,
         if (CollectionUtils.isNullOrEmpty(keys)) {
             return Mono.error(new IllegalArgumentException("keys is null or empty"));
         }
-        //todo:need to get the specific env from system properties
+        // todo:need to get the specific env from system properties
         String applicationName = appId + "_FAT";
         String configurationName = keys.get(0);
         String clientConfigurationVersion = getCurVersion(applicationName, configurationName);
@@ -115,17 +118,17 @@ protected <T> Mono<List<ConfigurationItem<T>>> doGet(String appId, String group,
                 .clientId(UUID.randomUUID().toString())
                 .configuration(configurationName)
                 .clientConfigurationVersion(clientConfigurationVersion)
-                .environment(DEFAULT_ENV)
+                .environment(AwsCapaConfigurationProperties.AppConfigProperties.Settings.getAwsAppConfigEnv())
                 .build();
 
         return Mono.fromFuture(() -> appConfigAsyncClient.getConfiguration(request))
                 .publishOn(AwsCapaConfigurationScheduler.INSTANCE.configPublisherScheduler)
                 .map(resp -> {
-                    //if version doesn't change, get from versionMap
+                    // if version doesn't change, get from versionMap
                     if (Objects.equals(clientConfigurationVersion, resp.configurationVersion())) {
                         items.add((ConfigurationItem<T>) getCurConfigurationItem(applicationName, configurationName));
                     } else {
-                        //if version changes,update versionMap and return
+                        // if version changes,update versionMap and return
                         Configuration<T> tConfiguration = updateConfigurationItem(applicationName, configurationName, type, resp.content(), resp.configurationVersion());
                         if (tConfiguration != null) {
                             items.add(tConfiguration.getConfigurationItem());
@@ -137,7 +140,7 @@ protected <T> Mono<List<ConfigurationItem<T>>> doGet(String appId, String group,
 
     @Override
     protected <T> Flux<SubscribeResp<T>> doSubscribe(String appId, String group, String label, List<String> keys, Map<String, String> metadata, TypeRef<T> type) {
-        //todo:need to get the specific env from system properties
+        // todo: need to get the specific env from system properties
         String applicationName = appId + "_FAT";
         String configurationName = keys.get(0);
 
@@ -146,7 +149,7 @@ protected <T> Flux<SubscribeResp<T>> doSubscribe(String appId, String group, Str
     }
 
     private synchronized <T> Mono<Boolean> initConfig(String applicationName, String configurationName, String group, String label, Map<String, String> metadata, TypeRef<T> type) {
-        //double check whether has been initialized
+        // double check whether has been initialized
         if (isInitialized(applicationName, configurationName)) {
             return Mono.just(true);
         }
@@ -160,7 +163,7 @@ private synchronized <T> Mono<Boolean> initConfig(String applicationName, String
                                 .clientId(UUID.randomUUID().toString())
                                 .configuration(configurationName)
                                 .clientConfigurationVersion(version)
-                                .environment(DEFAULT_ENV)
+                                .environment(AwsCapaConfigurationProperties.AppConfigProperties.Settings.getAwsAppConfigEnv())
                                 .build();
 
                         GetConfigurationResponse resp = null;
@@ -191,39 +194,40 @@ private synchronized <T> void createSubscribe(String applicationName, String con
             return;
         }
         Flux.create(fluxSink -> {
-            AwsCapaConfigurationScheduler.INSTANCE.configSubscribePollingScheduler
-                    .schedulePeriodically(() -> {
-                        String version = getCurVersion(applicationName, configurationName);
-
-                        GetConfigurationRequest request = GetConfigurationRequest.builder()
-                                .application(applicationName)
-                                .clientId(UUID.randomUUID().toString())
-                                .configuration(configurationName)
-                                .clientConfigurationVersion(version)
-                                .environment(DEFAULT_ENV)
-                                .build();
-
-                        GetConfigurationResponse resp = null;
-                        try {
-                            resp = appConfigAsyncClient.getConfiguration(request).get();
-                        } catch (InterruptedException | ExecutionException e) {
-                            LOGGER.error("error occurs when getConfiguration,configurationName:{},version:{}", request.configuration(), request.clientConfigurationVersion(), e);
-                        }
-                        //update subscribed status if needs
-                        getConfiguration(applicationName, configurationName).getSubscribed().compareAndSet(false, true);
-
-                        if (resp != null && !Objects.equals(resp.configurationVersion(), version)) {
-                            fluxSink.next(resp);
-                        }
-                        //todo:make the polling frequency configurable
-                    }, 0, 1, TimeUnit.SECONDS);
-        })
+                    AwsCapaConfigurationScheduler.INSTANCE.configSubscribePollingScheduler
+                            .schedulePeriodically(() -> {
+                                String version = getCurVersion(applicationName, configurationName);
+
+                                GetConfigurationRequest request = GetConfigurationRequest.builder()
+                                        .application(applicationName)
+                                        .clientId(UUID.randomUUID().toString())
+                                        .configuration(configurationName)
+                                        .clientConfigurationVersion(version)
+                                        .environment(AwsCapaConfigurationProperties.AppConfigProperties.Settings.getAwsAppConfigEnv())
+                                        .build();
+
+                                GetConfigurationResponse resp = null;
+                                try {
+                                    resp = appConfigAsyncClient.getConfiguration(request).get();
+                                } catch (InterruptedException | ExecutionException e) {
+                                    LOGGER.error("error occurs when getConfiguration,configurationName:{},version:{}", request.configuration(), request.clientConfigurationVersion(), e);
+                                }
+                                // update subscribed status if needs
+                                getConfiguration(applicationName, configurationName).getSubscribed().compareAndSet(false, true);
+
+                                if (resp != null && !Objects.equals(resp.configurationVersion(), version)) {
+                                    fluxSink.next(resp);
+                                }
+                                // todo: make the polling frequency configurable
+                            }, 0, 1, TimeUnit.SECONDS);
+                })
                 .publishOn(AwsCapaConfigurationScheduler.INSTANCE.configPublisherScheduler)
                 .map(origin -> {
                     GetConfigurationResponse resp = (GetConfigurationResponse) origin;
                     Configuration configuration = updateConfigurationItem(applicationName, configurationName, type, resp.content(), resp.configurationVersion());
                     return configuration == null ? Configuration.EMPTY : configuration;
-                }).filter(resp -> resp != Configuration.EMPTY)
+                })
+                .filter(resp -> resp != Configuration.EMPTY)
                 .subscribe(resp -> {
                     resp.triggers(resp.getConfigurationItem());
                 });
@@ -232,10 +236,10 @@ private synchronized <T> void createSubscribe(String applicationName, String con
     private <T> Flux<SubscribeResp<T>> doSub(String applicationName, String configurationName, String group, String label, Map<String, String> metadata, TypeRef<T> type, String appId) {
         Configuration<?> configuration = getConfiguration(applicationName, configurationName);
         return Flux.create(fluxSink -> {
-            configuration.addListener(configurationItem -> {
-                fluxSink.next(configurationItem);
-            });
-        })
+                    configuration.addListener(configurationItem -> {
+                        fluxSink.next(configurationItem);
+                    });
+                })
                 .map(resp -> (ConfigurationItem<T>) resp)
                 .map(resp -> convert(resp, appId));
     }
@@ -244,7 +248,7 @@ private <T> SubscribeResp<T> convert(ConfigurationItem<T> conf, String appId) {
         SubscribeResp<T> subscribeResp = new SubscribeResp<>();
         subscribeResp.setItems(Lists.newArrayList(conf));
         subscribeResp.setAppId(appId);
-        subscribeResp.setStoreName(AWS_APP_CONFIG_NAME);
+        subscribeResp.setStoreName(AwsCapaConfigurationProperties.AppConfigProperties.Settings.getAwsAppConfigName());
         return subscribeResp;
     }
 
@@ -286,15 +290,15 @@ private ConfigurationItem<?> getCurConfigurationItem(String applicationName, Str
     private <T> Configuration<T> updateConfigurationItem(String applicationName, String configurationName, TypeRef<T> type, SdkBytes contentSdkBytes, String version) {
         ConcurrentHashMap<String, Configuration<?>> configMap = versionMap.get(applicationName);
 
-        //in fact,configMap.get(configurationName) is always not null, as it has been initialized in initialization process
+        // in fact,configMap.get(configurationName) is always not null, as it has been initialized in initialization process
         Configuration<T> configuration = (Configuration<T>) configMap.get(configurationName);
 
         synchronized (configuration.lock) {
-            //check whether content has been updated by other thread
+            // check whether content has been updated by other thread
             if (configMap.containsKey(configurationName) && Objects.equals(configMap.get(configurationName).getClientConfigurationVersion(), version)) {
                 return null;
             }
-            //do need to update
+            // do need to update
             T content = serializerProcessor.deserialize(contentSdkBytes, type, configurationName);
             configuration.setClientConfigurationVersion(version);
 
@@ -350,5 +354,4 @@ private Configuration<?> getConfiguration(String applicationName, String configu
         }
         return Configuration.EMPTY;
     }
-
 }

capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfigurationProperties.java

@@ -0,0 +1,59 @@
+/*
+ * 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 group.rxcloud.capa.spi.aws.config;
+
+import group.rxcloud.capa.infrastructure.config.CapaProperties;
+
+import java.util.Properties;
+
+/**
+ * @author Reckless Xu
+ */
+public interface AwsCapaConfigurationProperties {
+
+    interface AppConfigProperties {
+
+        abstract class Settings {
+
+            private static String awsAppConfigName = "AWS AppConfig";
+            private static String awsAppConfigEnv = "ENV";
+
+            private static final String RPC_AWS_APP_CONFIG_NAME = "RPC_AWS_APP_CONFIG_NAME";
+            private static final String RPC_AWS_APP_CONFIG_ENV = "RPC_AWS_APP_CONFIG_ENV";
+
+            static {
+                Properties properties = CapaProperties.COMPONENT_PROPERTIES_SUPPLIER.apply("configuration-aws");
+
+                awsAppConfigName = properties.getProperty(RPC_AWS_APP_CONFIG_NAME, awsAppConfigName);
+
+                awsAppConfigEnv = properties.getProperty(RPC_AWS_APP_CONFIG_ENV, awsAppConfigEnv);
+            }
+
+            public static String getAwsAppConfigName() {
+                return awsAppConfigName;
+            }
+
+            public static String getAwsAppConfigEnv() {
+                return awsAppConfigEnv;
+            }
+
+            private Settings() {
+
+            }
+        }
+    }
+}

capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/common/constant/CapaAWSConstants.java

@@ -27,6 +27,7 @@
  * @author Reckless Xu
  */
 public class Configuration<T> {
+
     private static final Logger LOGGER = LoggerFactory.getLogger(Configuration.class);
 
     public static final Configuration<Void> EMPTY = new Configuration<>();
@@ -53,7 +54,9 @@ public synchronized void addListener(ConfigurationListener<T> listener) {
     public boolean triggers(ConfigurationItem<T> data) {
         boolean result = true;
         for (ConfigurationListener<T> listener : listeners) {
-            if (!trigger(listener, data)) result = false;
+            if (!trigger(listener, data)) {
+                result = false;
+            }
         }
         return result;
     }
@@ -68,7 +71,6 @@ private boolean trigger(ConfigurationListener<T> listener, ConfigurationItem<T>
         }
     }
 
-
     public String getClientConfigurationVersion() {
         return clientConfigurationVersion;
     }