You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by ea...@apache.org on 2023/02/25 10:26:56 UTC
[dubbo] branch 3.2 updated: remove duplicate flowcontroller listener (#11642)
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 837b9c5515 remove duplicate flowcontroller listener (#11642)
837b9c5515 is described below
commit 837b9c5515936ef16710743b54783f53c0581e4f
Author: 一个不知名的Java靓仔 <cl...@gmail.com>
AuthorDate: Sat Feb 25 18:26:47 2023 +0800
remove duplicate flowcontroller listener (#11642)
* remove duplicate flowcontroller listener
* remove duplicate flowcontroller listener
---
.../protocol/tri/TripleHttp2FrameCodecBuilder.java | 70 ++++++++++++++++++++++
.../rpc/protocol/tri/TripleHttp2Protocol.java | 9 ++-
2 files changed, 74 insertions(+), 5 deletions(-)
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameCodecBuilder.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameCodecBuilder.java
new file mode 100644
index 0000000000..217b8c1496
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameCodecBuilder.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dubbo.rpc.protocol.tri;
+
+import io.netty.handler.codec.http2.DefaultHttp2Connection;
+import io.netty.handler.codec.http2.Http2CodecUtil;
+import io.netty.handler.codec.http2.Http2Connection;
+import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
+import io.netty.handler.codec.http2.Http2LocalFlowController;
+import io.netty.handler.codec.http2.Http2RemoteFlowController;
+import org.apache.dubbo.common.utils.Assert;
+
+import java.util.function.Consumer;
+
+public class TripleHttp2FrameCodecBuilder extends Http2FrameCodecBuilder {
+
+ TripleHttp2FrameCodecBuilder(Http2Connection connection) {
+ connection(connection);
+ }
+
+ public static TripleHttp2FrameCodecBuilder fromConnection(Http2Connection connection) {
+ return new TripleHttp2FrameCodecBuilder(connection);
+ }
+
+ public static TripleHttp2FrameCodecBuilder forClient() {
+ return forClient(Http2CodecUtil.SMALLEST_MAX_CONCURRENT_STREAMS);
+ }
+
+ public static TripleHttp2FrameCodecBuilder forClient(int maxReservedStreams) {
+ return fromConnection(new DefaultHttp2Connection(false, maxReservedStreams));
+ }
+
+ public static TripleHttp2FrameCodecBuilder forServer() {
+ return forServer(Http2CodecUtil.SMALLEST_MAX_CONCURRENT_STREAMS);
+ }
+
+ public static TripleHttp2FrameCodecBuilder forServer(int maxReservedStreams) {
+ return fromConnection(new DefaultHttp2Connection(true, maxReservedStreams));
+ }
+
+ public TripleHttp2FrameCodecBuilder customizeConnection(Consumer<Http2Connection> connectionCustomizer) {
+ Http2Connection connection = this.connection();
+ Assert.notNull(connection, "connection cannot be null.");
+ connectionCustomizer.accept(connection);
+ return this;
+ }
+
+ public TripleHttp2FrameCodecBuilder remoteFlowController(Http2RemoteFlowController remoteFlowController) {
+ return this.customizeConnection((connection) -> connection.remote().flowController(remoteFlowController));
+ }
+
+ public TripleHttp2FrameCodecBuilder localFlowController(Http2LocalFlowController localFlowController) {
+ return this.customizeConnection((connection) -> connection.local().flowController(localFlowController));
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
index d13e4bfcd2..321af9d614 100644
--- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
@@ -43,7 +43,6 @@ import org.apache.dubbo.rpc.protocol.tri.transport.TripleWriteQueue;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http2.Http2FrameCodec;
-import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
import io.netty.handler.codec.http2.Http2FrameLogger;
import io.netty.handler.codec.http2.Http2MultiplexHandler;
import io.netty.handler.codec.http2.Http2Settings;
@@ -113,7 +112,8 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo
} else {
headFilters = Collections.emptyList();
}
- final Http2FrameCodec codec = Http2FrameCodecBuilder.forServer()
+ final Http2FrameCodec codec = TripleHttp2FrameCodecBuilder.forServer()
+ .customizeConnection((connection) -> connection.remote().flowController(new TriHttp2RemoteFlowController(connection, url.getOrDefaultApplicationModel())))
.gracefulShutdownTimeoutMillis(10000)
.initialSettings(new Http2Settings().headerTableSize(
config.getInt(H2_SETTINGS_HEADER_TABLE_SIZE_KEY, DEFAULT_SETTING_HEADER_LIST_SIZE))
@@ -127,7 +127,6 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo
.frameLogger(SERVER_LOGGER)
.build();
ExecutorSupport executorSupport = ExecutorRepository.getInstance(url.getOrDefaultApplicationModel()).getExecutorSupport(url);
- codec.connection().remote().flowController(new TriHttp2RemoteFlowController(codec.connection(), url.getOrDefaultApplicationModel()));
codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter());
TripleWriteQueue writeQueue = new TripleWriteQueue();
final Http2MultiplexHandler handler = new Http2MultiplexHandler(
@@ -152,7 +151,8 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo
@Override
public void configClientPipeline(URL url, ChannelOperator operator, ContextOperator contextOperator) {
- final Http2FrameCodec codec = Http2FrameCodecBuilder.forClient()
+ final Http2FrameCodec codec = TripleHttp2FrameCodecBuilder.forClient()
+ .customizeConnection((connection) -> connection.remote().flowController(new TriHttp2RemoteFlowController(connection, url.getOrDefaultApplicationModel())))
.gracefulShutdownTimeoutMillis(10000)
.initialSettings(new Http2Settings().headerTableSize(
config.getInt(H2_SETTINGS_HEADER_TABLE_SIZE_KEY, DEFAULT_SETTING_HEADER_LIST_SIZE))
@@ -166,7 +166,6 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo
DEFAULT_MAX_HEADER_LIST_SIZE)))
.frameLogger(CLIENT_LOGGER)
.build();
- codec.connection().remote().flowController(new TriHttp2RemoteFlowController(codec.connection(), url.getOrDefaultApplicationModel()));
codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter());
final Http2MultiplexHandler handler = new Http2MultiplexHandler(
new TripleClientHandler(frameworkModel));