core,netty: correctly count streams that ended due to client cancel (#3683)

Only bump the counter from AbstractServerStream.TransportState, and hole punch
from AbstractServerStream to TransportState when the application calls close.

1 files changed, 41 insertions, 2 deletions

diff --git a/testing/src/main/java/io/grpc/internal/testing/AbstractTransportTest.java b/testing/src/main/java/io/grpc/internal/testing/AbstractTransportTest.java
index b08d6e07a..d089d89e5 100644
--- a/testing/src/main/java/io/grpc/internal/testing/AbstractTransportTest.java
+++ b/testing/src/main/java/io/grpc/internal/testing/AbstractTransportTest.java
@@ -1432,7 +1432,7 @@ public abstract class AbstractTransportTest {
   }
 
   @Test
-  public void transportTracer_streamEnded_ok() throws Exception {
+  public void transportTracer_server_streamEnded_ok() throws Exception {
     server.start(serverListener);
     client = newClientTransport(server);
     runIfNotNull(client.start(mock(ManagedClientTransport.Listener.class)));
@@ -1462,7 +1462,7 @@ public abstract class AbstractTransportTest {
   }
 
   @Test
-  public void transportTracer_streamEnded_nonOk() throws Exception {
+  public void transportTracer_server_streamEnded_nonOk() throws Exception {
     server.start(serverListener);
     client = newClientTransport(server);
     runIfNotNull(client.start(mock(ManagedClientTransport.Listener.class)));
@@ -1491,6 +1491,45 @@ public abstract class AbstractTransportTest {
   }
 
   @Test
+  public void transportTracer_client_streamEnded_nonOk() throws Exception {
+    server.start(serverListener);
+    client = newClientTransport(server);
+    runIfNotNull(client.start(mockClientTransportListener));
+    ClientStream clientStream = client.newStream(methodDescriptor, new Metadata(), callOptions);
+    ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
+    clientStream.start(clientStreamListener);
+    MockServerTransportListener serverTransportListener =
+        serverListener.takeListenerOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    StreamCreation serverStreamCreation =
+        serverTransportListener.takeStreamOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    if (!haveTransportTracer()) {
+      return;
+    }
+
+    TransportTracer.Stats serverBefore =
+        serverTransportListener.transport.getTransportStats().get();
+    assertEquals(0, serverBefore.streamsFailed);
+    assertEquals(0, serverBefore.streamsSucceeded);
+    // TODO(zpencer): uncomment when integrated with client transport
+    // TransportTracer.Stats clientBefore = client.getTransportStats().get();
+    // assertEquals(0, clientBefore.streamsFailed);
+    // assertEquals(0, clientBefore.streamsSucceeded);
+
+    clientStream.cancel(Status.UNKNOWN);
+    // do not validate stats until close() has been called on server
+    assertNotNull(serverStreamCreation.listener.status.get(TIMEOUT_MS, TimeUnit.MILLISECONDS));
+
+    TransportTracer.Stats serverAfter =
+        serverTransportListener.transport.getTransportStats().get();
+    assertEquals(1, serverAfter.streamsFailed);
+    assertEquals(0, serverAfter.streamsSucceeded);
+    // TODO(zpencer): uncomment when integrated with client transport
+    // TransportTracer.Stats clientAfter = client.getTransportStats().get();
+    // assertEquals(1, clientAfter.streamsFailed);
+    // assertEquals(0, clientAfter.streamsSucceeded);
+  }
+
+  @Test
   public void transportTracer_receive_msg() throws Exception {
     server.start(serverListener);
     client = newClientTransport(server);