Skip to content

Commit c9d48a2

Browse files
jnan806jnan806
committed
keepalive OpensergoClient
1 parent 4270c11 commit c9d48a2

File tree

8 files changed

+290
-18
lines changed

8 files changed

+290
-18
lines changed

‎README.md‎

Lines changed: 62 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1,62 @@
1-
# OpenSergo Java SDK
1+
# OpenSergo Java SDK
2+
3+
4+
## How to use
5+
6+
### scene 1 : subscribe config-data
7+
``` java
8+
public static void main(String[] args) throws Exception {
9+
10+
// instant OpenSergoClient
11+
OpenSergoClient openSergoClient = new OpenSergoClient("33.1.33.1", 10246);
12+
13+
// registry SubscribeInfo of FaultToleranceRule
14+
// 1. instant SubscribeKey
15+
SubscribeKey subscribeKey = new SubscribeKey("default", "foo-app", ConfigKind.FAULT_TOLERANCE_RULE);
16+
// 2. construct SubscribeInfo
17+
OpensergoClientSubscribeInfo opensergoClientSubscribeInfo = new OpensergoClientSubscribeInfo(subscribeKey);
18+
opensergoClientSubscribeInfo.addSubscriber(new DefaultFaulttoleranceSubscriber());
19+
// 2. registry
20+
openSergoClient.registerSubscribeInfo(opensergoClientSubscribeInfo);
21+
22+
// start OpenSergoClient
23+
openSergoClient.start();
24+
}
25+
```
26+
27+
### scene 2 : subscribe config-data with custom-logic when config-data changed.
28+
Add a subscriber by implements the function in `subscribe.Subscriber`.
29+
``` java
30+
public class TestSubscriber implements OpenSergoConfigSubscriber {
31+
@Override
32+
public boolean onConfigUpdate(SubscribeKey subscribeKey, Object data) {
33+
System.out.println(data);
34+
return true;
35+
}
36+
}
37+
```
38+
And then registry it to `subscriber.SubscriberRegistry`.
39+
``` java
40+
public static void main(String[] args) throws Exception {
41+
42+
// instant OpenSergoClient
43+
OpenSergoClient openSergoClient = new OpenSergoClient("33.1.33.1", 10246);
44+
45+
// registry SubscribeInfo of FaultToleranceRule
46+
// 1. instant SubscribeKey
47+
SubscribeKey subscribeKey = new SubscribeKey("default", "foo-app", ConfigKind.FAULT_TOLERANCE_RULE);
48+
// 2. instant Subscriber
49+
TestSubscriber testSubscriber = new TestSubscriber();
50+
// 3. construct SubscribeInfo
51+
OpensergoClientSubscribeInfo opensergoClientSubscribeInfo = new OpensergoClientSubscribeInfo(subscribeKey);
52+
opensergoClientSubscribeInfo.addSubscriber(testSubscriber);
53+
// 4. registry
54+
openSergoClient.registerSubscribeInfo(opensergoClientSubscribeInfo);
55+
56+
openSergoClient.start();
57+
58+
// registry after OpenSergoClient started
59+
opensergoClientSubscribeInfo.addSubscriber(new OpensergoConfigDefaultSubscriber());
60+
openSergoClient.registerSubscribeInfo(opensergoClientSubscribeInfo);
61+
}
62+
```

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

Lines changed: 54 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,14 @@
2323
import io.opensergo.proto.transport.v1.SubscribeOpType;
2424
import io.opensergo.proto.transport.v1.SubscribeRequest;
2525
import io.opensergo.proto.transport.v1.SubscribeRequestTarget;
26-
import io.opensergo.subscribe.OpenSergoConfigSubscriber;
2726
import io.opensergo.subscribe.SubscribeKey;
2827
import io.opensergo.subscribe.SubscribeRegistry;
2928
import io.opensergo.subscribe.SubscribedConfigCache;
3029
import io.opensergo.util.AssertUtils;
3130
import io.opensergo.util.IdentifierUtils;
3231

32+
import java.util.Optional;
33+
import java.util.concurrent.TimeUnit;
3334
import java.util.concurrent.atomic.AtomicInteger;
3435

