stub: improve error message for Client/ServerCalls (#5656)

grpc · May 2, 2019 · d50c8d4 · d50c8d4
1 parent 97cb055
commit d50c8d4
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 0 deletions.
diff --git a/stub/src/main/java/io/grpc/stub/ClientCalls.java b/stub/src/main/java/io/grpc/stub/ClientCalls.java
@@ -17,6 +17,7 @@
 package io.grpc.stub;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
 
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
@@ -321,6 +322,8 @@ private static final class CallToStreamObserverAdapter<T> extends ClientCallStre
     private final ClientCall<T, ?> call;
     private Runnable onReadyHandler;
     private boolean autoFlowControlEnabled = true;
+    private boolean aborted = false;
+    private boolean completed = false;
 
     // Non private to avoid synthetic class
     CallToStreamObserverAdapter(ClientCall<T, ?> call) {
@@ -333,17 +336,21 @@ private void freeze() {
 
     @Override
     public void onNext(T value) {
+      checkState(!aborted, "Stream was terminated by error, no further calls are allowed");
+      checkState(!completed, "Stream is already completed, no further calls are allowed");
       call.sendMessage(value);
     }
 
     @Override
     public void onError(Throwable t) {
       call.cancel("Cancelled by client with StreamObserver.onError()", t);
+      aborted = true;
     }
 
     @Override
     public void onCompleted() {
       call.halfClose();
+      completed = true;
     }
 
     @Override

diff --git a/stub/src/main/java/io/grpc/stub/ServerCalls.java b/stub/src/main/java/io/grpc/stub/ServerCalls.java
@@ -310,6 +310,8 @@ private static final class ServerCallStreamObserverImpl<ReqT, RespT>
     private boolean sentHeaders;
     private Runnable onReadyHandler;
     private Runnable onCancelHandler;
+    private boolean aborted = false;
+    private boolean completed = false;
 
     // Non private to avoid synthetic class
     ServerCallStreamObserverImpl(ServerCall<ReqT, RespT> call) {
@@ -338,6 +340,8 @@ public void onNext(RespT response) {
         }
         return;
       }
+      checkState(!aborted, "Stream was terminated by error, no further calls are allowed");
+      checkState(!completed, "Stream is already completed, no further calls are allowed");
       if (!sentHeaders) {
         call.sendHeaders(new Metadata());
         sentHeaders = true;
@@ -352,6 +356,7 @@ public void onError(Throwable t) {
         metadata = new Metadata();
       }
       call.close(Status.fromThrowable(t), metadata);
+      aborted = true;
     }
 
     @Override
@@ -362,6 +367,7 @@ public void onCompleted() {
         }
       } else {
         call.close(Status.OK, new Metadata());
+        completed = true;
       }
     }