Close stream when channel was inactive (#14643)

refine

Author: finefuture

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java

@@ -136,6 +136,14 @@ public void onComplete(
         }
     }
 
+    @Override
+    public void onClose() {
+        if (done) {
+            return;
+        }
+        onCancelByRemote(TriRpcStatus.CANCELLED);
+    }
+
     @Override
     public void onStart() {
         listener.onStart(TripleClientCall.this);

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/ClientStream.java

@@ -56,6 +56,8 @@ default void onComplete(
                 boolean isReturnTriException) {
             onComplete(status, attachments);
         }
+
+        void onClose();
     }
 
     /**

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java

@@ -472,5 +472,10 @@ public void cancelByRemote(long errorCode) {
                 finishProcess(transportError, null, false);
             });
         }
+
+        @Override
+        public void onClose() {
+            executor.execute(listener::onClose);
+        }
     }
 }

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java

@@ -483,6 +483,14 @@ public void cancelByRemote(long errorCode) {
             executor.execute(() -> listener.onCancelByRemote(
                     TriRpcStatus.CANCELLED.withDescription("Canceled by client ,errorCode=" + errorCode)));
         }
+
+        @Override
+        public void onClose() {
+            if (listener == null) {
+                return;
+            }
+            executor.execute(() -> listener.onCancelByRemote(TriRpcStatus.CANCELLED));
+        }
     }
 
     private static class ServerDecoderListener implements TriDecoder.Listener {

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/H2TransportListener.java

@@ -42,4 +42,6 @@ public interface H2TransportListener {
     void onData(ByteBuf data, boolean endStream);
 
     void cancelByRemote(long errorCode);
+
+    void onClose();
 }

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleHttp2ClientResponseHandler.java

@@ -80,6 +80,7 @@ private void onResetRead(ChannelHandlerContext ctx, Http2ResetFrame resetFrame)
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) {
+        transportListener.onClose();
         ctx.close();
     }
 

dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/stream/MockClientStreamListener.java

@@ -36,6 +36,9 @@ public void onComplete(TriRpcStatus status, Map<String, Object> attachments) {
         this.status = status;
     }
 
+    @Override
+    public void onClose() {}
+
     @Override
     public void onMessage(byte[] message, boolean isNeedReturnException) {
         this.message = message;

dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/transport/AbstractH2TransportListenerTest.java

@@ -40,6 +40,9 @@ public void onData(ByteBuf data, boolean endStream) {}
 
             @Override
             public void cancelByRemote(long errorCode) {}
+
+            @Override
+            public void onClose() {}
         };
         DefaultHttp2Headers headers = new DefaultHttp2Headers();
         headers.scheme(HTTPS.name()).path("/foo.bar").method(HttpMethod.POST.asciiName());