[dubbo] branch 3.2 updated: reset all active stream on connection close (#12451)

[earthchen]

This is an automated email from the ASF dual-hosted git repository.

earthchen pushed a commit to branch 3.2
in repository https://gitbox.apache.org/repos/asf/dubbo.git


The following commit(s) were added to refs/heads/3.2 by this push:
     new eba976fec6 reset all active stream on connection close (#12451)
eba976fec6 is described below

commit eba976fec63eb1ccbedd5751162e2f2a888391df
Author: icodening <cleverb...@gmail.com>
AuthorDate: Wed Jun 7 11:03:26 2023 +0800

    reset all active stream on connection close (#12451)
    
    * reset all active stream on connection close
    
    * reset all active stream on connection close
---
 .../rpc/protocol/tri/call/BiStreamServerCallListener.java   |  2 +-
 .../tri/transport/TripleServerConnectionHandler.java        | 13 +++++++++++++
 2 files changed, 14 insertions(+), 1 deletion(-)

diff --git 
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/BiStreamServerCallListener.java
 
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/BiStreamServerCallListener.java
index 8a7f169e9c..754ef77f3c 100644
--- 
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/BiStreamServerCallListener.java
+++ 
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/BiStreamServerCallListener.java
@@ -52,7 +52,7 @@ public class BiStreamServerCallListener extends 
AbstractServerCallListener {
 
     @Override
     public void onCancel(TriRpcStatus status) {
-        responseObserver.onError(status.asException());
+        requestObserver.onError(status.asException());
     }
 
 
diff --git 
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleServerConnectionHandler.java
 
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleServerConnectionHandler.java
index 7aefff58fc..9703cb287b 100644
--- 
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleServerConnectionHandler.java
+++ 
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleServerConnectionHandler.java
@@ -17,6 +17,8 @@
 
 package org.apache.dubbo.rpc.protocol.tri.transport;
 
+import io.netty.handler.codec.http2.DefaultHttp2ResetFrame;
+import io.netty.handler.codec.http2.Http2Error;
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
 
@@ -67,6 +69,17 @@ public class TripleServerConnectionHandler extends 
Http2ChannelDuplexHandler {
         }
     }
 
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        super.channelInactive(ctx);
+        //reset all active stream on connection close
+        forEachActiveStream(stream -> {
+            DefaultHttp2ResetFrame resetFrame = new 
DefaultHttp2ResetFrame(Http2Error.NO_ERROR).stream(stream);
+            ctx.fireChannelRead(resetFrame);
+            return true;
+        });
+    }
+
     private boolean isQuiteException(Throwable t) {
         if (QUIET_EXCEPTIONS_CLASS.contains(t.getClass())) {
             return true;