support passing detect context to help build netty handlers (#12460)

icodening · web-flow · commit 391d3cdb3581 · 2023-06-07T15:19:06.000+08:00
diff --git a/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/pu/QosDetector.java b/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/pu/QosDetector.java
@@ -37,20 +37,20 @@ public QosDetector(FrameworkModel frameworkModel) {
     @Override
     public Result detect(ChannelBuffer in) {
         if(!QosEnableFlag) {
-            return Result.UNRECOGNIZED;
+            return Result.unrecognized();
         }
         Result h1Res = qosHTTP1Detector.detect(in);
-        if(h1Res.equals(Result.RECOGNIZED)) {
+        if(h1Res.equals(Result.recognized())) {
             return h1Res;
         }
         Result telRes = telnetDetector.detect(in);
-        if(telRes.equals(Result.RECOGNIZED)) {
+        if(telRes.equals(Result.recognized())) {
             return telRes;
         }
-        if(h1Res.equals(Result.NEED_MORE_DATA) || telRes.equals(Result.NEED_MORE_DATA)) {
-            return Result.NEED_MORE_DATA;
+        if(h1Res.equals(Result.needMoreData()) || telRes.equals(Result.needMoreData())) {
+            return Result.needMoreData();
         }
-        return Result.UNRECOGNIZED;
+        return Result.unrecognized();
     }
 
 }
diff --git a/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/pu/QosHTTP1Detector.java b/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/pu/QosHTTP1Detector.java
@@ -27,13 +27,13 @@ private static boolean isHttp(int magic) {
     @Override
     public Result detect(ChannelBuffer in) {
         if (in.readableBytes() < 2) {
-            return Result.NEED_MORE_DATA;
+            return Result.needMoreData();
         }
         final int magic = in.getByte(in.readerIndex());
         // h2 starts with "PR"
         if (isHttp(magic) && in.getByte(in.readerIndex()+1) != 'R' ){
-            return Result.RECOGNIZED;
+            return Result.recognized();
         }
-        return Result.UNRECOGNIZED;
+        return Result.unrecognized();
     }
 }
diff --git a/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/pu/TelnetDetector.java b/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/pu/TelnetDetector.java
@@ -43,20 +43,20 @@ public TelnetDetector(FrameworkModel frameworkModel) {
     @Override
     public Result detect(ChannelBuffer in) {
         if (in.readableBytes() >= MaxSize) {
-            return Result.UNRECOGNIZED;
+            return Result.unrecognized();
         }
         Result resCommand = commandDetect(in);
-        if (resCommand.equals(Result.RECOGNIZED)) {
+        if (resCommand.equals(Result.recognized())) {
             return resCommand;
         }
         Result resAyt = telnetAytDetect(in);
-        if (resAyt.equals(Result.RECOGNIZED)) {
+        if (resAyt.equals(Result.recognized())) {
             return resAyt;
         }
-        if (resAyt.equals(Result.UNRECOGNIZED) && resCommand.equals(Result.UNRECOGNIZED)) {
-            return Result.UNRECOGNIZED;
+        if (resAyt.equals(Result.unrecognized()) && resCommand.equals(Result.unrecognized())) {
+            return Result.unrecognized();
         }
-        return Result.NEED_MORE_DATA;
+        return Result.needMoreData();
     }
 
     private Result commandDetect(ChannelBuffer in) {
@@ -75,26 +75,26 @@ private Result commandDetect(ChannelBuffer in) {
         s = s.trim();
         CommandContext commandContext = TelnetCommandDecoder.decode(s);
         if (frameworkModel.getExtensionLoader(BaseCommand.class).hasExtension(commandContext.getCommandName())) {
-            return Result.RECOGNIZED;
+            return Result.recognized();
         }
-        return Result.UNRECOGNIZED;
+        return Result.unrecognized();
     }
 
     private Result telnetAytDetect(ChannelBuffer in) {
         // detect if remote channel send a telnet ayt command to server
         int prefaceLen = AytPreface.readableBytes();
         int bytesRead = min(in.readableBytes(), prefaceLen);
         if (bytesRead == 0 || !ChannelBuffers.prefixEquals(in, AytPreface, bytesRead)) {
-            return Result.UNRECOGNIZED;
+            return Result.unrecognized();
         }
         if (bytesRead == prefaceLen) {
             // we need to consume preface because it's not a qos command
             // consume and remember to mark, pu server handler reset reader index
             in.readBytes(AytPreface.readableBytes());
             in.markReaderIndex();
-            return Result.RECOGNIZED;
+            return Result.recognized();
         }
-        return Result.NEED_MORE_DATA;
+        return Result.needMoreData();
     }
 
 }
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ProtocolDetector.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ProtocolDetector.java
@@ -19,16 +19,59 @@
 
 import org.apache.dubbo.remoting.buffer.ChannelBuffer;
 
+import java.util.HashMap;
+import java.util.Map;
+
 
 /**
  * Determine incoming bytes belong to the specific protocol.
- *
  */
 public interface ProtocolDetector {
 
     Result detect(ChannelBuffer in);
 
-    enum Result {
+    class Result {
+
+        private final Flag flag;
+
+        private final Map<String, String> detectContext = new HashMap<>(4);
+
+        private Result(Flag flag) {
+            this.flag = flag;
+        }
+
+        public void setAttribute(String key, String value) {
+            this.detectContext.put(key, value);
+        }
+
+        public String getAttribute(String key) {
+            return this.detectContext.get(key);
+        }
+
+        public void removeAttribute(String key) {
+            this.detectContext.remove(key);
+        }
+
+        public Flag flag() {
+            return flag;
+        }
+
+        public static Result recognized(){
+            return new Result(Flag.RECOGNIZED);
+        }
+
+
+        public static Result unrecognized(){
+            return new Result(Flag.UNRECOGNIZED);
+        }
+
+
+        public static Result needMoreData(){
+            return new Result(Flag.NEED_MORE_DATA);
+        }
+    }
+
+    enum Flag {
         RECOGNIZED, UNRECOGNIZED, NEED_MORE_DATA
     }
 }
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/pu/ChannelOperator.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/pu/ChannelOperator.java
@@ -17,9 +17,12 @@
 package org.apache.dubbo.remoting.api.pu;
 
 import org.apache.dubbo.remoting.ChannelHandler;
+import org.apache.dubbo.remoting.api.ProtocolDetector;
 
 import java.util.List;
 
 public interface ChannelOperator {
     void configChannelHandler(List<ChannelHandler> handlerList);
+
+    ProtocolDetector.Result detectResult();
 }
diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConfigOperator.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConfigOperator.java
@@ -24,6 +24,7 @@
 import org.apache.dubbo.remoting.Codec;
 import org.apache.dubbo.remoting.Codec2;
 import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.api.ProtocolDetector;
 import org.apache.dubbo.remoting.api.pu.ChannelHandlerPretender;
 import org.apache.dubbo.remoting.api.pu.ChannelOperator;
 import org.apache.dubbo.remoting.api.pu.DefaultCodec;
@@ -36,6 +37,8 @@ public class NettyConfigOperator implements ChannelOperator {
     private final Channel channel;
     private ChannelHandler handler;
 
+    private ProtocolDetector.Result detectResult;
+
     public NettyConfigOperator(NettyChannel channel, ChannelHandler handler) {
         this.channel = channel;
         this.handler = handler;
@@ -91,6 +94,15 @@ public void configChannelHandler(List<ChannelHandler> handlerList) {
         }
     }
 
+    public void setDetectResult(ProtocolDetector.Result detectResult) {
+        this.detectResult = detectResult;
+    }
+
+    @Override
+    public ProtocolDetector.Result detectResult() {
+        return detectResult;
+    }
+
     private boolean isClientSide(Channel channel) {
         return channel.getUrl().getSide("").equalsIgnoreCase(CommonConstants.CONSUMER);
     }
diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServerHandler.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServerHandler.java
@@ -107,7 +107,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
                 ChannelBuffer buf = new NettyBackedChannelBuffer(in);
                 final ProtocolDetector.Result result = protocol.detector().detect(buf);
                 in.resetReaderIndex();
-                switch (result) {
+                switch (result.flag()) {
                     case UNRECOGNIZED:
                         continue;
                     case RECOGNIZED:
@@ -117,6 +117,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
                         URL localURL = this.urlMapper.getOrDefault(protocolName, url);
                         channel.setUrl(localURL);
                         NettyConfigOperator operator = new NettyConfigOperator(channel, localHandler);
+                        operator.setDetectResult(result);
                         protocol.configServerProtocolHandler(url, operator);
                         ctx.pipeline().remove(this);
                     case NEED_MORE_DATA:
diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/pu/DubboDetector.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/pu/DubboDetector.java
@@ -36,12 +36,12 @@ public Result detect(ChannelBuffer in) {
         int bytesRead = min(in.readableBytes(), prefaceLen);
 
         if (bytesRead ==0 || !ChannelBuffers.prefixEquals(in,  Preface,  bytesRead)) {
-            return Result.UNRECOGNIZED;
+            return Result.unrecognized();
         }
         if (bytesRead == prefaceLen) {
-            return Result.RECOGNIZED;
+            return Result.recognized();
         }
 
-        return Result.NEED_MORE_DATA;
+        return Result.needMoreData();
     }
 }
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/Http2ProtocolDetector.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/Http2ProtocolDetector.java
@@ -36,11 +36,11 @@ public Result detect(ChannelBuffer in) {
 
         // If the input so far doesn't match the preface, break the connection.
         if (bytesRead == 0 || !ChannelBuffers.prefixEquals(in, clientPrefaceString, bytesRead)) {
-            return Result.UNRECOGNIZED;
+            return Result.unrecognized();
         }
         if (bytesRead == prefaceLen) {
-            return Result.RECOGNIZED;
+            return Result.recognized();
         }
-        return Result.NEED_MORE_DATA;
+        return Result.needMoreData();
     }
 }
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/Http2ProtocolDetectorTest.java b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/Http2ProtocolDetectorTest.java
@@ -42,16 +42,16 @@ void testDetect() {
         ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
         ChannelBuffer in = new ByteBufferBackedChannelBuffer(byteBuf.nioBuffer());
         ProtocolDetector.Result result = detector.detect(in);
-        Assertions.assertEquals(result, ProtocolDetector.Result.UNRECOGNIZED);
+        Assertions.assertEquals(result.flag(), ProtocolDetector.Result.unrecognized().flag());
 
         byteBuf.writeBytes(connectionPrefaceBuf);
         result = detector.detect(new ByteBufferBackedChannelBuffer(byteBuf.nioBuffer()));
-        Assertions.assertEquals(result, ProtocolDetector.Result.RECOGNIZED);
+        Assertions.assertEquals(result.flag(), ProtocolDetector.Result.recognized().flag());
 
         byteBuf.clear();
         byteBuf.writeBytes(connectionPrefaceBuf, 0, 1);
         result = detector.detect(new ByteBufferBackedChannelBuffer(byteBuf.nioBuffer()));
-        Assertions.assertEquals(result, ProtocolDetector.Result.NEED_MORE_DATA);
+        Assertions.assertEquals(result.flag(), ProtocolDetector.Result.needMoreData().flag());
 
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -37,20 +37,20 @@ public QosDetector(FrameworkModel frameworkModel) {`
`37`	`37`	`@Override`
`38`	`38`	`public Result detect(ChannelBuffer in) {`
`39`	`39`	`if(!QosEnableFlag) {`
`40`		`- return Result.UNRECOGNIZED;`
	`40`	`+ return Result.unrecognized();`
`41`	`41`	`}`
`42`	`42`	`Result h1Res = qosHTTP1Detector.detect(in);`
`43`		`- if(h1Res.equals(Result.RECOGNIZED)) {`
	`43`	`+ if(h1Res.equals(Result.recognized())) {`
`44`	`44`	`return h1Res;`
`45`	`45`	`}`
`46`	`46`	`Result telRes = telnetDetector.detect(in);`
`47`		`- if(telRes.equals(Result.RECOGNIZED)) {`
	`47`	`+ if(telRes.equals(Result.recognized())) {`
`48`	`48`	`return telRes;`
`49`	`49`	`}`
`50`		`- if(h1Res.equals(Result.NEED_MORE_DATA) \|\| telRes.equals(Result.NEED_MORE_DATA)) {`
`51`		`- return Result.NEED_MORE_DATA;`
	`50`	`+ if(h1Res.equals(Result.needMoreData()) \|\| telRes.equals(Result.needMoreData())) {`
	`51`	`+ return Result.needMoreData();`
`52`	`52`	`}`
`53`		`- return Result.UNRECOGNIZED;`
	`53`	`+ return Result.unrecognized();`
`54`	`54`	`}`
`55`	`55`
`56`	`56`	`}`
Original file line number	Diff line number	Diff line change
`@@ -27,13 +27,13 @@ private static boolean isHttp(int magic) {`
`27`	`27`	`@Override`
`28`	`28`	`public Result detect(ChannelBuffer in) {`
`29`	`29`	`if (in.readableBytes() < 2) {`
`30`		`- return Result.NEED_MORE_DATA;`
	`30`	`+ return Result.needMoreData();`
`31`	`31`	`}`
`32`	`32`	`final int magic = in.getByte(in.readerIndex());`
`33`	`33`	`// h2 starts with "PR"`
`34`	`34`	`if (isHttp(magic) && in.getByte(in.readerIndex()+1) != 'R' ){`
`35`		`- return Result.RECOGNIZED;`
	`35`	`+ return Result.recognized();`
`36`	`36`	`}`
`37`		`- return Result.UNRECOGNIZED;`
	`37`	`+ return Result.unrecognized();`
`38`	`38`	`}`
`39`	`39`	`}`
Original file line number	Diff line number	Diff line change
`@@ -43,20 +43,20 @@ public TelnetDetector(FrameworkModel frameworkModel) {`
`43`	`43`	`@Override`
`44`	`44`	`public Result detect(ChannelBuffer in) {`
`45`	`45`	`if (in.readableBytes() >= MaxSize) {`
`46`		`- return Result.UNRECOGNIZED;`
	`46`	`+ return Result.unrecognized();`
`47`	`47`	`}`
`48`	`48`	`Result resCommand = commandDetect(in);`
`49`		`- if (resCommand.equals(Result.RECOGNIZED)) {`
	`49`	`+ if (resCommand.equals(Result.recognized())) {`
`50`	`50`	`return resCommand;`
`51`	`51`	`}`
`52`	`52`	`Result resAyt = telnetAytDetect(in);`
`53`		`- if (resAyt.equals(Result.RECOGNIZED)) {`
	`53`	`+ if (resAyt.equals(Result.recognized())) {`
`54`	`54`	`return resAyt;`
`55`	`55`	`}`
`56`		`- if (resAyt.equals(Result.UNRECOGNIZED) && resCommand.equals(Result.UNRECOGNIZED)) {`
`57`		`- return Result.UNRECOGNIZED;`
	`56`	`+ if (resAyt.equals(Result.unrecognized()) && resCommand.equals(Result.unrecognized())) {`
	`57`	`+ return Result.unrecognized();`
`58`	`58`	`}`
`59`		`- return Result.NEED_MORE_DATA;`
	`59`	`+ return Result.needMoreData();`
`60`	`60`	`}`
`61`	`61`
`62`	`62`	`private Result commandDetect(ChannelBuffer in) {`
`@@ -75,26 +75,26 @@ private Result commandDetect(ChannelBuffer in) {`
`75`	`75`	`s = s.trim();`
`76`	`76`	`CommandContext commandContext = TelnetCommandDecoder.decode(s);`
`77`	`77`	`if (frameworkModel.getExtensionLoader(BaseCommand.class).hasExtension(commandContext.getCommandName())) {`
`78`		`- return Result.RECOGNIZED;`
	`78`	`+ return Result.recognized();`
`79`	`79`	`}`
`80`		`- return Result.UNRECOGNIZED;`
	`80`	`+ return Result.unrecognized();`
`81`	`81`	`}`
`82`	`82`
`83`	`83`	`private Result telnetAytDetect(ChannelBuffer in) {`
`84`	`84`	`// detect if remote channel send a telnet ayt command to server`
`85`	`85`	`int prefaceLen = AytPreface.readableBytes();`
`86`	`86`	`int bytesRead = min(in.readableBytes(), prefaceLen);`
`87`	`87`	`if (bytesRead == 0 \|\| !ChannelBuffers.prefixEquals(in, AytPreface, bytesRead)) {`
`88`		`- return Result.UNRECOGNIZED;`
	`88`	`+ return Result.unrecognized();`
`89`	`89`	`}`
`90`	`90`	`if (bytesRead == prefaceLen) {`
`91`	`91`	`// we need to consume preface because it's not a qos command`
`92`	`92`	`// consume and remember to mark, pu server handler reset reader index`
`93`	`93`	`in.readBytes(AytPreface.readableBytes());`
`94`	`94`	`in.markReaderIndex();`
`95`		`- return Result.RECOGNIZED;`
	`95`	`+ return Result.recognized();`
`96`	`96`	`}`
`97`		`- return Result.NEED_MORE_DATA;`
	`97`	`+ return Result.needMoreData();`
`98`	`98`	`}`
`99`	`99`
`100`	`100`	`}`
Original file line number	Diff line number	Diff line change
`@@ -36,12 +36,12 @@ public Result detect(ChannelBuffer in) {`
`36`	`36`	`int bytesRead = min(in.readableBytes(), prefaceLen);`
`37`	`37`
`38`	`38`	`if (bytesRead ==0 \|\| !ChannelBuffers.prefixEquals(in, Preface, bytesRead)) {`
`39`		`- return Result.UNRECOGNIZED;`
	`39`	`+ return Result.unrecognized();`
`40`	`40`	`}`
`41`	`41`	`if (bytesRead == prefaceLen) {`
`42`		`- return Result.RECOGNIZED;`
	`42`	`+ return Result.recognized();`
`43`	`43`	`}`
`44`	`44`
`45`		`- return Result.NEED_MORE_DATA;`
	`45`	`+ return Result.needMoreData();`
`46`	`46`	`}`
`47`	`47`	`}`
Original file line number	Diff line number	Diff line change
`@@ -36,11 +36,11 @@ public Result detect(ChannelBuffer in) {`
`36`	`36`
`37`	`37`	`// If the input so far doesn't match the preface, break the connection.`
`38`	`38`	`if (bytesRead == 0 \|\| !ChannelBuffers.prefixEquals(in, clientPrefaceString, bytesRead)) {`
`39`		`- return Result.UNRECOGNIZED;`
	`39`	`+ return Result.unrecognized();`
`40`	`40`	`}`
`41`	`41`	`if (bytesRead == prefaceLen) {`
`42`		`- return Result.RECOGNIZED;`
	`42`	`+ return Result.recognized();`
`43`	`43`	`}`
`44`		`- return Result.NEED_MORE_DATA;`
	`44`	`+ return Result.needMoreData();`
`45`	`45`	`}`
`46`	`46`	`}`