Skip to content

Commit 32b0cf9

Browse files
jnan806jnan806
committed
force-pushed after main-branch was force-pushed.
Co-authored-by: Eric Zhao <[email protected]> Co-authored-by: Jiangnan Jia <[email protected]> Signed-off-by: Jiangnan Jia <[email protected]>
1 parent 6ca87a2 commit 32b0cf9

File tree

6 files changed

+177
-25
lines changed

6 files changed

+177
-25
lines changed

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,7 @@ public enum ConfigKind {
2626
FAULT_TOLERANCE_RULE("fault-tolerance.opensergo.io/v1alpha1/FaultToleranceRule", "FaultToleranceRule"),
2727
RATE_LIMIT_STRATEGY("fault-tolerance.opensergo.io/v1alpha1/RateLimitStrategy", "RateLimitStrategy"),
2828
THROTTLING_STRATEGY("fault-tolerance.opensergo.io/v1alpha1/ThrottlingStrategy", "ThrottlingStrategy"),
29-
CONCURRENCY_LIMIT_STRATEGY("fault-tolerance.opensergo.io/v1alpha1/ConcurrencyLimitStrategy",
30-
"ConcurrencyLimitStrategy"),
29+
CONCURRENCY_LIMIT_STRATEGY("fault-tolerance.opensergo.io/v1alpha1/ConcurrencyLimitStrategy", "ConcurrencyLimitStrategy"),
3130
CIRCUIT_BREAKER_STRATEGY("fault-tolerance.opensergo.io/v1alpha1/CircuitBreakerStrategy", "CircuitBreakerStrategy");
3231

3332
private final String kindName;

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

Lines changed: 67 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,8 @@
3030
import io.opensergo.util.AssertUtils;
3131
import io.opensergo.util.IdentifierUtils;
3232

33+
import java.util.Optional;
34+
import java.util.concurrent.TimeUnit;
3335
import java.util.concurrent.atomic.AtomicInteger;
3436

3537
/**
@@ -46,6 +48,7 @@ public class OpenSergoClient implements AutoCloseable {
4648
private final SubscribeRegistry subscribeRegistry;
4749

4850
private AtomicInteger reqId;
51+
protected volatile OpenSergoClientStatus status;
4952

5053
public OpenSergoClient(String host, int port) {
5154
this.channel = ManagedChannelBuilder.forAddress(host, port)
@@ -56,17 +59,72 @@ public OpenSergoClient(String host, int port) {
5659
this.configCache = new SubscribedConfigCache();
5760
this.subscribeRegistry = new SubscribeRegistry();
5861
this.reqId = new AtomicInteger(0);
62+
status = OpenSergoClientStatus.INITIAL;
63+
}
64+
65+
public void registerSubscribeInfo(OpenSergoClientSubscribeInfo subscribeInfo) {
66+
// Register subscriber to local.
67+
if (Optional.of(subscribeInfo.getSubscriberList()).isPresent() && subscribeInfo.getSubscriberList().size() > 0) {
68+
subscribeInfo.getSubscriberList().forEach(subscriber -> {
69+
this.subscribeRegistry.registerSubscriber(subscribeInfo.getSubscribeKey(), subscriber);
70+
OpenSergoLogger.info("OpenSergo subscribeinfo registered, subscribeKey={}, subscriber={}", subscribeInfo.getSubscribeKey(), subscriber);
71+
72+
if (requestAndResponseWriter != null && this.status == OpenSergoClientStatus.STARTED) {
73+
this.subscribeConfig(subscribeInfo.getSubscribeKey());
74+
}
75+
});
76+
}
5977
}
6078

6179
public void start() throws Exception {
80+
OpenSergoLogger.info("OpensergoClient is starting...");
81+
82+
if (status == OpenSergoClientStatus.INITIAL) {
83+
OpenSergoLogger.info("open keepavlive thread");
84+
Thread keepAliveThread = new Thread(this::keepAlive);
85+
keepAliveThread.setName("thread-opensergo-keepalive-" + keepAliveThread.getId());
86+
keepAliveThread.setDaemon(true);
87+
keepAliveThread.start();
88+
}
89+
90+
status = OpenSergoClientStatus.STARTING;
91+
6292
this.requestAndResponseWriter = transportGrpcStub.withWaitForReady()
63-
.subscribeConfig(new OpenSergoSubscribeClientObserver(configCache, subscribeRegistry));
93+
.subscribeConfig(new OpenSergoSubscribeClientObserver(this));
94+
95+
OpenSergoLogger.info("begin to subscribe config-data...");
96+
this.subscribeRegistry.getSubscriberKeysAll().forEach(subscribeKey -> {
97+
this.subscribeConfig(subscribeKey);
98+
});
99+
100+
OpenSergoLogger.info("openSergoClient is started");
101+
status = OpenSergoClientStatus.STARTED;
102+
}
103+
104+
private void keepAlive() {
105+
try {
106+
if (status != OpenSergoClientStatus.STARTING
107+
&& status != OpenSergoClientStatus.STARTED
108+
&& status != OpenSergoClientStatus.SHUTDOWN) {
109+
OpenSergoLogger.info("try to restart openSergoClient...");
110+
this.start();
111+
}
112+
Thread.sleep(TimeUnit.SECONDS.toMillis(10));
113+
if( status != OpenSergoClientStatus.SHUTDOWN) {
114+
keepAlive();
115+
}
116+
} catch (Exception e) {
117+
e.printStackTrace();
118+
}
64119
}
65120

66121
@Override
67122
public void close() throws Exception {
68123
requestAndResponseWriter.onCompleted();
69124

125+
// stop the keepAliveThread
126+
status = OpenSergoClientStatus.SHUTDOWN;
127+
70128
// gracefully drain the requests, then close the connection
71129
channel.shutdown();
72130
}
@@ -77,8 +135,8 @@ public boolean unsubscribeConfig(SubscribeKey subscribeKey) {
77135
AssertUtils.assertNotNull(subscribeKey.getKind(), "kind cannot be null");
78136

79137
if (requestAndResponseWriter == null) {
80-
// TODO: return status that indicates not ready
81-
throw new IllegalStateException("gRPC stream is not ready");
138+
OpenSergoLogger.error("Fatal error occurred on OpenSergo gRPC ClientObserver", new IllegalStateException("gRPC stream is not ready"));
139+
status = OpenSergoClientStatus.INTERRUPTED;
82140
}
83141
SubscribeRequestTarget subTarget = SubscribeRequestTarget.newBuilder()
84142
.setNamespace(subscribeKey.getNamespace()).setApp(subscribeKey.getApp())
@@ -106,8 +164,8 @@ public boolean subscribeConfig(SubscribeKey subscribeKey, OpenSergoConfigSubscri
106164
AssertUtils.assertNotNull(subscribeKey.getKind(), "kind cannot be null");
107165

108166
if (requestAndResponseWriter == null) {
109-
// TODO: return status that indicates not ready
110-
throw new IllegalStateException("gRPC stream is not ready");
167+
OpenSergoLogger.error("Fatal error occurred on OpenSergo gRPC ClientObserver", new IllegalStateException("gRPC stream is not ready"));
168+
status = OpenSergoClientStatus.INTERRUPTED;
111169
}
112170
SubscribeRequestTarget subTarget = SubscribeRequestTarget.newBuilder()
113171
.setNamespace(subscribeKey.getNamespace()).setApp(subscribeKey.getApp())
@@ -121,18 +179,15 @@ public boolean subscribeConfig(SubscribeKey subscribeKey, OpenSergoConfigSubscri
121179
// Send SubscribeRequest
122180
requestAndResponseWriter.onNext(request);
123181

124-
// Register subscriber to local.
125-
if (subscriber != null) {
126-
subscribeRegistry.registerSubscriber(subscribeKey, subscriber);
127-
OpenSergoLogger.info("OpenSergo config subscriber registered, subscribeKey={}, subscriber={}",
128-
subscribeKey, subscriber);
129-
}
130-
131182
return true;
132183
}
133184

134185
public SubscribedConfigCache getConfigCache() {
135186
return configCache;
136187
}
137188

189+
public SubscribeRegistry getSubscribeRegistry() {
190+
return subscribeRegistry;
191+
}
192+
138193
}

‎src/main/java/io/opensergo/OpenSergoClientStatus.java

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
/*
2+
* Copyright 2022, OpenSergo Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.opensergo;
17+
18+
/**
19+
* @author Jiangnan Jia
20+
**/
21+
public enum OpenSergoClientStatus {
22+
23+
/* initial*/
24+
INITIAL,
25+
STARTING,
26+
STARTED,
27+
INTERRUPTED,
28+
SHUTDOWN
29+
30+
}

‎src/main/java/io/opensergo/OpenSergoClientSubscribeInfo.java

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
/*
2+
* Copyright 2022, OpenSergo Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.opensergo;
17+
18+
import com.google.common.collect.Lists;
19+
import io.opensergo.subscribe.OpenSergoConfigSubscriber;
20+
import io.opensergo.subscribe.SubscribeKey;
21+
22+
import java.util.List;
23+
24+
/**
25+
* @author Jiangnan Jia
26+
**/
27+
public class OpenSergoClientSubscribeInfo {
28+
29+
private SubscribeKey subscribeKey;
30+
private List<OpenSergoConfigSubscriber> subscriberList;
31+
32+
public OpenSergoClientSubscribeInfo(SubscribeKey subscribeKey) {
33+
this.subscribeKey = subscribeKey;
34+
this.subscriberList = Lists.newArrayList();
35+
}
36+
public OpenSergoClientSubscribeInfo(SubscribeKey subscribeKey, List<OpenSergoConfigSubscriber> subscriberList) {
37+
this.subscribeKey = subscribeKey;
38+
this.subscriberList = subscriberList;
39+
}
40+
41+
public OpenSergoClientSubscribeInfo addSubscriber(OpenSergoConfigSubscriber subscriber) {
42+
// TODO distinct the same OpenSergoConfigSubscriber
43+
this.subscriberList.add(subscriber);
44+
return this;
45+
}
46+
47+
public SubscribeKey getSubscribeKey() {
48+
return subscribeKey;
49+
}
50+
51+
public List<OpenSergoConfigSubscriber> getSubscriberList() {
52+
return subscriberList;
53+
}
54+
55+
@Override
56+
public String toString() {
57+
return "OpensergoClientSubscribeInfo{" +
58+
"subscribeKey=" + subscribeKey +
59+
", subscriberList=" + subscriberList +
60+
'}';
61+
}
62+
}

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

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,6 @@
3030
import io.opensergo.proto.transport.v1.SubscribeResponse;
3131
import io.opensergo.subscribe.OpenSergoConfigSubscriber;
3232
import io.opensergo.subscribe.SubscribeKey;
33-
import io.opensergo.subscribe.SubscribeRegistry;
34-
import io.opensergo.subscribe.SubscribedConfigCache;
3533
import io.opensergo.util.StringUtils;
3634

3735
/**
@@ -41,13 +39,10 @@ public class OpenSergoSubscribeClientObserver implements ClientResponseObserver<
4139

4240
private ClientCallStreamObserver<SubscribeRequest> requestStream;
4341

44-
private final SubscribedConfigCache configCache;
45-
private final SubscribeRegistry subscribeRegistry;
42+
private OpenSergoClient openSergoClient;
4643

47-
public OpenSergoSubscribeClientObserver(SubscribedConfigCache configCache,
48-
SubscribeRegistry subscribeRegistry) {
49-
this.configCache = configCache;
50-
this.subscribeRegistry = subscribeRegistry;
44+
public OpenSergoSubscribeClientObserver(OpenSergoClient openSergoClient) {
45+
this.openSergoClient = openSergoClient;
5146
}
5247

5348
@Override
@@ -58,7 +53,7 @@ public void beforeStart(ClientCallStreamObserver<SubscribeRequest> requestStream
5853
private LocalDataNotifyResult notifyDataChange(SubscribeKey subscribeKey, DataWithVersion dataWithVersion)
5954
throws Exception {
6055
long receivedVersion = dataWithVersion.getVersion();
61-
SubscribedData cachedData = configCache.getDataFor(subscribeKey);
56+
SubscribedData cachedData = this.openSergoClient.getConfigCache().getDataFor(subscribeKey);
6257
if (cachedData != null && cachedData.getVersion() > receivedVersion) {
6358
// The upcoming data is out-dated, so we'll not resolve the push request.
6459
return new LocalDataNotifyResult().setCode(OpenSergoTransportConstants.CODE_ERROR_VERSION_OUTDATED);
@@ -67,9 +62,9 @@ private LocalDataNotifyResult notifyDataChange(SubscribeKey subscribeKey, DataWi
6762
// Decode actual data from the raw "Any" data.
6863
List<Object> dataList = decodeActualData(subscribeKey.getKind().getKindName(), dataWithVersion.getDataList());
6964
// Update to local config cache.
70-
configCache.updateData(subscribeKey, dataList, receivedVersion);
65+
this.openSergoClient.getConfigCache().updateData(subscribeKey, dataList, receivedVersion);
7166

72-
List<OpenSergoConfigSubscriber> subscribers = subscribeRegistry.getSubscribersOf(subscribeKey);
67+
List<OpenSergoConfigSubscriber> subscribers = this.openSergoClient.getSubscribeRegistry().getSubscribersOf(subscribeKey);
7368
if (subscribers == null || subscribers.isEmpty()) {
7469
// no-subscriber is acceptable (just for cache-and-pull mode)
7570
return LocalDataNotifyResult.withSuccess(dataList);
@@ -178,6 +173,11 @@ private List<Object> decodeActualData(String kind, List<Any> rawList) throws Exc
178173

179174
@Override
180175
public void onError(Throwable t) {
176+
// TODO add handles for different io.grpc.Status of Throwable from ClientCallStreamObserver<SubscribeRequest>
177+
io.grpc.Status.Code errorCode = io.grpc.Status.fromThrowable(t).getCode();
178+
if(errorCode.equals(io.grpc.Status.UNAVAILABLE.getCode())) {
179+
this.openSergoClient.status = OpenSergoClientStatus.INTERRUPTED;
180+
}
181181
OpenSergoLogger.error("Fatal error occurred on OpenSergo gRPC ClientObserver", t);
182182
}
183183

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
package io.opensergo.subscribe;
1717

1818
import java.util.List;
19+
import java.util.Set;
1920
import java.util.concurrent.ConcurrentHashMap;
2021
import java.util.concurrent.ConcurrentMap;
2122
import java.util.concurrent.CopyOnWriteArrayList;
@@ -33,9 +34,14 @@ public void registerSubscriber(SubscribeKey key, OpenSergoConfigSubscriber subsc
3334
AssertUtils.assertNotNull(key, "subscribeKey cannot be null");
3435
AssertUtils.assertNotNull(subscriber, "subscriber cannot be null");
3536
List<OpenSergoConfigSubscriber> list = subscriberMap.computeIfAbsent(key, v -> new CopyOnWriteArrayList<>());
37+
// TODO distinct the same OpenSergoConfigSubscriber
3638
list.add(subscriber);
3739
}
3840

41+
public Set<SubscribeKey> getSubscriberKeysAll() {
42+
return subscriberMap.keySet();
43+
}
44+
3945
public List<OpenSergoConfigSubscriber> getSubscribersOf(SubscribeKey key) {
4046
if (key == null) {
4147
return null;

0 commit comments

Comments
 (0)