Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[ISSUE #4804] Fix SubStreamHandler exception loop by closeOnError #4807

Merged
merged 9 commits into from
Apr 13, 2024
Original file line number Diff line number Diff line change
@@ -87,8 +87,7 @@ public EventMeshGrpcConsumer(final EventMeshGrpcClientConfig clientConfig) {
}

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 void subscribe(final List<SubscriptionItem> subscriptionItems) {

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 void subscribe(final List<SubscriptionItem> subscriptionItems) {
}

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 Response unsubscribe(final List<SubscriptionItem> subscriptionItems, fina

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 Response unsubscribe(final List<SubscriptionItem> subscriptionItems) {

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 @@ private void resubscribe() {

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 void close() {
}
}

@Data
private static class SubscriptionInfo {

private transient SubscriptionItem subscriptionItem;
@@ -314,25 +312,5 @@ private static class SubscriptionInfo {
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;
}
}
}
Original file line number Diff line number Diff line change
@@ -40,6 +40,8 @@ public class SubStreamHandler<T> extends Thread implements Serializable {

private final transient CountDownLatch latch = new CountDownLatch(1);

private volatile boolean isClosed = false;

private final transient ConsumerServiceStub consumerAsyncClient;

private final transient EventMeshGrpcClientConfig clientConfig;
@@ -96,6 +98,7 @@ public void onNext(final CloudEvent message) {
@Override
public void onError(final Throwable t) {
log.error("Received Server side error", t);
close();
}

@Override
@@ -129,6 +132,12 @@ public void run() {
}

public void close() {
// Avoid closing multiple times
if (isClosed) {
return;
}
isClosed = true;
Pil0tXia marked this conversation as resolved.
Show resolved Hide resolved

if (this.sender != null) {
senderOnComplete();
}
@@ -145,13 +154,15 @@ private void senderOnNext(final CloudEvent subscription) {
}
} catch (Exception e) {
log.error("StreamObserver Error onNext", e);
close();
}
}

private void senderOnComplete() {
try {
synchronized (sender) {
sender.onCompleted();
sender = null;
Pil0tXia marked this conversation as resolved.
Show resolved Hide resolved
}
} catch (Exception e) {
log.error("StreamObserver Error onComplete", e);