core: change serverimpl,servercallimpl's internalclose to cancel stre…

…am (#4038)

The HTTP/2 error code will be INTERNAL_ERROR for all cancel statuses,
except for DEADLINE_EXCEEDED and CANCELLED, which are mapped to
CANCELLED.

core/src/main/java/io/grpc/internal/ServerCallImpl.java

@@ -37,6 +37,7 @@
 import io.grpc.ServerCall;
 import io.grpc.Status;
 import java.io.InputStream;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 final class ServerCallImpl<ReqT, RespT> extends ServerCall<ReqT, RespT> {
@@ -205,7 +206,8 @@ public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
    * on.
    */
   private void internalClose(Status internalError) {
-    stream.close(internalError, new Metadata());
+    log.log(Level.WARNING, "Cancelling the stream with status {0}", new Object[] {internalError});
+    stream.cancel(internalError);
   }
 
   /**

core/src/main/java/io/grpc/internal/ServerImpl.java

@@ -629,8 +629,7 @@ void setListener(ServerStreamListener listener) {
      * Like {@link ServerCall#close(Status, Metadata)}, but thread-safe for internal use.
      */
     private void internalClose() {
-      // TODO(ejona86): this is not thread-safe :)
-      stream.close(Status.UNKNOWN, new Metadata());
+      stream.cancel(Status.INTERNAL);
     }
 
     @Override

core/src/test/java/io/grpc/internal/ServerCallImplTest.java

@@ -192,10 +192,9 @@ private void sendMessage_serverSendsOne_closeOnSecondCall(
     verify(stream, times(1)).writeMessage(any(InputStream.class));
     ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
     ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
-    verify(stream, times(1)).close(statusCaptor.capture(), metadataCaptor.capture());
+    verify(stream, times(1)).cancel(statusCaptor.capture());
     assertEquals(Status.Code.INTERNAL, statusCaptor.getValue().getCode());
     assertEquals(ServerCallImpl.TOO_MANY_RESPONSES, statusCaptor.getValue().getDescription());
-    assertTrue(metadataCaptor.getValue().keys().isEmpty());
   }
 
   @Test
@@ -221,7 +220,7 @@ private void sendMessage_serverSendsOne_closeOnSecondCall_appRunToCompletion(
     serverCall.sendMessage(1L);
     serverCall.sendMessage(1L);
     verify(stream, times(1)).writeMessage(any(InputStream.class));
-    verify(stream, times(1)).close(any(Status.class), any(Metadata.class));
+    verify(stream, times(1)).cancel(any(Status.class));
 
     // App runs to completion but everything is ignored
     serverCall.sendMessage(1L);
@@ -255,11 +254,9 @@ private void serverSendsOne_okFailsOnMissingResponse(
         CompressorRegistry.getDefaultInstance());
     serverCall.close(Status.OK, new Metadata());
     ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
-    ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
-    verify(stream, times(1)).close(statusCaptor.capture(), metadataCaptor.capture());
+    verify(stream, times(1)).cancel(statusCaptor.capture());
     assertEquals(Status.Code.INTERNAL, statusCaptor.getValue().getCode());
     assertEquals(ServerCallImpl.MISSING_RESPONSE, statusCaptor.getValue().getDescription());
-    assertTrue(metadataCaptor.getValue().keys().isEmpty());
   }
 
   @Test

core/src/test/java/io/grpc/internal/ServerImplTest.java

@@ -1108,7 +1108,7 @@ public void messageRead_errorCancelsCall() throws Exception {
       fail("Expected exception");
     } catch (TestError t) {
       assertSame(expectedT, t);
-      ensureServerStateNotLeaked();
+      ensureServerStateIsCancelled();
     }
   }
 
@@ -1133,7 +1133,7 @@ public void messageRead_runtimeExceptionCancelsCall() throws Exception {
       fail("Expected exception");
     } catch (RuntimeException t) {
       assertSame(expectedT, t);
-      ensureServerStateNotLeaked();
+      ensureServerStateIsCancelled();
     }
   }
 
@@ -1156,7 +1156,7 @@ public void halfClosed_errorCancelsCall() {
       fail("Expected exception");
     } catch (TestError t) {
       assertSame(expectedT, t);
-      ensureServerStateNotLeaked();
+      ensureServerStateIsCancelled();
     }
   }
 
@@ -1179,7 +1179,7 @@ public void halfClosed_runtimeExceptionCancelsCall() {
       fail("Expected exception");
     } catch (RuntimeException t) {
       assertSame(expectedT, t);
-      ensureServerStateNotLeaked();
+      ensureServerStateIsCancelled();
     }
   }
 
@@ -1202,7 +1202,7 @@ public void onReady_errorCancelsCall() {
       fail("Expected exception");
     } catch (TestError t) {
       assertSame(expectedT, t);
-      ensureServerStateNotLeaked();
+      ensureServerStateIsCancelled();
     }
   }
 
@@ -1225,7 +1225,7 @@ public void onReady_runtimeExceptionCancelsCall() {
       fail("Expected exception");
     } catch (RuntimeException t) {
       assertSame(expectedT, t);
-      ensureServerStateNotLeaked();
+      ensureServerStateIsCancelled();
     }
   }
 
@@ -1396,6 +1396,12 @@ private void ensureServerStateNotLeaked() {
     assertTrue(metadataCaptor.getValue().keys().isEmpty());
   }
 
+  private void ensureServerStateIsCancelled() {
+    verify(stream).cancel(statusCaptor.capture());
+    assertEquals(Status.INTERNAL, statusCaptor.getValue());
+    assertNull(statusCaptor.getValue().getCause());
+  }
+
   private static class SimpleServer implements io.grpc.internal.InternalServer {
     ServerListener listener;
 

interop-testing/src/test/java/io/grpc/testing/integration/MoreInProcessTest.java

@@ -247,7 +247,7 @@ public void onCompleted() {
         .onNext(StreamingInputCallRequest.getDefaultInstance());
 
     assertTrue(finishLatch.await(900, TimeUnit.MILLISECONDS));
-    assertEquals(Status.UNKNOWN, Status.fromThrowable(throwableRef.get()));
+    assertEquals(Status.CANCELLED.getCode(), Status.fromThrowable(throwableRef.get()).getCode());
     assertNull(responseRef.get());
   }
 }

netty/src/main/java/io/grpc/netty/NettyServerHandler.java

@@ -654,8 +654,13 @@ private void cancelStream(ChannelHandlerContext ctx, CancelServerStreamCommand c
       ChannelPromise promise) {
     // Notify the listener if we haven't already.
     cmd.stream().transportReportStatus(cmd.reason());
+    Http2Error http2Error = Http2Error.INTERNAL_ERROR;
+    if (Status.DEADLINE_EXCEEDED.getCode().equals(cmd.reason().getCode()) || Status.CANCELLED
+        .getCode().equals(cmd.reason().getCode())) {
+      http2Error = Http2Error.CANCEL;
+    }
     // Terminate the stream.
-    encoder().writeRstStream(ctx, cmd.stream().id(), Http2Error.CANCEL.code(), promise);
+    encoder().writeRstStream(ctx, cmd.stream().id(), http2Error.code(), promise);
   }
 
   private void forcefulClose(final ChannelHandlerContext ctx, final ForcefulCloseCommand msg,