WIP

Author: panchenko

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

@@ -888,7 +888,7 @@ static <T> boolean iterableContains(Iterable<T> iterable, T item) {
    *       byte is then represented by the 3-character string "%XY", where "XY" is the two-digit,
    *       uppercase, hexadecimal representation of the byte value.
    * </ul>
-   * 
+   *
    * <p>This section does not use URLEscapers from Guava Net as its not Android-friendly thus core
    *    can't depend on it.
    */
@@ -972,7 +972,16 @@ public static boolean getFlag(String envVarName, boolean enableByDefault) {
     }
   }
 
-
+  /**
+   * Marker to be used in {@link Status#withCause(Throwable)} to signal that stream should be closed
+   * by sending headers.
+   */
+  public static final Throwable CLOSE_WITH_HEADERS = new Throwable("CLOSE_WITH_HEADERS") {
+    @Override
+    public synchronized Throwable fillInStackTrace() {
+      return this;
+    }
+  };
 
   private GrpcUtil() {}
 }

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

@@ -327,22 +327,24 @@ private void messagesAvailableInternal(final MessageProducer producer) {
         return;
       }
 
+      InputStream message = null;
       try {
-        InputStream message;
         while ((message = producer.next()) != null) {
-          ReqT parsedMessage;
-          try (InputStream ignored = message) {
-            parsedMessage = call.method.parseRequest(message);
+          ReqT parsed;
+          try {
+            parsed = call.method.parseRequest(message);
           } catch (StatusRuntimeException e) {
             GrpcUtil.closeQuietly(message);
             GrpcUtil.closeQuietly(producer);
             call.cancelled = true;
-            call.close(e.getStatus(), new Metadata());
+            call.handleInternalError(InternalStatus.asRuntimeExceptionWithoutStacktrace(
+                e.getStatus().withCause(GrpcUtil.CLOSE_WITH_HEADERS), null));
             return;
           }
-          listener.onMessage(parsedMessage);
+          listener.onMessage(parsed);
         }
       } catch (Throwable t) {
+        GrpcUtil.closeQuietly(message);
         GrpcUtil.closeQuietly(producer);
         Throwables.throwIfUnchecked(t);
         throw new RuntimeException(t);

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

@@ -23,6 +23,7 @@
 import io.grpc.Metadata;
 import io.grpc.Status;
 import io.grpc.internal.AbstractServerStream;
+import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.StatsTraceContext;
 import io.grpc.internal.TransportTracer;
 import io.grpc.internal.WritableBuffer;
@@ -130,7 +131,11 @@ public void writeTrailers(Metadata trailers, boolean headersSent, Status status)
     @Override
     public void cancel(Status status) {
       try (TaskCloseable ignore = PerfMark.traceTask("NettyServerStream$Sink.cancel")) {
-        writeQueue.enqueue(CancelServerStreamCommand.withReset(transportState(), status), true);
+        CancelServerStreamCommand cmd =
+            status.getCause() == GrpcUtil.CLOSE_WITH_HEADERS
+                ? CancelServerStreamCommand.withReason(transportState(), status)
+                : CancelServerStreamCommand.withReset(transportState(), status);
+        writeQueue.enqueue(cmd, true);
       }
     }
   }