Merge branch 'refs/heads/apache-3.2' into 3.2.16-release

apache · Sep 7, 2024 · c2894d1 · c2894d1
2 parents 4a74897 + 08c74b7
commit c2894d1
Show file tree

Hide file tree

Showing 8 changed files with 32 additions and 0 deletions.
diff --git a/...bbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java b/...bbo-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);

diff --git a/...dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/ClientStream.java b/...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();
     }
 
     /**

diff --git a/...rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java b/...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);
+        }
     }
 }
diff --git a/...rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java b/...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 {

diff --git a/...triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/H2TransportListener.java b/...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();
 }
diff --git a/...in/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleHttp2ClientResponseHandler.java b/...in/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();
     }
 

diff --git a/...iple/src/test/java/org/apache/dubbo/rpc/protocol/tri/stream/MockClientStreamListener.java b/...iple/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;

diff --git a/...est/java/org/apache/dubbo/rpc/protocol/tri/transport/AbstractH2TransportListenerTest.java b/...est/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());