Keepalive OpensergoClient.

force-pushed after main-branch was force-pushed by Eric Zhao.

Co-authored-by: Eric Zhao <[email protected]>

Signed-off-by: Jiangnan Jia <[email protected]>

Author: jnan806

src/main/java/io/opensergo/ConfigKind.java

@@ -26,8 +26,7 @@ public enum ConfigKind {
     FAULT_TOLERANCE_RULE("fault-tolerance.opensergo.io/v1alpha1/FaultToleranceRule", "FaultToleranceRule"),
     RATE_LIMIT_STRATEGY("fault-tolerance.opensergo.io/v1alpha1/RateLimitStrategy", "RateLimitStrategy"),
     THROTTLING_STRATEGY("fault-tolerance.opensergo.io/v1alpha1/ThrottlingStrategy", "ThrottlingStrategy"),
-    CONCURRENCY_LIMIT_STRATEGY("fault-tolerance.opensergo.io/v1alpha1/ConcurrencyLimitStrategy",
-        "ConcurrencyLimitStrategy"),
+    CONCURRENCY_LIMIT_STRATEGY("fault-tolerance.opensergo.io/v1alpha1/ConcurrencyLimitStrategy", "ConcurrencyLimitStrategy"),
     CIRCUIT_BREAKER_STRATEGY("fault-tolerance.opensergo.io/v1alpha1/CircuitBreakerStrategy", "CircuitBreakerStrategy");
 
     private final String kindName;

src/main/java/io/opensergo/OpenSergoClient.java

@@ -30,6 +30,8 @@
 import io.opensergo.util.AssertUtils;
 import io.opensergo.util.IdentifierUtils;
 
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -46,6 +48,7 @@ public class OpenSergoClient implements AutoCloseable {
     private final SubscribeRegistry subscribeRegistry;
 
     private AtomicInteger reqId;
+    protected volatile OpensergoClientStatus status;
 
     public OpenSergoClient(String host, int port) {
         this.channel = ManagedChannelBuilder.forAddress(host, port)
@@ -56,17 +59,68 @@ public OpenSergoClient(String host, int port) {
         this.configCache = new SubscribedConfigCache();
         this.subscribeRegistry = new SubscribeRegistry();
         this.reqId = new AtomicInteger(0);
+        status = OpensergoClientStatus.INITIAL;
+    }
+
+    public void registerSubscribeInfo(OpensergoClientSubscribeInfo subscribeInfo) {
+        // Register subscriber to local.
+        if (Optional.of(subscribeInfo.getSubscriberList()).isPresent() && subscribeInfo.getSubscriberList().size() > 0) {
+            subscribeInfo.getSubscriberList().forEach(subscriber -> {
+                this.subscribeRegistry.registerSubscriber(subscribeInfo.getSubscribeKey(), subscriber);
+                OpenSergoLogger.info("OpenSergo subscribeinfo registered, subscribeKey={}, subscriber={}", subscribeInfo.getSubscribeKey(), subscriber);
+
+                if (requestAndResponseWriter != null && this.status == OpensergoClientStatus.STARTED) {
+                    this.subscribeConfig(subscribeInfo.getSubscribeKey());
+                }
+            });
+        }
     }
 
     public void start() throws Exception {
+        OpenSergoLogger.info("OpensergoClient is starting...");
+
+        if (status == OpensergoClientStatus.INITIAL) {
+            OpenSergoLogger.info("open keepavlive thread");
+            new Thread(this::keepAlive).start();
+        }
+
+        status = OpensergoClientStatus.STARTING;
+
         this.requestAndResponseWriter = transportGrpcStub.withWaitForReady()
-            .subscribeConfig(new OpenSergoSubscribeClientObserver(configCache, subscribeRegistry));
+            .subscribeConfig(new OpenSergoSubscribeClientObserver(this));
+
+        OpenSergoLogger.info("begin to subscribe config-data...");
+        this.subscribeRegistry.getSubscriberKeysAll().forEach(subscribeKey -> {
+            this.subscribeConfig(subscribeKey);
+        });
+
+        OpenSergoLogger.info("openSergoClient is started");
+        status = OpensergoClientStatus.STARTED;
+    }
+
+    private void keepAlive() {
+        try {
+            if (status != OpensergoClientStatus.STARTING
+                    && status != OpensergoClientStatus.STARTED
+                    && status != OpensergoClientStatus.SHUTDOWN) {
+                OpenSergoLogger.info("try to restart openSergoClient...");
+                this.start();
+            }
+            Thread.sleep(TimeUnit.SECONDS.toMillis(10));
+            if( status != OpensergoClientStatus.SHUTDOWN) {
+                keepAlive();
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
     }
 
     @Override
     public void close() throws Exception {
         requestAndResponseWriter.onCompleted();
 
+        status = OpensergoClientStatus.SHUTDOWN;
+
         // gracefully drain the requests, then close the connection
         channel.shutdown();
     }
@@ -77,8 +131,8 @@ public boolean unsubscribeConfig(SubscribeKey subscribeKey) {
         AssertUtils.assertNotNull(subscribeKey.getKind(), "kind cannot be null");
 
         if (requestAndResponseWriter == null) {
-            // TODO: return status that indicates not ready
-            throw new IllegalStateException("gRPC stream is not ready");
+            OpenSergoLogger.error("Fatal error occurred on OpenSergo gRPC ClientObserver", new IllegalStateException("gRPC stream is not ready"));
+            status = OpensergoClientStatus.INTERRUPTED;
         }
         SubscribeRequestTarget subTarget = SubscribeRequestTarget.newBuilder()
             .setNamespace(subscribeKey.getNamespace()).setApp(subscribeKey.getApp())
@@ -106,8 +160,8 @@ public boolean subscribeConfig(SubscribeKey subscribeKey, OpenSergoConfigSubscri
         AssertUtils.assertNotNull(subscribeKey.getKind(), "kind cannot be null");
 
         if (requestAndResponseWriter == null) {
-            // TODO: return status that indicates not ready
-            throw new IllegalStateException("gRPC stream is not ready");
+            OpenSergoLogger.error("Fatal error occurred on OpenSergo gRPC ClientObserver", new IllegalStateException("gRPC stream is not ready"));
+            status = OpensergoClientStatus.INTERRUPTED;
         }
         SubscribeRequestTarget subTarget = SubscribeRequestTarget.newBuilder()
             .setNamespace(subscribeKey.getNamespace()).setApp(subscribeKey.getApp())
@@ -121,18 +175,15 @@ public boolean subscribeConfig(SubscribeKey subscribeKey, OpenSergoConfigSubscri
         // Send SubscribeRequest
         requestAndResponseWriter.onNext(request);
 
-        // Register subscriber to local.
-        if (subscriber != null) {
-            subscribeRegistry.registerSubscriber(subscribeKey, subscriber);
-            OpenSergoLogger.info("OpenSergo config subscriber registered, subscribeKey={}, subscriber={}",
-                subscribeKey, subscriber);
-        }
-
         return true;
     }
 
     public SubscribedConfigCache getConfigCache() {
         return configCache;
     }
 
+    public SubscribeRegistry getSubscribeRegistry() {
+        return subscribeRegistry;
+    }
+
 }

src/main/java/io/opensergo/OpenSergoSubscribeClientObserver.java

@@ -30,8 +30,6 @@
 import io.opensergo.proto.transport.v1.SubscribeResponse;
 import io.opensergo.subscribe.OpenSergoConfigSubscriber;
 import io.opensergo.subscribe.SubscribeKey;
-import io.opensergo.subscribe.SubscribeRegistry;
-import io.opensergo.subscribe.SubscribedConfigCache;
 import io.opensergo.util.StringUtils;
 
 /**
@@ -41,13 +39,10 @@ public class OpenSergoSubscribeClientObserver implements ClientResponseObserver<
 
     private ClientCallStreamObserver<SubscribeRequest> requestStream;
 
-    private final SubscribedConfigCache configCache;
-    private final SubscribeRegistry subscribeRegistry;
+    private OpenSergoClient openSergoClient;
 
-    public OpenSergoSubscribeClientObserver(SubscribedConfigCache configCache,
-                                            SubscribeRegistry subscribeRegistry) {
-        this.configCache = configCache;
-        this.subscribeRegistry = subscribeRegistry;
+    public OpenSergoSubscribeClientObserver(OpenSergoClient openSergoClient) {
+        this.openSergoClient = openSergoClient;
     }
 
     @Override
@@ -58,7 +53,7 @@ public void beforeStart(ClientCallStreamObserver<SubscribeRequest> requestStream
     private LocalDataNotifyResult notifyDataChange(SubscribeKey subscribeKey, DataWithVersion dataWithVersion)
         throws Exception {
         long receivedVersion = dataWithVersion.getVersion();
-        SubscribedData cachedData = configCache.getDataFor(subscribeKey);
+        SubscribedData cachedData = this.openSergoClient.getConfigCache().getDataFor(subscribeKey);
         if (cachedData != null && cachedData.getVersion() > receivedVersion) {
             // The upcoming data is out-dated, so we'll not resolve the push request.
             return new LocalDataNotifyResult().setCode(OpenSergoTransportConstants.CODE_ERROR_VERSION_OUTDATED);
@@ -67,9 +62,9 @@ private LocalDataNotifyResult notifyDataChange(SubscribeKey subscribeKey, DataWi
         // Decode actual data from the raw "Any" data.
         List<Object> dataList = decodeActualData(subscribeKey.getKind().getKindName(), dataWithVersion.getDataList());
         // Update to local config cache.
-        configCache.updateData(subscribeKey, dataList, receivedVersion);
+        this.openSergoClient.getConfigCache().updateData(subscribeKey, dataList, receivedVersion);
 
-        List<OpenSergoConfigSubscriber> subscribers = subscribeRegistry.getSubscribersOf(subscribeKey);
+        List<OpenSergoConfigSubscriber> subscribers = this.openSergoClient.getSubscribeRegistry().getSubscribersOf(subscribeKey);
         if (subscribers == null || subscribers.isEmpty()) {
             // no-subscriber is acceptable (just for cache-and-pull mode)
             return LocalDataNotifyResult.withSuccess(dataList);
@@ -178,6 +173,11 @@ private List<Object> decodeActualData(String kind, List<Any> rawList) throws Exc
 
     @Override
     public void onError(Throwable t) {
+        // TODO add handles for different io.grpc.Status of Throwable from ClientCallStreamObserver<SubscribeRequest>
+        io.grpc.Status.Code errorCode = io.grpc.Status.fromThrowable(t).getCode();
+        if(errorCode.equals(io.grpc.Status.UNAVAILABLE.getCode())) {
+            this.openSergoClient.status = OpensergoClientStatus.INTERRUPTED;
+        }
         OpenSergoLogger.error("Fatal error occurred on OpenSergo gRPC ClientObserver", t);
     }
 

src/main/java/io/opensergo/OpensergoClientStatus.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2022, OpenSergo Authors
+ *
+ * Licensed 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
+ *
+ *      https://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 io.opensergo;
+
+/**
+ * @author Jiangnan Jia
+ **/
+public enum OpensergoClientStatus {
+
+    /* initial*/
+    INITIAL,
+    STARTING,
+    STARTED,
+    INTERRUPTED,
+    SHUTDOWN
+
+}

src/main/java/io/opensergo/OpensergoClientSubscribeInfo.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2022, OpenSergo Authors
+ *
+ * Licensed 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
+ *
+ *      https://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 io.opensergo;
+
+import com.google.common.collect.Lists;
+import io.opensergo.subscribe.OpenSergoConfigSubscriber;
+import io.opensergo.subscribe.SubscribeKey;
+
+import java.util.List;
+
+/**
+ * @author Jiangnan Jia
+ **/
+public class OpensergoClientSubscribeInfo {
+
+    private SubscribeKey subscribeKey;
+    private List<OpenSergoConfigSubscriber> subscriberList;
+
+    public OpensergoClientSubscribeInfo(SubscribeKey subscribeKey) {
+        this.subscribeKey = subscribeKey;
+        this.subscriberList = Lists.newArrayList();
+    }
+    public OpensergoClientSubscribeInfo(SubscribeKey subscribeKey, List<OpenSergoConfigSubscriber> subscriberList) {
+        this.subscribeKey = subscribeKey;
+        this.subscriberList = subscriberList;
+    }
+
+    public OpensergoClientSubscribeInfo addSubscriber(OpenSergoConfigSubscriber subscriber) {
+        // TODO distinct the same OpenSergoConfigSubscriber
+        this.subscriberList.add(subscriber);
+        return this;
+    }
+
+    public SubscribeKey getSubscribeKey() {
+        return subscribeKey;
+    }
+
+    public List<OpenSergoConfigSubscriber> getSubscriberList() {
+        return subscriberList;
+    }
+
+    @Override
+    public String toString() {
+        return "OpensergoClientSubscribeInfo{" +
+                "subscribeKey=" + subscribeKey +
+                ", subscriberList=" + subscriberList +
+                '}';
+    }
+}

src/main/java/io/opensergo/subscribe/SubscribeRegistry.java

@@ -16,6 +16,7 @@
 package io.opensergo.subscribe;
 
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -33,9 +34,14 @@ public void registerSubscriber(SubscribeKey key, OpenSergoConfigSubscriber subsc
         AssertUtils.assertNotNull(key, "subscribeKey cannot be null");
         AssertUtils.assertNotNull(subscriber, "subscriber cannot be null");
         List<OpenSergoConfigSubscriber> list = subscriberMap.computeIfAbsent(key, v -> new CopyOnWriteArrayList<>());
+        // TODO distinct the same OpenSergoConfigSubscriber
         list.add(subscriber);
     }
 
+    public Set<SubscribeKey> getSubscriberKeysAll() {
+        return subscriberMap.keySet();
+    }
+
     public List<OpenSergoConfigSubscriber> getSubscribersOf(SubscribeKey key) {
         if (key == null) {
             return null;