[Issue apache#718] add synchronized calls for grpc streamObserver

xwm1992 · Feb 16, 2022 · c22bd1e · c22bd1e
1 parent 651850f
commit c22bd1e
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 6 deletions.
diff --git a/...src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/StreamPushRequest.java b/...src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/StreamPushRequest.java
@@ -1,5 +1,6 @@
 package org.apache.eventmesh.runtime.core.protocol.grpc.push;
 
+import io.grpc.stub.StreamObserver;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.collections4.MapUtils;
 import org.apache.commons.lang3.RandomUtils;
@@ -54,7 +55,10 @@ public void tryPushRequest() {
                 .build();
             try {
                 // catch the error and retry, don't use eventEmitter.onNext() to hide the error
-                eventEmitter.getEmitter().onNext(simpleMessage);
+                StreamObserver<SimpleMessage> emitter = eventEmitter.getEmitter();
+                synchronized (emitter) {
+                    emitter.onNext(simpleMessage);
+                }
 
                 long cost = System.currentTimeMillis() - lastPushTime;
                 messageLogger.info(

diff --git a/...e/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/EventEmitter.java b/...e/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/EventEmitter.java
@@ -13,23 +13,23 @@ public EventEmitter(StreamObserver<T> emitter) {
         this.emitter = emitter;
     }
 
-    public void onNext(T event) {
+    public synchronized void onNext(T event) {
         try {
             emitter.onNext(event);
         } catch (Throwable t) {
             logger.warn("StreamObserver Error onNext. {}", t.getMessage());
         }
     }
 
-    public void onCompleted() {
+    public synchronized void onCompleted() {
         try {
             emitter.onCompleted();
         } catch (Throwable t) {
             logger.warn("StreamObserver Error onCompleted. {}", t.getMessage());
         }
     }
 
-    public void onError(Throwable t) {
+    public synchronized void onError(Throwable t) {
         try {
             emitter.onError(t);
         } catch (Throwable t1) {

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
@@ -122,15 +122,19 @@ public void close() {
 
     private void senderOnNext(Subscription subscription) {
         try {
-            sender.onNext(subscription);
+            synchronized (sender) {
+                sender.onNext(subscription);
+            }
         } catch (Throwable t) {
             logger.warn("StreamObserver Error onNext {}", t.getMessage());
         }
     }
 
     private void senderOnComplete() {
         try {
-            sender.onCompleted();
+            synchronized (sender) {
+                sender.onCompleted();
+            }
         } catch (Throwable t) {
             logger.warn("StreamObserver Error onComplete {}", t.getMessage());
         }