[dubbo] branch 3.0 updated: [3.0-Triple] Avoid listener npe on rst (#9916)

[earthchen]

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

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


The following commit(s) were added to refs/heads/3.0 by this push:
     new 2f6bc17d31 [3.0-Triple] Avoid listener npe  on rst (#9916)
2f6bc17d31 is described below

commit 2f6bc17d31eb72f080c56fed0adb998480ef7e86
Author: earthchen <earthchen1...@gmail.com>
AuthorDate: Thu Apr 14 10:34:29 2022 +0800

    [3.0-Triple] Avoid listener npe  on rst (#9916)
    
    * Avoid OnReset npe
    
    * format & Add stream closed judgment
    
    * format
    
    * fix judgment
---
 .../rpc/protocol/tri/stream/TripleServerStream.java  | 20 ++++++++++++--------
 1 file changed, 12 insertions(+), 8 deletions(-)

diff --git 
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java
 
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java
index 207a3c04b9..a9aa76c2aa 100644
--- 
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java
+++ 
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java
@@ -84,11 +84,11 @@ public class TripleServerStream extends AbstractStream 
implements ServerStream {
     private Deframer deframer;
 
     public TripleServerStream(Channel channel,
-        FrameworkModel frameworkModel,
-        Executor executor,
-        PathResolver pathResolver,
-        String acceptEncoding,
-        List<HeaderFilter> filters) {
+                              FrameworkModel frameworkModel,
+                              Executor executor,
+                              PathResolver pathResolver,
+                              String acceptEncoding,
+                              List<HeaderFilter> filters) {
         super(executor, frameworkModel);
         this.channel = channel;
         this.pathResolver = pathResolver;
@@ -349,6 +349,10 @@ public class TripleServerStream extends AbstractStream 
implements ServerStream {
                 return;
             }
 
+            if (endStream) {
+                return;
+            }
+
             DeCompressor deCompressor = DeCompressor.NONE;
             CharSequence messageEncoding = 
headers.get(TripleHeaderEnum.GRPC_ENCODING.getHeader());
             if (null != messageEncoding) {
@@ -389,9 +393,6 @@ public class TripleServerStream extends AbstractStream 
implements ServerStream {
             if (listener == null) {
                 deframer.close();
             }
-            if (endStream) {
-                deframer.close();
-            }
         }
 
 
@@ -417,6 +418,9 @@ public class TripleServerStream extends AbstractStream 
implements ServerStream {
                 // send rst if stream not closed
                 reset(Http2Error.valueOf(errorCode));
             }
+            if (listener == null) {
+                return;
+            }
             executor.execute(() -> {
                 listener.onCancelByRemote(TriRpcStatus.CANCELLED
                     .withDescription("Canceled by client ,errorCode=" + 
errorCode));