3536
/**
@@ -46,6 +47,7 @@ public class OpenSergoClient implements AutoCloseable {
4647
private final SubscribeRegistry subscribeRegistry;
4748

4849
private AtomicInteger reqId;
50+
protected static volatile OpensergoClientStatus status;
4951

5052
public OpenSergoClient(String host, int port) {
5153
this.channel = ManagedChannelBuilder.forAddress(host, port)
@@ -56,17 +58,64 @@ public OpenSergoClient(String host, int port) {
5658
this.configCache = new SubscribedConfigCache();
5759
this.subscribeRegistry = new SubscribeRegistry();
5860
this.reqId = new AtomicInteger(0);
61+
status = OpensergoClientStatus.INITIAL;
62+
}
63+
64+
public void registerSubscribeInfo(OpensergoClientSubscribeInfo subscribeInfo) {
65+
// Register subscriber to local.
66+
if (Optional.of(subscribeInfo.getSubscriberList()).isPresent() && subscribeInfo.getSubscriberList().size() > 0) {
67+
subscribeInfo.getSubscriberList().forEach(subscriber -> {
68+
this.subscribeRegistry.registerSubscriber(subscribeInfo.getSubscribeKey(), subscriber);
69+
OpenSergoLogger.info("OpenSergo subscribeinfo registered, subscribeKey={}, subscriber={}", subscribeInfo.getSubscribeKey(), subscriber);
70+
});
71+
}
5972
}
6073

6174
public void start() throws Exception {
75+
OpenSergoLogger.info("OpensergoClient is starting...");
76+
77+
if (status == OpensergoClientStatus.INITIAL) {
78+
OpenSergoLogger.info("open keepavlive thread");
79+
new Thread(this::keepAlive).start();
80+
}
81+
82+
status = OpensergoClientStatus.STARTING;
83+
6284
this.requestAndResponseWriter = transportGrpcStub.withWaitForReady()
6385
.subscribeConfig(new OpenSergoSubscribeClientObserver(configCache, subscribeRegistry));
86+
87+
OpenSergoLogger.info("begin to subscribe config-data...");
88+
this.subscribeRegistry.getSubscriberKeysAll().forEach(subscribeKey -> {
89+
this.subscribeConfig(subscribeKey);
90+
});
91+
92+
OpenSergoLogger.info("openSergoClient is started");
93+
status = OpensergoClientStatus.STARTED;
94+
}
95+
96+
private void keepAlive() {
97+
try {
98+
if (status != OpensergoClientStatus.STARTING
99+
&& status != OpensergoClientStatus.STARTED
100+
&& status != OpensergoClientStatus.SHUTDOWN) {
101+
OpenSergoLogger.info("try to restart openSergoClient...");
102+
this.start();
103+
}
104+
Thread.sleep(TimeUnit.SECONDS.toMillis(10));
105+
if( status != OpensergoClientStatus.SHUTDOWN) {
106+
keepAlive();
107+
}
108+
} catch (Exception e) {
109+
e.printStackTrace();
110+
}
64111
}
65112

66113
@Override
67114
public void close() throws Exception {
68115
requestAndResponseWriter.onCompleted();
69116

117+
status = OpensergoClientStatus.SHUTDOWN;
118+
70119
// gracefully drain the requests, then close the connection
71120
channel.shutdown();
72121
}
@@ -77,8 +126,8 @@ public boolean unsubscribeConfig(SubscribeKey subscribeKey) {
77126
AssertUtils.assertNotNull(subscribeKey.getKind(), "kind cannot be null");
78127

79128
if (requestAndResponseWriter == null) {
80-
// TODO: return status that indicates not ready
81-
throw new IllegalStateException("gRPC stream is not ready");
129+
OpenSergoLogger.error("Fatal error occurred on OpenSergo gRPC ClientObserver", new IllegalStateException("gRPC stream is not ready"));
130+
status = OpensergoClientStatus.INTERRUPTED;
82131
}
83132
SubscribeRequestTarget subTarget = SubscribeRequestTarget.newBuilder()
84133
.setNamespace(subscribeKey.getNamespace()).setApp(subscribeKey.getApp())
@@ -97,17 +146,13 @@ public boolean unsubscribeConfig(SubscribeKey subscribeKey) {
97146
}
98147

99148
public boolean subscribeConfig(SubscribeKey subscribeKey) {
100-
return subscribeConfig(subscribeKey, null);
101-
}
102-
103-
public boolean subscribeConfig(SubscribeKey subscribeKey, OpenSergoConfigSubscriber subscriber) {
104149
AssertUtils.assertNotNull(subscribeKey, "subscribeKey cannot be null");
105150
AssertUtils.assertNotNull(subscribeKey.getApp(), "app cannot be null");
106151
AssertUtils.assertNotNull(subscribeKey.getKind(), "kind cannot be null");
107152

108153
if (requestAndResponseWriter == null) {
109-
// TODO: return status that indicates not ready
110-
throw new IllegalStateException("gRPC stream is not ready");
154+
OpenSergoLogger.error("Fatal error occurred on OpenSergo gRPC ClientObserver", new IllegalStateException("gRPC stream is not ready"));
155+
status = OpensergoClientStatus.INTERRUPTED;
111156
}
112157
SubscribeRequestTarget subTarget = SubscribeRequestTarget.newBuilder()
113158
.setNamespace(subscribeKey.getNamespace()).setApp(subscribeKey.getApp())
@@ -121,13 +166,6 @@ public boolean subscribeConfig(SubscribeKey subscribeKey, OpenSergoConfigSubscri
121166
// Send SubscribeRequest
122167
requestAndResponseWriter.onNext(request);
123168

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-
131169
return true;
132170
}
133171

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -178,6 +178,11 @@ private List<Object> decodeActualData(String kind, List<Any> rawList) throws Exc
178178

179179
@Override
180180
public void onError(Throwable t) {
181+
// TODO add handles for different io.grpc.Status of Throwable from ClientCallStreamObserver<SubscribeRequest>
182+
io.grpc.Status.Code errorCode = io.grpc.Status.fromThrowable(t).getCode();
183+
if(errorCode.equals(io.grpc.Status.UNAVAILABLE.getCode())) {
184+
OpenSergoClient.status = OpensergoClientStatus.INTERRUPTED;
185+
}
181186
OpenSergoLogger.error("Fatal error occurred on OpenSergo gRPC ClientObserver", t);
182187
}
183188

‎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: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
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+
import java.util.stream.Collectors;
24+
25+
/**
26+
* @author Jiangnan Jia
27+
**/
28+
public class OpensergoClientSubscribeInfo {
29+
30+
private SubscribeKey subscribeKey;
31+
private List<OpenSergoConfigSubscriber> subscriberList;
32+
33+
public OpensergoClientSubscribeInfo(SubscribeKey subscribeKey) {
34+
this.subscribeKey = subscribeKey;
35+
this.subscriberList = Lists.newArrayList();
36+
}
37+
public OpensergoClientSubscribeInfo(SubscribeKey subscribeKey, List<OpenSergoConfigSubscriber> subscriberList) {
38+
this.subscribeKey = subscribeKey;
39+
this.subscriberList = subscriberList;
40+
}
41+
42+
public OpensergoClientSubscribeInfo addSubscriber(OpenSergoConfigSubscriber subscriber) {
43+
int sameClassNum = this.subscriberList.stream()
44+
.filter(item -> item.getClass() == subscriber.getClass())
45+
.collect(Collectors.toList())
46+
.size();
47+
if (sameClassNum == 0) {
48+
this.subscriberList.add(subscriber);
49+
}
50+
return this;
51+
}
52+
53+
public SubscribeKey getSubscribeKey() {
54+
return subscribeKey;
55+
}
56+
57+
public List<OpenSergoConfigSubscriber> getSubscriberList() {
58+
return subscriberList;
59+
}
60+
61+
@Override
62+
public String toString() {
63+
return "OpensergoClientSubscribeInfo{" +
64+
"subscribeKey=" + subscribeKey +
65+
", subscriberList=" + subscriberList +
66+
'}';
67+
}
68+
}

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

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
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.subscribe;
17+
18+
/**
19+
* @author Jiangnan Jia
20+
**/
21+
public class DefaultFaulttoleranceSubscriber implements OpenSergoConfigSubscriber {
22+
@Override
23+
public boolean onConfigUpdate(SubscribeKey subscribeKey, Object data) {
24+
System.out.println(data);
25+
return true;
26+
}
27+
}

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

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
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.subscribe;
17+
18+
import io.opensergo.log.OpenSergoLogger;
19+
20+
/**
21+
* @author Jiangnan Jia
22+
**/
23+
public class OpensergoConfigDefaultSubscriber implements OpenSergoConfigSubscriber {
24+
@Override
25+
public boolean onConfigUpdate(SubscribeKey subscribeKey, Object data) {
26+
OpenSergoLogger.info(data.toString());
27+
return true;
28+
}
29+
}

0 commit comments

Comments
 (0)