core, binder: notify clientCancelled directly in transportReportStatus and add binder cancel test

Author: AgraVator

binder/src/androidTest/java/io/grpc/binder/internal/BinderClientTransportTest.java

@@ -292,6 +292,22 @@ public void testMessageProducerClosedAfterStream_b169313545() throws Exception {
     streamListener.drainMessages();
   }
 
+  @Test
+  public void testCancelStream_notifiesTracer() throws Exception {
+    transport = new BinderClientTransportBuilder().build();
+    startAndAwaitReady(transport, transportListener);
+    
+    ClientStreamTracer mockTracer = org.mockito.Mockito.mock(ClientStreamTracer.class);
+    ClientStream stream =
+        transport.newStream(methodDesc, new Metadata(), CallOptions.DEFAULT, new ClientStreamTracer[]{mockTracer});
+
+    stream.start(streamListener);
+    Status cancelStatus = Status.CANCELLED.withDescription("Client cancelled");
+    stream.cancel(cancelStatus);
+
+    org.mockito.Mockito.verify(mockTracer, org.mockito.Mockito.timeout(5000)).cancelled(org.mockito.ArgumentMatchers.eq(cancelStatus));
+  }
+
   @Test
   public void testNewStreamBeforeTransportReadyFails() throws Exception {
     // Use a special SecurityPolicy that lets us act before the transport is setup/ready.

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

@@ -197,9 +197,6 @@ public final void halfClose() {
   @Override
   public final void cancel(Status reason) {
     Preconditions.checkArgument(!reason.isOk(), "Should not cancel with OK status");
-    if (cancelled || transportState().isListenerClosed()) {
-      return;
-    }
     cancelled = true;
     abstractClientStreamSink().cancel(reason);
   }
@@ -254,9 +251,7 @@ protected TransportState(
       }
     }
 
-    protected final boolean isListenerClosed() {
-      return listenerClosed;
-    }
+
 
     private void setFullStreamDecompression(boolean fullStreamDecompression) {
       this.fullStreamDecompression = fullStreamDecompression;
@@ -398,16 +393,16 @@ protected void inboundTrailersReceived(Metadata trailers, Status status) {
      * method must be called from the transport thread.
      *
      * @param status the new status to set
-     * @param stopDelivery if {@code true}, interrupts any further delivery of inbound messages that
+     * @param cancelled if {@code true}, interrupts any further delivery of inbound messages that
      *        may already be queued up in the deframer. If {@code false}, the listener will be
      *        notified immediately after all currently completed messages in the deframer have been
      *        delivered to the application.
      * @param trailers new instance of {@code Trailers}, either empty or those returned by the
      *        server
      */
-    public final void transportReportStatus(final Status status, boolean stopDelivery,
+    public final void transportReportStatus(final Status status, boolean cancelled,
         final Metadata trailers) {
-      transportReportStatus(status, RpcProgress.PROCESSED, stopDelivery, trailers);
+      transportReportStatus(status, RpcProgress.PROCESSED, cancelled, trailers);
     }
 
     /**
@@ -418,7 +413,7 @@ public final void transportReportStatus(final Status status, boolean stopDeliver
      * @param rpcProgress RPC progress that the
      *        {@link ClientStreamListener#closed(Status, RpcProgress, Metadata)}
      *        will receive
-     * @param stopDelivery if {@code true}, interrupts any further delivery of inbound messages that
+     * @param cancelled if {@code true}, interrupts any further delivery of inbound messages that
      *        may already be queued up in the deframer and overrides any previously queued status.
      *        If {@code false}, the listener will be notified immediately after all currently
      *        completed messages in the deframer have been delivered to the application.
@@ -428,30 +423,33 @@ public final void transportReportStatus(final Status status, boolean stopDeliver
     public final void transportReportStatus(
         final Status status,
         final RpcProgress rpcProgress,
-        boolean stopDelivery,
+        boolean cancelled,
         final Metadata trailers) {
       checkNotNull(status, "status");
       checkNotNull(trailers, "trailers");
-      // If stopDelivery, we continue in case previous invocation is waiting for stall
-      if (statusReported && !stopDelivery) {
+      // If cancelled, we continue in case previous invocation is waiting for stall
+      if (statusReported && !cancelled) {
         return;
       }
       statusReported = true;
       statusReportedIsOk = status.isOk();
+      if (cancelled) {
+        statsTraceCtx.clientCancelled(status);
+      }
       onStreamDeallocated();
 
       if (deframerClosed) {
         deframerClosedTask = null;
-        closeListener(status, rpcProgress, trailers, stopDelivery);
+        closeListener(status, rpcProgress, trailers);
       } else {
         deframerClosedTask =
             new Runnable() {
               @Override
               public void run() {
-                closeListener(status, rpcProgress, trailers, stopDelivery);
+                closeListener(status, rpcProgress, trailers);
               }
             };
-        closeDeframer(stopDelivery);
+        closeDeframer(cancelled);
       }
     }
 
@@ -461,12 +459,9 @@ public void run() {
      * @throws IllegalStateException if the call has not yet been started.
      */
     private void closeListener(
-        Status status, RpcProgress rpcProgress, Metadata trailers, boolean stopDelivery) {
+        Status status, RpcProgress rpcProgress, Metadata trailers) {
       if (!listenerClosed) {
         listenerClosed = true;
-        if (stopDelivery) {
-          statsTraceCtx.clientCancelled(status);
-        }
         statsTraceCtx.streamClosed(status);
         if (getTransportTracer() != null) {
           getTransportTracer().reportStreamClosed(status.isOk());