Handle exception loop by closeOnError

apache · Apr 2, 2024 · a540e8a · a540e8a
1 parent 172804a
commit a540e8a
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/...sh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java b/...sh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
@@ -96,6 +96,7 @@ public void onNext(final CloudEvent message) {
             @Override
             public void onError(final Throwable t) {
                 log.error("Received Server side error", t);
+                close();
             }
 
             @Override
@@ -145,13 +146,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;
             }
         } catch (Exception e) {
             log.error("StreamObserver Error onComplete", e);