This is an automated email from the ASF dual-hosted git repository.
mxsm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new fe2bd7c84 [ISSUE #4804] Fix SubStreamHandler exception loop by
closeOnError (#4807)
fe2bd7c84 is described below
commit fe2bd7c840c9900ad7143099d89a1cc1905a9a09
Author: Pil0tXia <[email protected]>
AuthorDate: Sat Apr 13 22:01:59 2024 +0800
[ISSUE #4804] Fix SubStreamHandler exception loop by closeOnError (#4807)
* Handle exception loop by closeOnError
* Lombok optimization
* some format optimization
* Avoid closing multiple times
* Remove redundant set null
* Revert "Avoid closing multiple times"
This reverts commit 774397fcae4905dc0ec8230eab18bf655f202e8f.
* Use synchronized latch to keep senderOnComplete called once
* Use boolean to prevent latch called by somebody else
* Remove the unique callee/caller close() of onCompleted()
---
.../grpc/consumer/EventMeshGrpcConsumer.java | 46 ++++++----------------
.../client/grpc/consumer/SubStreamHandler.java | 4 +-
2 files changed, 14 insertions(+), 36 deletions(-)
diff --git
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
index 6ceefbe76..623ca1c6f 100644
---
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
+++
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
@@ -87,8 +87,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
}
public void init() {
- this.channel =
ManagedChannelBuilder.forAddress(clientConfig.getServerAddr(),
clientConfig.getServerPort()).usePlaintext()
- .build();
+ this.channel =
ManagedChannelBuilder.forAddress(clientConfig.getServerAddr(),
clientConfig.getServerPort()).usePlaintext().build();
this.consumerClient = ConsumerServiceGrpc.newBlockingStub(channel);
this.consumerAsyncClient = ConsumerServiceGrpc.newStub(channel);
this.heartbeatClient = HeartbeatServiceGrpc.newBlockingStub(channel);
@@ -125,8 +124,8 @@ public class EventMeshGrpcConsumer implements AutoCloseable
{
addSubscription(subscriptionItems, SDK_STREAM_URL, GrpcType.STREAM);
- CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(clientConfig,
EventMeshProtocolType.EVENT_MESH_MESSAGE, null,
- subscriptionItems);
+ CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(
+ clientConfig, EventMeshProtocolType.EVENT_MESH_MESSAGE, null,
subscriptionItems);
synchronized (this) {
if (subStreamHandler == null) {
subStreamHandler = new SubStreamHandler<>(consumerAsyncClient,
clientConfig, listener);
@@ -137,8 +136,8 @@ public class EventMeshGrpcConsumer implements AutoCloseable
{
}
private Response subscribeWebhook(List<SubscriptionItem>
subscriptionItems, String url) {
- final CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(clientConfig,
EventMeshProtocolType.EVENT_MESH_MESSAGE,
- url, subscriptionItems);
+ final CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(
+ clientConfig, EventMeshProtocolType.EVENT_MESH_MESSAGE, url,
subscriptionItems);
try {
CloudEvent response = consumerClient.subscribe(subscription);
log.info("Received response:{}", response);
@@ -169,8 +168,8 @@ public class EventMeshGrpcConsumer implements AutoCloseable
{
removeSubscription(subscriptionItems);
- final CloudEvent cloudEvent =
EventMeshCloudEventBuilder.buildEventSubscription(clientConfig,
EventMeshProtocolType.EVENT_MESH_MESSAGE, url,
- subscriptionItems);
+ final CloudEvent cloudEvent =
EventMeshCloudEventBuilder.buildEventSubscription(
+ clientConfig, EventMeshProtocolType.EVENT_MESH_MESSAGE, url,
subscriptionItems);
try {
final CloudEvent response = consumerClient.unsubscribe(cloudEvent);
log.info("Received response:{}", response);
@@ -191,8 +190,8 @@ public class EventMeshGrpcConsumer implements AutoCloseable
{
removeSubscription(subscriptionItems);
- final CloudEvent cloudEvent =
EventMeshCloudEventBuilder.buildEventSubscription(clientConfig,
EventMeshProtocolType.EVENT_MESH_MESSAGE, null,
- subscriptionItems);
+ final CloudEvent cloudEvent =
EventMeshCloudEventBuilder.buildEventSubscription(
+ clientConfig, EventMeshProtocolType.EVENT_MESH_MESSAGE, null,
subscriptionItems);
try {
final CloudEvent response = consumerClient.unsubscribe(cloudEvent);
@@ -277,14 +276,12 @@ public class EventMeshGrpcConsumer implements
AutoCloseable {
subscriptionGroup.forEach((url, items) -> {
if (isStreamSub.get()) {
- CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(clientConfig,
EventMeshProtocolType.EVENT_MESH_MESSAGE,
- url,
- items);
+ CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(
+ clientConfig, EventMeshProtocolType.EVENT_MESH_MESSAGE,
url, items);
subStreamHandler.sendSubscription(subscription);
} else {
subscribeWebhook(items, url);
}
-
});
}
@@ -303,6 +300,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable
{
}
}
+ @Data
private static class SubscriptionInfo {
private transient SubscriptionItem subscriptionItem;
@@ -314,25 +312,5 @@ public class EventMeshGrpcConsumer implements
AutoCloseable {
this.url = url;
this.grpcType = grpcType;
}
-
- public GrpcType getGrpcType() {
- return grpcType;
- }
-
- public SubscriptionItem getSubscriptionItem() {
- return subscriptionItem;
- }
-
- public void setSubscriptionItem(final SubscriptionItem
subscriptionItem) {
- this.subscriptionItem = subscriptionItem;
- }
-
- public String getUrl() {
- return url;
- }
-
- public void setUrl(final String url) {
- this.url = url;
- }
}
}
diff --git
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
index d34a0c79b..59436be08 100644
---
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
+++
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
@@ -96,12 +96,12 @@ public class SubStreamHandler<T> extends Thread implements
Serializable {
@Override
public void onError(final Throwable t) {
log.error("Received Server side error", t);
+ close();
}
@Override
public void onCompleted() {
log.info("Finished receiving messages from server.");
- close();
}
};
}
@@ -134,7 +134,6 @@ public class SubStreamHandler<T> extends Thread implements
Serializable {
}
latch.countDown();
-
log.info("SubStreamHandler closed.");
}
@@ -145,6 +144,7 @@ public class SubStreamHandler<T> extends Thread implements
Serializable {
}
} catch (Exception e) {
log.error("StreamObserver Error onNext", e);
+ close();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
