You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by li...@apache.org on 2021/01/26 09:26:29 UTC
[dubbo] branch 3.0 updated: Triple support (#7087)
This is an automated email from the ASF dual-hosted git repository.
liujun 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 1ab487e Triple support (#7087)
1ab487e is described below
commit 1ab487e487c9117ecbcae292710471b550934f84
Author: panxiaojun233 <43...@users.noreply.github.com>
AuthorDate: Tue Jan 26 17:25:11 2021 +0800
Triple support (#7087)
---
.../dubbo/common/constants/CommonConstants.java | 2 +
.../java/org/apache/dubbo/config/Constants.java | 2 +
dubbo-dependencies-bom/pom.xml | 4 +-
dubbo-distribution/dubbo-all/pom.xml | 16 ++
dubbo-distribution/dubbo-bom/pom.xml | 5 +
.../client/migration/MigrationRuleListener.java | 3 +-
dubbo-remoting/dubbo-remoting-api/pom.xml | 4 +
.../org/apache/dubbo/remoting/api/Connection.java | 318 +++++++++++++++++++++
.../dubbo/remoting/api/ConnectionHandler.java | 176 ++++++++++++
.../dubbo/remoting/api/ConnectionManager.java | 35 +++
.../dubbo/remoting/api/Http2ProtocolDetector.java | 48 ++++
.../dubbo/remoting/api/Http2WireProtocol.java | 39 +++
.../api/MultiplexProtocolConnectionManager.java | 47 +++
.../dubbo/remoting/api/NettyEventLoopFactory.java | 66 +++++
.../dubbo/remoting/api/PortUnificationServer.java | 243 ++++++++++++++++
.../remoting/api/PortUnificationServerHandler.java | 119 ++++++++
.../dubbo/remoting/api/ProtocolDetector.java | 38 +++
.../api/SingleProtocolConnectionManager.java | 60 ++++
.../apache/dubbo/remoting/api/WireProtocol.java | 37 +++
.../exchange/PortUnificationExchanger.java | 56 ++++
.../remoting/exchange/support/DefaultFuture2.java | 295 +++++++++++++++++++
...org.apache.dubbo.remoting.api.ConnectionManager | 2 +
.../dubbo/remoting/netty4/ConnectionTest.java | 62 ++++
.../dubbo/remoting/netty4/EmptyProtocol.java | 48 ++++
.../dubbo/remoting/netty4/NettyServerTest.java | 36 +++
.../org.apache.dubbo.remoting.netty4.WireProtocol | 1 +
.../java/org/apache/dubbo/rpc/RpcInvocation.java | 33 ++-
dubbo-rpc/dubbo-rpc-triple/pom.xml | 74 +++++
.../dubbo/rpc/protocol/tri/AbstractStream.java | 179 ++++++++++++
.../dubbo/rpc/protocol/tri/ClassLoadUtil.java | 32 +++
.../dubbo/rpc/protocol/tri/ClientStream.java | 224 +++++++++++++++
.../dubbo/rpc/protocol/tri/GracefulShutdown.java | 81 ++++++
.../dubbo/rpc/protocol/tri/GrpcDataDecoder.java | 81 ++++++
.../apache/dubbo/rpc/protocol/tri/GrpcStatus.java | 87 ++++++
.../dubbo/rpc/protocol/tri/PathResolver.java | 36 +++
.../dubbo/rpc/protocol/tri/ServerStream.java | 279 ++++++++++++++++++
.../protocol/tri/SingleProtobufSerialization.java | 122 ++++++++
.../org/apache/dubbo/rpc/protocol/tri/Stream.java | 38 +++
.../rpc/protocol/tri/TripleClientHandler.java | 68 +++++
.../protocol/tri/TripleClientInboundHandler.java | 35 +++
.../dubbo/rpc/protocol/tri/TripleConstant.java | 36 +++
.../tri/TripleHttp2ClientResponseHandler.java | 85 ++++++
.../tri/TripleHttp2FrameServerHandler.java | 158 ++++++++++
.../rpc/protocol/tri/TripleHttp2Protocol.java | 68 +++++
.../dubbo/rpc/protocol/tri/TripleInvoker.java | 182 ++++++++++++
.../dubbo/rpc/protocol/tri/TriplePathResolver.java | 54 ++++
.../dubbo/rpc/protocol/tri/TripleProtocol.java | 107 +++++++
.../dubbo/rpc/protocol/tri/TripleRpcException.java | 35 +++
.../tri/TripleServerConnectionHandler.java | 68 +++++
.../protocol/tri/TripleServerInboundHandler.java | 35 +++
.../rpc/protocol/tri/TripleServerInitializer.java | 36 +++
.../apache/dubbo/rpc/protocol/tri/TripleUtil.java | 309 ++++++++++++++++++++
.../src/main/proto/triple_wrapper.proto | 17 ++
.../org.apache.dubbo.remoting.api.WireProtocol | 1 +
.../dubbo/internal/org.apache.dubbo.rpc.Protocol | 1 +
.../org.apache.dubbo.rpc.protocol.tri.PathResolver | 1 +
.../dubbo/rpc/protocol/tri/TripleUtilTest.java | 34 +++
.../org.apache.dubbo.remoting.telnet.TelnetHandler | 1 +
.../dubbo-rpc-triple/src/test/resources/log4j.xml | 38 +++
dubbo-rpc/pom.xml | 1 +
.../apache/dubbo/common/serialize/Constants.java | 1 +
.../serialize/DefaultMultipleSerialization.java | 56 ++++
.../common/serialize/MultipleSerialization.java | 36 +++
...he.dubbo.common.serialize.MultipleSerialization | 1 +
64 files changed, 4475 insertions(+), 17 deletions(-)
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
index 1145f9e..268c313 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
@@ -27,6 +27,8 @@ import java.util.regex.Pattern;
public interface CommonConstants {
String DUBBO = "dubbo";
+ String TRIPLE= "tri";
+
String PROVIDER = "provider";
String CONSUMER = "consumer";
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/config/Constants.java b/dubbo-common/src/main/java/org/apache/dubbo/config/Constants.java
index 5af0de4..f8fed84 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/config/Constants.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/config/Constants.java
@@ -115,4 +115,6 @@ public interface Constants {
String ZOOKEEPER_PROTOCOL = "zookeeper";
String REGISTER_KEY = "register";
+
+ String MULTI_SERIALIZATION_KEY = "serialize.multiple";
}
diff --git a/dubbo-dependencies-bom/pom.xml b/dubbo-dependencies-bom/pom.xml
index f409ae8..e6b67c1 100644
--- a/dubbo-dependencies-bom/pom.xml
+++ b/dubbo-dependencies-bom/pom.xml
@@ -92,7 +92,7 @@
<spring_version>4.3.16.RELEASE</spring_version>
<javassist_version>3.20.0-GA</javassist_version>
<netty_version>3.2.5.Final</netty_version>
- <netty4_version>4.1.51.Final</netty4_version>
+ <netty4_version>4.1.56.Final</netty4_version>
<mina_version>1.1.7</mina_version>
<grizzly_version>2.4.4</grizzly_version>
<httpclient_version>4.5.3</httpclient_version>
@@ -109,7 +109,7 @@
<cxf_version>3.1.15</cxf_version>
<thrift_version>0.12.0</thrift_version>
<hessian_version>4.0.38</hessian_version>
- <protobuf-java_version>3.6.0</protobuf-java_version>
+ <protobuf-java_version>3.11.0</protobuf-java_version>
<javax_annotation-api_version>1.3.2</javax_annotation-api_version>
<servlet_version>3.1.0</servlet_version>
<jetty_version>9.4.11.v20180605</jetty_version>
diff --git a/dubbo-distribution/dubbo-all/pom.xml b/dubbo-distribution/dubbo-all/pom.xml
index 6b009b4..5729680 100644
--- a/dubbo-distribution/dubbo-all/pom.xml
+++ b/dubbo-distribution/dubbo-all/pom.xml
@@ -124,6 +124,13 @@
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
+ <artifactId>dubbo-rpc-triple</artifactId>
+ <version>${project.version}</version>
+ <scope>compile</scope>
+ <optional>true</optional>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-registry-api</artifactId>
<version>${project.version}</version>
<scope>compile</scope>
@@ -372,6 +379,7 @@
<include>org.apache.dubbo:dubbo-rpc-injvm</include>
<include>org.apache.dubbo:dubbo-rpc-rest</include>
<include>org.apache.dubbo:dubbo-rpc-grpc</include>
+ <include>org.apache.dubbo:dubbo-rpc-triple</include>
<include>org.apache.dubbo:dubbo-cluster</include>
<include>org.apache.dubbo:dubbo-registry-api</include>
<include>org.apache.dubbo:dubbo-registry-default</include>
@@ -714,6 +722,14 @@
</resource>
</transformer>
+ <!-- @since 3.0.0 -->
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
+ <resource>
+ META-INF/dubbo/internal/org.apache.dubbo.remoting.api.WireProtocol
+ </resource>
+ </transformer>
+
</transformers>
<filters>
<filter>
diff --git a/dubbo-distribution/dubbo-bom/pom.xml b/dubbo-distribution/dubbo-bom/pom.xml
index 5826d10..1c1c30d 100644
--- a/dubbo-distribution/dubbo-bom/pom.xml
+++ b/dubbo-distribution/dubbo-bom/pom.xml
@@ -180,6 +180,11 @@
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
+ <artifactId>dubbo-serialization-protobuf</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-compatible</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleListener.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleListener.java
index 45a3156..4db36d3 100644
--- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleListener.java
+++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleListener.java
@@ -141,7 +141,8 @@ public class MigrationRuleListener implements RegistryProtocolListener, Configur
@Override
public void onDestroy() {
- configuration.removeListener(RULE_KEY, this);
+ if (configuration != null)
+ configuration.removeListener(RULE_KEY, this);
}
/**
diff --git a/dubbo-remoting/dubbo-remoting-api/pom.xml b/dubbo-remoting/dubbo-remoting-api/pom.xml
index 69ed094..7bc17ae 100644
--- a/dubbo-remoting/dubbo-remoting-api/pom.xml
+++ b/dubbo-remoting/dubbo-remoting-api/pom.xml
@@ -41,6 +41,10 @@
<version>${project.parent.version}</version>
</dependency>
<dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-serialization-hessian2</artifactId>
<version>${project.parent.version}</version>
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java
new file mode 100644
index 0000000..1eb53e9
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java
@@ -0,0 +1,318 @@
+/*
+ *
+ * * 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.remoting.api;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.Version;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ExecutorUtil;
+import org.apache.dubbo.common.utils.NamedThreadFactory;
+import org.apache.dubbo.common.utils.NetUtils;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.RemotingException;
+import org.apache.dubbo.remoting.utils.UrlUtils;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.timeout.IdleStateHandler;
+import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.AttributeKey;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.ReferenceCounted;
+import io.netty.util.Timer;
+import io.netty.util.concurrent.DefaultPromise;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GlobalEventExecutor;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import io.netty.util.concurrent.Promise;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_CLIENT_THREADPOOL;
+import static org.apache.dubbo.common.constants.CommonConstants.THREADPOOL_KEY;
+import static org.apache.dubbo.remoting.api.NettyEventLoopFactory.socketChannelClass;
+
+public class Connection extends AbstractReferenceCounted implements ReferenceCounted {
+
+ public static final Timer TIMER = new HashedWheelTimer(
+ new NamedThreadFactory("dubbo-network-timer", true), 30, TimeUnit.MILLISECONDS);
+ public static final AttributeKey<Connection> CONNECTION = AttributeKey.valueOf("connection");
+ private static final Logger logger = LoggerFactory.getLogger(Connection.class);
+ private final URL url;
+ private final Bootstrap bootstrap;
+ private final int connectTimeout;
+ private final WireProtocol protocol;
+ private final Promise<Void> closeFuture;
+ private final InetSocketAddress remote;
+ private final AtomicReference<ConnectionStatus> status;
+ private volatile Channel channel;
+ private volatile Future<Channel> connectFuture;
+ private int retryAttempts = 1;
+ private long lastReconnectTime;
+
+ public Connection(URL url) {
+ url = ExecutorUtil.setThreadName(url, "DubboClientHandler");
+ url = url.addParameterIfAbsent(THREADPOOL_KEY, DEFAULT_CLIENT_THREADPOOL);
+ this.url = url;
+ this.status = new AtomicReference<>(ConnectionStatus.DISCONNECTED);
+ this.protocol = ExtensionLoader.getExtensionLoader(WireProtocol.class).getExtension(url.getProtocol());
+ this.connectTimeout = url.getPositiveParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT);
+ this.closeFuture = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
+ this.remote = getConnectAddress();
+ this.bootstrap = open();
+ }
+
+ public static Connection getConnectionFromChannel(Channel channel) {
+ return channel.attr(CONNECTION).get();
+ }
+
+ public Promise<Void> getCloseFuture() {
+ return closeFuture;
+ }
+
+
+ public void init() throws RemotingException {
+ concurrentConnect();
+ }
+
+ public Bootstrap open() {
+ final Bootstrap bootstrap = new Bootstrap();
+ bootstrap.group(NettyEventLoopFactory.NIO_EVENT_LOOP_GROUP)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
+ .channel(socketChannelClass());
+
+ bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Math.max(3000, connectTimeout));
+ final ConnectionHandler connectionHandler = new ConnectionHandler(bootstrap, TIMER);
+ bootstrap.handler(new ChannelInitializer<SocketChannel>() {
+
+ @Override
+ protected void initChannel(SocketChannel ch) {
+ ch.attr(CONNECTION).set(Connection.this);
+ Connection.this.channel = ch;
+ int heartbeatInterval = UrlUtils.getHeartbeat(getUrl());
+ // TODO support SSL
+ final ChannelPipeline p = ch.pipeline();//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
+ p.addLast("client-idle-handler", new IdleStateHandler(heartbeatInterval, 0, 0, MILLISECONDS));
+ p.addLast(connectionHandler);
+ // TODO support ssl
+ protocol.configClientPipeline(p, null);
+ // TODO support Socks5
+ }
+ });
+ return bootstrap;
+ }
+
+ public Channel getChannel() {
+ return channel;
+ }
+
+ @Override
+ public String toString() {
+ return getRemote() + ",channel=" + channel + ", status=" + status;
+ }
+
+ public void onIdle() {
+ if (logger.isDebugEnabled()) {
+ logger.debug(String.format("Connection:%s disconnected cause idle", this));
+ }
+ setStatus(ConnectionStatus.DISCONNECTED);
+ this.channel = null;
+ }
+
+ public boolean onDisConnected() {
+ if (logger.isDebugEnabled()) {
+ logger.debug(String.format("Connection:%s disconnected ", this));
+ }
+ return setStatus(ConnectionStatus.DISCONNECTED);
+ }
+
+ public void onConnected(Channel channel) {
+ setStatus(ConnectionStatus.CONNECTED);
+ this.channel = channel;
+ channel.attr(CONNECTION).set(this);
+ if (logger.isDebugEnabled()) {
+ logger.debug(String.format("Connection:%s connected ", this));
+ }
+ }
+
+ public boolean isAvailable() {
+ return ConnectionStatus.CONNECTED == getStatus() && channel != null && channel.isActive();
+ }
+
+ public ConnectionStatus getStatus() {
+ return status.get();
+ }
+
+ public boolean setStatus(ConnectionStatus status) {
+ return this.status.getAndSet(status) != status;
+ }
+
+ public boolean isClosed() {
+ return getStatus() == ConnectionStatus.CLOSED;
+ }
+
+ public void close() {
+ setStatus(ConnectionStatus.CLOSED);
+ if (channel != null) {
+ channel.close();
+ }
+ }
+
+ public ChannelFuture write(Object request) throws RemotingException {
+ if (channel == null || !channel.isActive()) {
+ throw new RemotingException(null, null, "Failed to send request " + request + ", cause: The channel to " + remote + " is closed!");
+ }
+
+ return getChannel().writeAndFlush(request);
+ }
+
+ public InetSocketAddress getRemote() {
+ return remote;
+ }
+
+ @Override
+ protected void deallocate() {
+ setStatus(ConnectionStatus.CLOSED);
+ if (channel != null) {
+ channel.close();
+ }
+ closeFuture.setSuccess(null);
+ }
+
+ @Override
+ public ReferenceCounted touch(Object hint) {
+ return this;
+ }
+
+ private InetSocketAddress getConnectAddress() {
+ return new InetSocketAddress(NetUtils.filterLocalHost(getUrl().getHost()), getUrl().getPort());
+ }
+
+ protected Future<Channel> connectAsync() {
+ final Promise<Channel> promise = ImmediateEventExecutor.INSTANCE.newPromise();
+ ChannelFuture future = bootstrap.connect(getRemote());
+ final ChannelFutureListener listener = new ChannelFutureListener() {
+ @Override
+ public void operationComplete(ChannelFuture future) {
+ if (bootstrap.config().group().isShuttingDown()) {
+ promise.tryFailure(new IllegalStateException("Client is shutdown"));
+ return;
+ }
+ if (future.isSuccess()) {
+ if (logger.isInfoEnabled()) {
+ logger.info("Succeed connect to server " + future.channel().remoteAddress() + " from " + getClass().getSimpleName() + " "
+ + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion()
+ + ", channel is " + future.channel());
+ }
+ bootstrap.config().group().execute(() -> promise.setSuccess(future.channel()));
+ } else {
+ bootstrap.config().group().execute(() -> {
+ final RemotingException cause = new RemotingException(null, getRemote(), "client(url: " + getUrl() + ") failed to connect to server "
+ + future.channel().remoteAddress() + ", error message is:" + future.cause().getMessage(), future.cause());
+ promise.tryFailure(cause);
+ });
+ }
+ Connection.this.connectFuture = null;
+ }
+ };
+ future.addListener(listener);
+ return promise;
+ }
+
+
+ private void concurrentConnect() throws RemotingException {
+ if (channel != null && channel.isActive()) {
+ return;
+ }
+ synchronized (this) {
+ if (channel != null && channel.isActive()) {
+ return;
+ }
+ if (connectFuture != null) {
+ connectFuture.awaitUninterruptibly(getConnectTimeout());
+ } else {
+ int backoff = 2 << retryAttempts;
+ if (System.currentTimeMillis() - lastReconnectTime < backoff) {
+ return;
+ }
+ connectWithGuard();
+ }
+ }
+ }
+
+ protected synchronized void connectWithGuard() throws RemotingException {
+ this.lastReconnectTime = System.currentTimeMillis();
+ this.retryAttempts = Math.min(12, retryAttempts + 1);
+ long start = System.currentTimeMillis();
+ final Future<Channel> connectFuture = connectAsync();
+ this.connectFuture = connectFuture;
+ connectFuture.awaitUninterruptibly(getConnectTimeout());
+ if (!connectFuture.isSuccess()) {
+ if (connectFuture.isDone()) {
+ throw new RemotingException(null, getRemote(), "client(url: " + getUrl() + ") failed to connect to server .error message is:" + connectFuture.cause().getMessage(),
+ connectFuture.cause());
+ } else {
+ throw new RemotingException(null, getRemote(), "client(url: " + getUrl() + ") failed to connect to server. client-side timeout "
+ + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms) from netty client "
+ + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion());
+ }
+ } else {
+ this.retryAttempts = 0;
+ }
+ }
+
+
+ /**
+ * get url.
+ *
+ * @return url
+ */
+ public URL getUrl() {
+ return url;
+ }
+
+ private int getConnectTimeout() {
+ return connectTimeout;
+ }
+
+ public enum ConnectionStatus {
+ DISCONNECTED,
+ CONNECTING,
+ CONNECTED,
+ READ_ONLY,
+ CLOSED
+ }
+
+}
+
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java
new file mode 100644
index 0000000..dc42c05
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java
@@ -0,0 +1,176 @@
+/*
+ *
+ * * 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.remoting.api;
+
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.util.AttributeKey;
+import io.netty.util.Timer;
+
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+@ChannelHandler.Sharable
+public class ConnectionHandler extends ChannelInboundHandlerAdapter {
+ private static final Logger log = LoggerFactory.getLogger(ConnectionHandler.class);
+
+ private static final int MIN_FAST_RECONNECT_INTERVAL = 4000;
+ private static final int BACKOFF_CAP = 12;
+ private static final AttributeKey<Boolean> GO_AWAY_KEY = AttributeKey.valueOf("dubbo_channel_goaway");
+ private final Timer timer;
+ private final Bootstrap bootstrap;
+ private final Semaphore permit = new Semaphore(1);
+ private volatile long lastReconnect;
+
+ public ConnectionHandler(Bootstrap bootstrap, Timer timer) {
+ this.bootstrap = bootstrap;
+ this.timer = timer;
+ }
+
+ public void onGoAway(Channel channel) {
+ channel.attr(GO_AWAY_KEY).set(true);
+ final Connection connection = Connection.getConnectionFromChannel(channel);
+ if (connection != null) {
+ connection.onIdle();
+ }
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ ctx.fireChannelActive();
+ final Connection connection = Connection.getConnectionFromChannel(ctx.channel());
+ if (connection != null) {
+ connection.onConnected(ctx.channel());
+ }
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ log.warn(String.format("Channel error:%s", ctx.channel()), cause);
+ ctx.close();
+ }
+
+ public boolean shouldFastReconnect() {
+ final long period = System.currentTimeMillis() - lastReconnect;
+ return period > MIN_FAST_RECONNECT_INTERVAL;
+ }
+
+
+ @Override
+ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+ super.userEventTriggered(ctx, evt);
+ if (evt instanceof IdleStateEvent) {
+ final Connection connection = Connection.getConnectionFromChannel(ctx.channel());
+ if (connection != null) {
+ connection.onIdle();
+ ctx.close();
+ }
+ }
+ }
+
+ private boolean isGoAway(Channel channel) {
+ return Boolean.TRUE.equals(channel.attr(GO_AWAY_KEY).get());
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ // Reconnect event will be triggered by Connection.init();
+ if (isGoAway(ctx.channel())) {
+ return;
+ }
+ Connection connection = Connection.getConnectionFromChannel(ctx.channel());
+ if (connection != null) {
+ if (!connection.isClosed() && connection.onDisConnected()) {
+ if (shouldFastReconnect()) {
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Connection %s inactive, schedule fast reconnect", connection));
+ }
+ reconnect(connection, 1);
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Connection %s inactive, schedule normal reconnect", connection));
+ }
+ reconnect(connection, BACKOFF_CAP);
+ }
+ }
+ }
+ ctx.fireChannelInactive();
+ }
+
+ private void reconnect(final Connection connection, final int attempts) {
+ this.lastReconnect = System.currentTimeMillis();
+
+ int timeout = 2 << attempts;
+ if (bootstrap.config().group().isShuttingDown()) {
+ return;
+ }
+
+ if (permit.tryAcquire()) {
+ timer.newTimeout(timeout1 -> tryReconnect(connection, Math.min(BACKOFF_CAP, attempts + 1)), timeout, TimeUnit.MILLISECONDS);
+ }
+ }
+
+
+ private void tryReconnect(final Connection connection, final int nextAttempt) {
+ permit.release();
+
+ if (connection.isClosed() || bootstrap.config().group().isShuttingDown()) {
+ return;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Connection %s is reconnecting, attempt=%d", connection, nextAttempt));
+ }
+
+ bootstrap.connect(connection.getRemote()).addListener((ChannelFutureListener) future -> {
+ if (connection.isClosed() || bootstrap.config().group().isShuttingDown()) {
+ if (future.isSuccess()) {
+ Channel ch = future.channel();
+ Connection con = Connection.getConnectionFromChannel(ch);
+ if (con != null) {
+ con.close();
+ }
+ }
+ return;
+ }
+
+ if (future.isSuccess()) {
+ final Channel channel = future.channel();
+ connection.onConnected(channel);
+ if (!connection.isClosed()) {
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("%s connected to %s", connection, connection.getRemote()));
+ }
+ } else {
+ channel.close();
+ }
+ }
+ reconnect(connection, nextAttempt);
+ });
+ }
+
+}
\ No newline at end of file
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionManager.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionManager.java
new file mode 100644
index 0000000..fc5578f
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionManager.java
@@ -0,0 +1,35 @@
+/*
+ *
+ * * 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.remoting.api;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.SPI;
+import org.apache.dubbo.remoting.RemotingException;
+
+import java.util.function.Consumer;
+
+@SPI
+public interface ConnectionManager {
+
+ Connection connect(URL url) throws RemotingException;
+
+ void forEachConnection(Consumer<Connection> connectionConsumer);
+
+}
\ No newline at end of file
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Http2ProtocolDetector.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Http2ProtocolDetector.java
new file mode 100644
index 0000000..bafc615
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Http2ProtocolDetector.java
@@ -0,0 +1,48 @@
+/*
+ *
+ * * 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.remoting.api;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http2.Http2CodecUtil;
+
+import static java.lang.Math.min;
+
+public class Http2ProtocolDetector implements ProtocolDetector {
+ private final ByteBuf clientPrefaceString = Http2CodecUtil.connectionPrefaceBuf();
+
+ @Override
+ public Result detect(ChannelHandlerContext ctx, ByteBuf in) {
+ int prefaceLen = clientPrefaceString.readableBytes();
+ int bytesRead = min(in.readableBytes(), prefaceLen);
+
+ // If the input so far doesn't match the preface, break the connection.
+ if (bytesRead == 0 || !ByteBufUtil.equals(in, 0,
+ clientPrefaceString, 0, bytesRead)) {
+
+ return Result.UNRECOGNIZED;
+ }
+ if (bytesRead == prefaceLen) {
+ return Result.RECOGNIZED;
+ }
+ return Result.NEED_MORE_DATA;
+ }
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Http2WireProtocol.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Http2WireProtocol.java
new file mode 100644
index 0000000..c261a21
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Http2WireProtocol.java
@@ -0,0 +1,39 @@
+/*
+ *
+ * * 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.remoting.api;
+
+import io.netty.handler.codec.http2.Http2FrameLogger;
+
+import static io.netty.handler.logging.LogLevel.DEBUG;
+
+public abstract class Http2WireProtocol implements WireProtocol {
+ public static final Http2FrameLogger CLIENT_LOGGER = new Http2FrameLogger(DEBUG, "H2_CLIENT");
+ public static final Http2FrameLogger SERVER_LOGGER = new Http2FrameLogger(DEBUG, "H2_SERVER");
+ private final ProtocolDetector detector = new Http2ProtocolDetector();
+
+ @Override
+ public ProtocolDetector detector() {
+ return detector;
+ }
+
+ @Override
+ public void close() {
+ }
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/MultiplexProtocolConnectionManager.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/MultiplexProtocolConnectionManager.java
new file mode 100644
index 0000000..dd45619
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/MultiplexProtocolConnectionManager.java
@@ -0,0 +1,47 @@
+/*
+ *
+ * * 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.remoting.api;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.remoting.RemotingException;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.function.Consumer;
+
+public class MultiplexProtocolConnectionManager implements ConnectionManager {
+ private final ConcurrentMap<String, ConnectionManager> protocols = new ConcurrentHashMap<>();
+
+ @Override
+ public Connection connect(URL url) throws RemotingException {
+ final ConnectionManager manager = protocols.computeIfAbsent(url.getProtocol(), this::createSingleProtocolConnectionManager);
+ return manager.connect(url);
+ }
+
+ @Override
+ public void forEachConnection(Consumer<Connection> connectionConsumer) {
+ protocols.values().forEach(p -> p.forEachConnection(connectionConsumer));
+ }
+
+ private ConnectionManager createSingleProtocolConnectionManager(String protocol) {
+ return ExtensionLoader.getExtensionLoader(ConnectionManager.class).getExtension("single");
+ }
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/NettyEventLoopFactory.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/NettyEventLoopFactory.java
new file mode 100644
index 0000000..d6e2698
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/NettyEventLoopFactory.java
@@ -0,0 +1,66 @@
+/*
+ * 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.remoting.api;
+
+import org.apache.dubbo.common.config.Configuration;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.Epoll;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.epoll.EpollSocketChannel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.ServerSocketChannel;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.concurrent.DefaultThreadFactory;
+
+import java.util.concurrent.ThreadFactory;
+
+public class NettyEventLoopFactory {
+ /**
+ * netty client bootstrap
+ */
+ public static final EventLoopGroup NIO_EVENT_LOOP_GROUP = eventLoopGroup(Constants.DEFAULT_IO_THREADS, "NettyClientWorker");
+
+ public static EventLoopGroup eventLoopGroup(int threads, String threadFactoryName) {
+ ThreadFactory threadFactory = new DefaultThreadFactory(threadFactoryName, true);
+ return shouldEpoll() ? new EpollEventLoopGroup(threads, threadFactory) :
+ new NioEventLoopGroup(threads, threadFactory);
+ }
+
+ public static Class<? extends SocketChannel> socketChannelClass() {
+ return shouldEpoll() ? EpollSocketChannel.class : NioSocketChannel.class;
+ }
+
+ public static Class<? extends ServerSocketChannel> serverSocketChannelClass() {
+ return shouldEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class;
+ }
+
+ private static boolean shouldEpoll() {
+ Configuration configuration = ApplicationModel.getEnvironment().getConfiguration();
+ if (configuration.getBoolean("netty.epoll.enable", false)) {
+ String osName = configuration.getString("os.name");
+ return osName.toLowerCase().contains("linux") && Epoll.isAvailable();
+ }
+
+ return false;
+ }
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/PortUnificationServer.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/PortUnificationServer.java
new file mode 100644
index 0000000..4cf58ca
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/PortUnificationServer.java
@@ -0,0 +1,243 @@
+/*
+ * 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.remoting.api;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ExecutorUtil;
+import org.apache.dubbo.common.utils.NetUtils;
+import org.apache.dubbo.config.SslConfig;
+import org.apache.dubbo.config.context.ConfigManager;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.utils.UrlUtils;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.group.ChannelGroupFuture;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.ssl.ClientAuth;
+import io.netty.handler.ssl.OpenSsl;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslProvider;
+import io.netty.handler.timeout.IdleStateHandler;
+
+import javax.net.ssl.SSLException;
+import java.net.InetSocketAddress;
+import java.security.Provider;
+import java.security.Security;
+import java.util.List;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_VALUE;
+import static org.apache.dubbo.common.constants.CommonConstants.IO_THREADS_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.SSL_ENABLED_KEY;
+
+/**
+ * PortUnificationServer.
+ */
+public class PortUnificationServer {
+
+ private static final Logger logger = LoggerFactory.getLogger(PortUnificationServer.class);
+ private final List<WireProtocol> protocols;
+ private final URL url;
+ /**
+ * netty server bootstrap.
+ */
+ private ServerBootstrap bootstrap;
+ /**
+ * the boss channel that receive connections and dispatch these to worker channel.
+ */
+ private Channel channel;
+ private DefaultChannelGroup channelGroup;
+ private EventLoopGroup bossGroup;
+ private EventLoopGroup workerGroup;
+
+ public PortUnificationServer(URL url) {
+ // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants.
+ // the handler will be wrapped: MultiMessageHandler->HeartbeatHandler->handler
+ this.url = ExecutorUtil.setThreadName(url, "DubboPUServerHandler");
+ this.protocols = ExtensionLoader.getExtensionLoader(WireProtocol.class).getActivateExtension(url, new String[0]);
+ }
+
+ private static boolean checkJdkProvider() {
+ Provider[] jdkProviders = Security.getProviders("SSLContext.TLS");
+ return (jdkProviders != null && jdkProviders.length > 0);
+ }
+
+ private static SslProvider findSslProvider() {
+ if (OpenSsl.isAvailable()) {
+ logger.info("Using OPENSSL provider.");
+ return SslProvider.OPENSSL;
+ } else if (checkJdkProvider()) {
+ logger.info("Using JDK provider.");
+ return SslProvider.JDK;
+ }
+ throw new IllegalStateException(
+ "Could not find any valid TLS provider, please check your dependency or deployment environment, " +
+ "usually netty-tcnative, Conscrypt, or Jetty NPN/ALPN is needed.");
+ }
+
+ public static SslContext buildServerSslContext(URL url) {
+ ConfigManager globalConfigManager = ApplicationModel.getConfigManager();
+ SslConfig sslConfig = globalConfigManager.getSsl().orElseThrow(() -> new IllegalStateException("Ssl enabled, but no ssl cert information provided!"));
+
+ SslContextBuilder sslClientContextBuilder = null;
+ try {
+ String password = sslConfig.getServerKeyPassword();
+ if (password != null) {
+ sslClientContextBuilder = SslContextBuilder.forServer(sslConfig.getServerKeyCertChainPathStream(),
+ sslConfig.getServerPrivateKeyPathStream(), password);
+ } else {
+ sslClientContextBuilder = SslContextBuilder.forServer(sslConfig.getServerKeyCertChainPathStream(),
+ sslConfig.getServerPrivateKeyPathStream());
+ }
+
+ if (sslConfig.getServerTrustCertCollectionPathStream() != null) {
+ sslClientContextBuilder.trustManager(sslConfig.getServerTrustCertCollectionPathStream());
+ sslClientContextBuilder.clientAuth(ClientAuth.REQUIRE);
+ }
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Could not find certificate file or the certificate is invalid.", e);
+ }
+ try {
+ return sslClientContextBuilder.sslProvider(findSslProvider()).build();
+ } catch (SSLException e) {
+ throw new IllegalStateException("Build SslSession failed.", e);
+ }
+ }
+
+ public URL getUrl() {
+ return url;
+ }
+
+ public void bind() {
+ if (channel == null) {
+ doOpen();
+ }
+ }
+
+ public void close() throws Throwable {
+ if (channel != null) {
+ doClose();
+ }
+ }
+
+ /**
+ * Init and start netty server
+ */
+ protected void doOpen() {
+ bootstrap = new ServerBootstrap();
+
+ bossGroup = NettyEventLoopFactory.eventLoopGroup(1, "NettyServerBoss");
+ workerGroup = NettyEventLoopFactory.eventLoopGroup(
+ getUrl().getPositiveParameter(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
+ "NettyServerWorker");
+
+ bootstrap.group(bossGroup, workerGroup)
+ .channel(NettyEventLoopFactory.serverSocketChannelClass())
+ .option(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
+ .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
+ .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
+ .childHandler(new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ // FIXME: should we use getTimeout()?
+ int idleTimeout = UrlUtils.getIdleTimeout(getUrl());
+ final ChannelPipeline p = ch.pipeline();
+// p.addLast(new LoggingHandler(LogLevel.DEBUG));
+ // TODO add SSL support
+ final boolean enableSSL = getUrl().getParameter(SSL_ENABLED_KEY, false);
+ final PortUnificationServerHandler puHandler;
+ if (enableSSL) {
+ final SslContext sslContext = buildServerSslContext(getUrl());
+ puHandler = new PortUnificationServerHandler(sslContext, protocols);
+ } else {
+ puHandler = new PortUnificationServerHandler(protocols);
+ }
+ p.addLast("server-idle-handler", new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS));
+ p.addLast("negotiation", puHandler);
+ channelGroup = puHandler.getChannels();
+ }
+ });
+ // bind
+
+ String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost());
+ int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort());
+ if (url.getParameter(ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) {
+ bindIp = ANYHOST_VALUE;
+ }
+ InetSocketAddress bindAddress = new InetSocketAddress(bindIp, bindPort);
+ ChannelFuture channelFuture = bootstrap.bind(bindAddress);
+ channelFuture.syncUninterruptibly();
+ channel = channelFuture.channel();
+ }
+
+ protected void doClose() throws Throwable {
+ final long st = System.currentTimeMillis();
+
+ try {
+ if (channel != null) {
+ // unbind.
+ channel.close();
+ channel = null;
+ }
+
+ if (channelGroup != null) {
+ ChannelGroupFuture closeFuture = channelGroup.close();
+ closeFuture.await(15000);
+ }
+ final long cost = System.currentTimeMillis() - st;
+ logger.info("Port unification server closed. cost:" + cost);
+ } catch (InterruptedException e) {
+ logger.warn("Interrupted while shutting down", e);
+ }
+
+ for (WireProtocol protocol : protocols) {
+ protocol.close();
+ }
+
+ try {
+ if (bootstrap != null) {
+ bossGroup.shutdownGracefully().syncUninterruptibly();
+ workerGroup.shutdownGracefully().syncUninterruptibly();
+ }
+ } catch (Throwable e) {
+ logger.warn(e.getMessage(), e);
+ }
+ }
+
+ public boolean isBound() {
+ return channel.isActive();
+ }
+
+ public InetSocketAddress getLocalAddress() {
+ return (InetSocketAddress) channel.localAddress();
+ }
+
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/PortUnificationServerHandler.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/PortUnificationServerHandler.java
new file mode 100644
index 0000000..238b391
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/PortUnificationServerHandler.java
@@ -0,0 +1,119 @@
+/*
+ *
+ * * 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.remoting.api;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GlobalEventExecutor;
+
+import java.util.List;
+
+public class PortUnificationServerHandler extends ByteToMessageDecoder {
+
+ private final SslContext sslCtx;
+ private final boolean detectSsl;
+ private final List<WireProtocol> protocols;
+ private final DefaultChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
+
+ public PortUnificationServerHandler(List<WireProtocol> protocols) {
+ this(null, false, protocols);
+ }
+
+ public PortUnificationServerHandler(SslContext sslCtx, List<WireProtocol> protocols) {
+ this(sslCtx, true, protocols);
+ }
+
+ public PortUnificationServerHandler(SslContext sslCtx, boolean detectSsl, List<WireProtocol> protocols) {
+ this.sslCtx = sslCtx;
+ this.protocols = protocols;
+ this.detectSsl = detectSsl;
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ super.exceptionCaught(ctx, cause);
+ }
+
+ public DefaultChannelGroup getChannels() {
+ return channels;
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ super.channelActive(ctx);
+ channels.add(ctx.channel());
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ super.channelInactive(ctx);
+ channels.remove(ctx.channel());
+ }
+
+ @Override
+ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+ // Will use the first five bytes to detect a protocol.
+ if (in.readableBytes() < 5) {
+ return;
+ }
+
+ if (isSsl(in)) {
+ enableSsl(ctx);
+ } else {
+ for (final WireProtocol protocol : protocols) {
+ in.markReaderIndex();
+ final ProtocolDetector.Result result = protocol.detector().detect(ctx, in);
+ in.resetReaderIndex();
+ switch (result) {
+ case UNRECOGNIZED:
+ continue;
+ case RECOGNIZED:
+ protocol.configServerPipeline(ctx.pipeline(), null);
+ ctx.pipeline().remove(this);
+ case NEED_MORE_DATA:
+ return;
+ }
+ }
+ // Unknown protocol; discard everything and close the connection.
+ in.clear();
+ ctx.close();
+ }
+ }
+
+ private boolean isSsl(ByteBuf buf) {
+ if (detectSsl) {
+ return SslHandler.isEncrypted(buf);
+ }
+ return false;
+ }
+
+ private void enableSsl(ChannelHandlerContext ctx) {
+ ChannelPipeline p = ctx.pipeline();
+ p.addLast("ssl", sslCtx.newHandler(ctx.alloc()));
+ p.addLast("unificationA", new PortUnificationServerHandler(sslCtx, false, protocols));
+ p.remove(this);
+ }
+
+}
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
new file mode 100644
index 0000000..f8183a9
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ProtocolDetector.java
@@ -0,0 +1,38 @@
+/*
+ *
+ * * 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.remoting.api;
+
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+
+/**
+ * Determine incoming bytes belong to the specific protocol.
+ *
+ * @author guohaoice@gmail.com
+ */
+public interface ProtocolDetector {
+
+ Result detect(final ChannelHandlerContext ctx, final ByteBuf in);
+
+ enum Result {
+ RECOGNIZED, UNRECOGNIZED, NEED_MORE_DATA
+ }
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/SingleProtocolConnectionManager.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/SingleProtocolConnectionManager.java
new file mode 100644
index 0000000..5135fe5
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/SingleProtocolConnectionManager.java
@@ -0,0 +1,60 @@
+/*
+ *
+ * * 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.remoting.api;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.remoting.RemotingException;
+
+import io.netty.util.internal.PlatformDependent;
+
+import java.util.concurrent.ConcurrentMap;
+import java.util.function.Consumer;
+
+import static org.apache.dubbo.common.constants.CommonConstants.LAZY_CONNECT_KEY;
+
+public class SingleProtocolConnectionManager implements ConnectionManager {
+ private final ConcurrentMap<String, Connection> connections = PlatformDependent.newConcurrentHashMap();
+
+ @Override
+ public Connection connect(URL url) throws RemotingException {
+ if (url == null) {
+ throw new IllegalArgumentException("url == null");
+ }
+ final Connection connection = connections.compute(url.getAddress(), (address, conn) -> {
+ if (conn == null) {
+ final Connection created = new Connection(url);
+ created.getCloseFuture().addListener(future -> connections.remove(address, created));
+ return created;
+ } else {
+ conn.retain();
+ return conn;
+ }
+ });
+ if (!url.getParameter(LAZY_CONNECT_KEY, false)) {
+ connection.init();
+ }
+ return connection;
+ }
+
+ @Override
+ public void forEachConnection(Consumer<Connection> connectionConsumer) {
+ connections.values().forEach(connectionConsumer);
+ }
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/WireProtocol.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/WireProtocol.java
new file mode 100644
index 0000000..988815b
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/WireProtocol.java
@@ -0,0 +1,37 @@
+/*
+ *
+ * * 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.remoting.api;
+
+import org.apache.dubbo.common.extension.SPI;
+
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.ssl.SslContext;
+
+@SPI
+public interface WireProtocol {
+
+ ProtocolDetector detector();
+
+ void configServerPipeline(ChannelPipeline pipeline, SslContext sslContext);
+
+ void configClientPipeline(ChannelPipeline pipeline, SslContext sslContext);
+
+ void close();
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
new file mode 100644
index 0000000..493c52d
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
@@ -0,0 +1,56 @@
+/*
+ *
+ * * 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.remoting.exchange;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.remoting.api.PortUnificationServer;
+
+import io.netty.util.internal.PlatformDependent;
+
+import java.util.ArrayList;
+import java.util.concurrent.ConcurrentMap;
+
+public class PortUnificationExchanger {
+
+ private static final Logger log = LoggerFactory.getLogger(PortUnificationExchanger.class);
+ private static final ConcurrentMap<String, PortUnificationServer> servers = PlatformDependent.newConcurrentHashMap();
+
+ public static void bind(URL url) {
+ servers.computeIfAbsent(url.getAddress(), addr -> {
+ final PortUnificationServer server = new PortUnificationServer(url);
+ server.bind();
+ return server;
+ });
+ }
+
+ public static void close() {
+ final ArrayList<PortUnificationServer> toClose = new ArrayList<>(servers.values());
+ servers.clear();
+ for (PortUnificationServer server : toClose) {
+ try {
+ server.close();
+ } catch (Throwable throwable) {
+ log.error("Close all port unification server failed", throwable);
+ }
+ }
+ }
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture2.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture2.java
new file mode 100644
index 0000000..afa7c7f
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture2.java
@@ -0,0 +1,295 @@
+/*
+ * 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.remoting.exchange.support;
+
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.threadpool.ThreadlessExecutor;
+import org.apache.dubbo.common.timer.HashedWheelTimer;
+import org.apache.dubbo.common.timer.Timeout;
+import org.apache.dubbo.common.timer.Timer;
+import org.apache.dubbo.common.timer.TimerTask;
+import org.apache.dubbo.common.utils.NamedThreadFactory;
+import org.apache.dubbo.remoting.RemotingException;
+import org.apache.dubbo.remoting.TimeoutException;
+import org.apache.dubbo.remoting.api.Connection;
+import org.apache.dubbo.remoting.exchange.Request;
+import org.apache.dubbo.remoting.exchange.Response;
+
+import io.netty.channel.Channel;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * DefaultFuture.
+ */
+public class DefaultFuture2 extends CompletableFuture<Object> {
+
+ public static final Timer TIME_OUT_TIMER = new HashedWheelTimer(
+ new NamedThreadFactory("dubbo-future-timeout", true),
+ 30,
+ TimeUnit.MILLISECONDS);
+ private static final Logger logger = LoggerFactory.getLogger(DefaultFuture2.class);
+ private static final Map<Long, Connection> CONNECTIONS = new ConcurrentHashMap<>();
+ private static final Map<Long, DefaultFuture2> FUTURES = new ConcurrentHashMap<>();
+ // invoke id.
+ private final Long id;
+ private final Connection connection;
+ private final Request request;
+ private final int timeout;
+ private final long start = System.currentTimeMillis();
+ private volatile long sent;
+ private Timeout timeoutCheckTask;
+
+ private ExecutorService executor;
+
+ private DefaultFuture2(Connection client2, Request request, int timeout) {
+ this.connection = client2;
+ this.request = request;
+ this.id = request.getId();
+ this.timeout = timeout;
+ // put into waiting map.
+ FUTURES.put(id, this);
+ CONNECTIONS.put(id, connection);
+ }
+
+ /**
+ * check time out of the future
+ */
+ private static void timeoutCheck(DefaultFuture2 future) {
+ TimeoutCheckTask task = new TimeoutCheckTask(future.getId());
+ future.timeoutCheckTask = TIME_OUT_TIMER.newTimeout(task, future.getTimeout(), TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * init a DefaultFuture
+ * 1.init a DefaultFuture
+ * 2.timeout check
+ *
+ * @param channel channel
+ * @param request the request
+ * @param timeout timeout
+ * @return a new DefaultFuture
+ */
+ public static DefaultFuture2 newFuture(Connection channel, Request request, int timeout, ExecutorService executor) {
+ final DefaultFuture2 future = new DefaultFuture2(channel, request, timeout);
+ future.setExecutor(executor);
+ // ThreadlessExecutor needs to hold the waiting future in case of circuit return.
+ if (executor instanceof ThreadlessExecutor) {
+ ((ThreadlessExecutor) executor).setWaitingFuture(future);
+ }
+ // timeout check
+ timeoutCheck(future);
+ return future;
+ }
+
+ public static DefaultFuture2 getFuture(long id) {
+ return FUTURES.get(id);
+ }
+
+ public static boolean hasFuture(Channel channel) {
+ return CONNECTIONS.containsValue(channel);
+ }
+
+ public static void sent(Request request) {
+ DefaultFuture2 future = FUTURES.get(request.getId());
+ if (future != null) {
+ future.doSent();
+ }
+ }
+
+ /**
+ * close a channel when a channel is inactive
+ * directly return the unfinished requests.
+ *
+ * @param connection channel to close
+ */
+ public static void closeChannel(Connection connection) {
+ for (Map.Entry<Long, Connection> entry : CONNECTIONS.entrySet()) {
+ if (connection.equals(entry.getValue())) {
+ DefaultFuture2 future = getFuture(entry.getKey());
+ if (future != null && !future.isDone()) {
+ ExecutorService futureExecutor = future.getExecutor();
+ if (futureExecutor != null && !futureExecutor.isTerminated()) {
+ futureExecutor.shutdownNow();
+ }
+
+ Response disconnectResponse = new Response(future.getId());
+ disconnectResponse.setStatus(Response.CHANNEL_INACTIVE);
+ disconnectResponse.setErrorMessage("Channel " +
+ connection +
+ " is inactive. Directly return the unFinished request : " +
+ future.getRequest());
+ DefaultFuture2.received(connection, disconnectResponse);
+ }
+ }
+ }
+ }
+
+ public static void received(Connection connection, Response response) {
+ received(connection, response, false);
+ }
+
+ public static void received(Connection connection, Response response, boolean timeout) {
+ try {
+ DefaultFuture2 future = FUTURES.remove(response.getId());
+ if (future != null) {
+ Timeout t = future.timeoutCheckTask;
+ if (!timeout) {
+ // decrease Time
+ t.cancel();
+ }
+ future.doReceived(response);
+ } else {
+ logger.warn("The timeout response finally returned at "
+ + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
+ + ", response status is " + response.getStatus()
+ + (connection == null ? "" : ", channel: " + connection.getChannel().localAddress()
+ + " -> " + connection.getRemote()) + ", please check provider side for detailed result.");
+ }
+ } finally {
+ CONNECTIONS.remove(response.getId());
+ }
+ }
+
+ public ExecutorService getExecutor() {
+ return executor;
+ }
+
+ public void setExecutor(ExecutorService executor) {
+ this.executor = executor;
+ }
+
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ Response errorResult = new Response(id);
+ errorResult.setStatus(Response.CLIENT_ERROR);
+ errorResult.setErrorMessage("request future has been canceled.");
+ this.doReceived(errorResult);
+ FUTURES.remove(id);
+ CONNECTIONS.remove(id);
+ return true;
+ }
+
+ public void cancel() {
+ this.cancel(true);
+ }
+
+ private void doReceived(Response res) {
+ if (res == null) {
+ throw new IllegalStateException("response cannot be null");
+ }
+ if (res.getStatus() == Response.OK) {
+ this.complete(res.getResult());
+ } else if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
+ this.completeExceptionally(new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, null, connection.getRemote(), res.getErrorMessage()));
+ } else {
+ this.completeExceptionally(new RemotingException(null, connection.getRemote(), res.getErrorMessage()));
+ }
+
+ // the result is returning, but the caller thread may still waiting
+ // to avoid endless waiting for whatever reason, notify caller thread to return.
+ if (executor != null && executor instanceof ThreadlessExecutor) {
+ ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor;
+ if (threadlessExecutor.isWaiting()) {
+ threadlessExecutor.notifyReturn(new IllegalStateException("The result has returned, but the biz thread is still waiting" +
+ " which is not an expected state, interrupt the thread manually by returning an exception."));
+ }
+ }
+ }
+
+ private long getId() {
+ return id;
+ }
+
+ private Connection getConnection() {
+ return connection;
+ }
+
+ private boolean isSent() {
+ return sent > 0;
+ }
+
+ public Request getRequest() {
+ return request;
+ }
+
+ private int getTimeout() {
+ return timeout;
+ }
+
+ private void doSent() {
+ sent = System.currentTimeMillis();
+ }
+
+ private String getTimeoutMessage(boolean scan) {
+ long nowTimestamp = System.currentTimeMillis();
+ return (sent > 0 ? "Waiting server-side response timeout" : "Sending request timeout in client-side")
+ + (scan ? " by scan timer" : "") + ". start time: "
+ + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date(start))) + ", end time: "
+ + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date(nowTimestamp))) + ","
+ + (sent > 0 ? " client elapsed: " + (sent - start)
+ + " ms, server elapsed: " + (nowTimestamp - sent)
+ : " elapsed: " + (nowTimestamp - start)) + " ms, timeout: "
+ + timeout + " ms, request: " + (logger.isDebugEnabled() ? request : getRequestWithoutData()) + ", channel: " + connection.getChannel();
+ }
+
+ private Request getRequestWithoutData() {
+ Request newRequest = request;
+ newRequest.setData(null);
+ return newRequest;
+ }
+
+ private static class TimeoutCheckTask implements TimerTask {
+
+ private final Long requestID;
+
+ TimeoutCheckTask(Long requestID) {
+ this.requestID = requestID;
+ }
+
+ @Override
+ public void run(Timeout timeout) {
+ DefaultFuture2 future = DefaultFuture2.getFuture(requestID);
+ if (future == null || future.isDone()) {
+ return;
+ }
+
+ if (future.getExecutor() != null) {
+ future.getExecutor().execute(() -> notifyTimeout(future));
+ } else {
+ notifyTimeout(future);
+ }
+ }
+
+ private void notifyTimeout(DefaultFuture2 future) {
+ // create exception response.
+ Response timeoutResponse = new Response(future.getId());
+ // set timeout status.
+ timeoutResponse.setStatus(future.isSent() ? Response.SERVER_TIMEOUT : Response.CLIENT_TIMEOUT);
+ timeoutResponse.setErrorMessage(future.getTimeoutMessage(true));
+ // handle response.
+ DefaultFuture2.received(future.getConnection(), timeoutResponse, true);
+ }
+ }
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.api.ConnectionManager b/dubbo-remoting/dubbo-remoting-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.api.ConnectionManager
new file mode 100644
index 0000000..b2c728f
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.api.ConnectionManager
@@ -0,0 +1,2 @@
+multiple=org.apache.dubbo.remoting.api.MultiplexProtocolConnectionManager
+single=org.apache.dubbo.remoting.api.SingleProtocolConnectionManager
diff --git a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/netty4/ConnectionTest.java b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/netty4/ConnectionTest.java
new file mode 100644
index 0000000..47d645d
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/netty4/ConnectionTest.java
@@ -0,0 +1,62 @@
+/*
+ *
+ * * 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.remoting.netty4;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.remoting.api.Connection;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+
+class ConnectionTest {
+
+ @Test
+ public void testRefCnt0() throws InterruptedException {
+ Connection connection = new Connection(URL.valueOf("empty://127.0.0.1:8080?foo=bar"));
+ CountDownLatch latch = new CountDownLatch(1);
+ connection.getCloseFuture().addListener(future -> latch.countDown());
+ connection.release();
+ latch.await();
+ Assertions.assertEquals(0, latch.getCount());
+ }
+
+ @Test
+ public void testRefCnt1() throws InterruptedException {
+ Connection connection = new Connection(URL.valueOf("empty://127.0.0.1:8080?foo=bar"));
+ CountDownLatch latch = new CountDownLatch(1);
+ connection.retain();
+ connection.getCloseFuture().addListener(future -> latch.countDown());
+ connection.release();
+ Assertions.assertEquals(1, latch.getCount());
+ }
+
+ @Test
+ public void testRefCnt2() throws InterruptedException {
+ Connection connection = new Connection(URL.valueOf("empty://127.0.0.1:8080?foo=bar"));
+ CountDownLatch latch = new CountDownLatch(1);
+ connection.retain();
+ connection.getCloseFuture().addListener(future -> latch.countDown());
+ connection.release(2);
+ latch.await();
+ Assertions.assertEquals(0, latch.getCount());
+ }
+}
\ No newline at end of file
diff --git a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/netty4/EmptyProtocol.java b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/netty4/EmptyProtocol.java
new file mode 100644
index 0000000..9a3651f
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/netty4/EmptyProtocol.java
@@ -0,0 +1,48 @@
+/*
+ *
+ * * 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.remoting.netty4;
+
+import org.apache.dubbo.remoting.api.ProtocolDetector;
+import org.apache.dubbo.remoting.api.WireProtocol;
+
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.ssl.SslContext;
+
+public class EmptyProtocol implements WireProtocol {
+ @Override
+ public ProtocolDetector detector() {
+ return null;
+ }
+
+ @Override
+ public void configServerPipeline(ChannelPipeline pipeline,SslContext sslContext) {
+
+ }
+
+ @Override
+ public void configClientPipeline(ChannelPipeline pipeline, SslContext sslContext) {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/netty4/NettyServerTest.java b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/netty4/NettyServerTest.java
new file mode 100644
index 0000000..cb6c6af
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/netty4/NettyServerTest.java
@@ -0,0 +1,36 @@
+/*
+ *
+ * * 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.remoting.netty4;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.RemotingException;
+import org.apache.dubbo.remoting.api.PortUnificationServer;
+
+public class NettyServerTest {
+
+ public static void main(String[] args) throws RemotingException {
+ URL url = new URL("transport", "localhost", 8898,
+ new String[]{Constants.BIND_PORT_KEY, String.valueOf(8898)});
+
+ final PortUnificationServer server = new PortUnificationServer(url);
+ System.out.println(server.isBound());
+ }
+}
diff --git a/dubbo-remoting/dubbo-remoting-api/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.netty4.WireProtocol b/dubbo-remoting/dubbo-remoting-api/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.netty4.WireProtocol
new file mode 100644
index 0000000..e5dd6c9
--- /dev/null
+++ b/dubbo-remoting/dubbo-remoting-api/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.netty4.WireProtocol
@@ -0,0 +1 @@
+empty=org.apache.dubbo.remoting.netty4.EmptyProtocol
\ No newline at end of file
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
index ba6f9b6..36ddee0 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
@@ -30,6 +30,7 @@ import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.Locale;
import java.util.Map;
import java.util.stream.Stream;
@@ -197,20 +198,24 @@ public class RpcInvocation implements Invocation, Serializable {
return targetServiceUniqueName;
}
+ public void setTargetServiceUniqueName(String targetServiceUniqueName) {
+ this.targetServiceUniqueName = targetServiceUniqueName;
+ }
+
@Override
public String getProtocolServiceKey() {
return protocolServiceKey;
}
- public void setTargetServiceUniqueName(String targetServiceUniqueName) {
- this.targetServiceUniqueName = targetServiceUniqueName;
- }
-
@Override
public String getMethodName() {
return methodName;
}
+ public void setMethodName(String methodName) {
+ this.methodName = methodName;
+ }
+
@Override
public String getServiceName() {
return serviceName;
@@ -220,10 +225,6 @@ public class RpcInvocation implements Invocation, Serializable {
this.serviceName = serviceName;
}
- public void setMethodName(String methodName) {
- this.methodName = methodName;
- }
-
@Override
public Class<?>[] getParameterTypes() {
return parameterTypes;
@@ -265,6 +266,10 @@ public class RpcInvocation implements Invocation, Serializable {
return attachments;
}
+ public void setObjectAttachments(Map<String, Object> attachments) {
+ this.attachments = attachments == null ? new HashMap<>() : attachments;
+ }
+
@Override
public void setAttachment(String key, String value) {
setObjectAttachment(key, value);
@@ -281,12 +286,8 @@ public class RpcInvocation implements Invocation, Serializable {
this.attachments = attachments == null ? new HashMap<>() : new HashMap<>(attachments);
}
- public void setObjectAttachments(Map<String, Object> attachments) {
- this.attachments = attachments == null ? new HashMap<>() : attachments;
- }
-
public void setAttachment(String key, Object value) {
- setObjectAttachment(key, value);
+ setObjectAttachment(key, value);
}
@Override
@@ -374,7 +375,11 @@ public class RpcInvocation implements Invocation, Serializable {
if (attachments == null) {
return null;
}
- return attachments.get(key);
+ final Object val = attachments.get(key);
+ if (val != null) {
+ return val;
+ }
+ return attachments.get(key.toLowerCase(Locale.ROOT));
}
@Override
diff --git a/dubbo-rpc/dubbo-rpc-triple/pom.xml b/dubbo-rpc/dubbo-rpc-triple/pom.xml
new file mode 100644
index 0000000..8ffe027
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/pom.xml
@@ -0,0 +1,74 @@
+<!--
+ 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.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.dubbo</groupId>
+ <artifactId>dubbo-rpc</artifactId>
+ <version>${revision}</version>
+ </parent>
+ <artifactId>dubbo-rpc-triple</artifactId>
+ <packaging>jar</packaging>
+ <name>${project.artifactId}</name>
+ <description>The triple protocol module</description>
+ <properties>
+ <skip_maven_deploy>false</skip_maven_deploy>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.dubbo</groupId>
+ <artifactId>dubbo-rpc-api</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <extensions>
+ <extension>
+ <groupId>kr.motd.maven</groupId>
+ <artifactId>os-maven-plugin</artifactId>
+ <version>1.6.2</version>
+ </extension>
+ </extensions>
+ <plugins>
+ <plugin>
+ <groupId>org.xolstice.maven.plugins</groupId>
+ <artifactId>protobuf-maven-plugin</artifactId>
+ <version>0.6.1</version>
+ <configuration>
+ <protocArtifact>com.google.protobuf:protoc:3.12.0:exe:${os.detected.classifier}</protocArtifact>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>compile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractStream.java
new file mode 100644
index 0000000..5fef2a2
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractStream.java
@@ -0,0 +1,179 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.serialize.MultipleSerialization;
+import org.apache.dubbo.common.utils.ConfigUtils;
+import org.apache.dubbo.config.Constants;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http2.Http2Headers;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+
+import static org.apache.dubbo.rpc.protocol.tri.TripleUtil.responseErr;
+
+public abstract class AbstractStream implements Stream {
+ public static final boolean ENABLE_ATTACHMENT_WRAP = Boolean.parseBoolean(ConfigUtils.getProperty("triple.attachment", "false"));
+ private static final Logger LOGGER = LoggerFactory.getLogger(AbstractStream.class);
+ private static final GrpcStatus TOO_MANY_DATA = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withDescription("Too many data");
+ private final ChannelHandlerContext ctx;
+ private final URL url;
+ private MultipleSerialization multipleSerialization;
+ private Http2Headers headers;
+ private Http2Headers te;
+ private boolean needWrap;
+ private InputStream data;
+ private String serializeType;
+
+ protected AbstractStream(URL url, ChannelHandlerContext ctx) {
+ this(url, ctx, false);
+ }
+
+ protected AbstractStream(URL url, ChannelHandlerContext ctx, boolean needWrap) {
+ this.ctx = ctx;
+ this.url = url;
+ this.needWrap = needWrap;
+ if (needWrap) {
+ loadFromURL(url);
+ }
+ }
+
+ protected void loadFromURL(URL url) {
+ final String value = url.getParameter(Constants.MULTI_SERIALIZATION_KEY, "default");
+ this.multipleSerialization = ExtensionLoader.getExtensionLoader(MultipleSerialization.class).getExtension(value);
+ }
+
+ public URL getUrl() {
+ return url;
+ }
+
+ public String getSerializeType() {
+ return serializeType;
+ }
+
+ protected void setSerializeType(String serializeType) {
+ this.serializeType = serializeType;
+ }
+
+ protected boolean isNeedWrap() {
+ return needWrap;
+ }
+
+ protected void setNeedWrap(boolean needWrap) {
+ this.needWrap = needWrap;
+ }
+
+ public ChannelHandlerContext getCtx() {
+ return ctx;
+ }
+
+ public Http2Headers getHeaders() {
+ return headers;
+ }
+
+ public Http2Headers getTe() {
+ return te;
+ }
+
+ public InputStream getData() {
+ return data;
+ }
+
+ public MultipleSerialization getMultipleSerialization() {
+ return multipleSerialization;
+ }
+
+ @Override
+ public void onData(InputStream in) {
+ if (data != null) {
+ responseErr(ctx, TOO_MANY_DATA);
+ return;
+ }
+
+ this.data = in;
+ }
+
+ public void onHeaders(Http2Headers headers) {
+ if (this.headers == null) {
+ this.headers = headers;
+ } else if (te == null) {
+ this.te = headers;
+ }
+ }
+
+ protected Map<String, Object> parseHeadersToMap(Http2Headers headers) {
+ Map<String, Object> attachments = new HashMap<>();
+ for (Map.Entry<CharSequence, CharSequence> header : headers) {
+ String key = header.getKey().toString();
+
+ if (ENABLE_ATTACHMENT_WRAP) {
+ if (key.endsWith("-tw-bin") && key.length() > 7) {
+ try {
+ attachments.put(key.substring(0, key.length() - 7), TripleUtil.decodeObjFromHeader(getUrl(), header.getValue(), getMultipleSerialization()));
+ } catch (Exception e) {
+ LOGGER.error("Failed to parse response attachment key=" + key, e);
+ }
+ }
+ }
+ if (key.endsWith("-bin") && key.length() > 4) {
+ try {
+ attachments.put(key.substring(0, key.length() - 4), TripleUtil.decodeASCIIByte(header.getValue()));
+ } catch (Exception e) {
+ LOGGER.error("Failed to parse response attachment key=" + key, e);
+ }
+ } else {
+ attachments.put(key, header.getValue().toString());
+ }
+ }
+ return attachments;
+ }
+
+ protected void convertAttachment(Http2Headers trailers, Map<String, Object> attachments) throws IOException {
+ for (Map.Entry<String, Object> entry : attachments.entrySet()) {
+ final String key = entry.getKey().toLowerCase(Locale.ROOT);
+ final Object v = entry.getValue();
+ if (!ENABLE_ATTACHMENT_WRAP) {
+ if (v instanceof String) {
+ trailers.addObject(key, v);
+ } else if (v instanceof byte[]) {
+ trailers.addObject(key + "-bin", TripleUtil.encodeBase64((byte[]) v));
+ }
+ } else {
+ if (v instanceof String || serializeType == null) {
+ trailers.addObject(key, v);
+ } else {
+ String encoded = TripleUtil.encodeWrapper(url, v, this.serializeType, getMultipleSerialization());
+ trailers.add(key + "-tw-bin", encoded);
+ }
+ }
+ }
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClassLoadUtil.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClassLoadUtil.java
new file mode 100644
index 0000000..5a408b2
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClassLoadUtil.java
@@ -0,0 +1,32 @@
+/*
+ *
+ * * 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;
+
+public class ClassLoadUtil {
+ public static void switchContextLoader(ClassLoader loader) {
+ try {
+ if (loader != null && loader != Thread.currentThread().getContextClassLoader()) {
+ Thread.currentThread().setContextClassLoader(loader);
+ }
+ } catch (SecurityException e) {
+ // ignore , ForkJoinPool & jdk8 & securityManager will cause this
+ }
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClientStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClientStream.java
new file mode 100644
index 0000000..1da06dd
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClientStream.java
@@ -0,0 +1,224 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.api.Connection;
+import org.apache.dubbo.remoting.exchange.Request;
+import org.apache.dubbo.remoting.exchange.Response;
+import org.apache.dubbo.remoting.exchange.support.DefaultFuture2;
+import org.apache.dubbo.rpc.AppResponse;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.ConsumerModel;
+import org.apache.dubbo.rpc.model.MethodDescriptor;
+import org.apache.dubbo.rpc.model.ServiceRepository;
+import org.apache.dubbo.triple.TripleWrapper;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
+import io.netty.handler.codec.http2.Http2Headers;
+import io.netty.handler.codec.http2.Http2NoMoreStreamIdsException;
+import io.netty.handler.codec.http2.Http2StreamChannel;
+import io.netty.handler.codec.http2.Http2StreamChannelBootstrap;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+
+import static org.apache.dubbo.rpc.Constants.CONSUMER_MODEL;
+import static org.apache.dubbo.rpc.protocol.tri.TripleUtil.responseErr;
+
+public class ClientStream extends AbstractStream implements Stream {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClientStream.class);
+ private static final GrpcStatus MISSING_RESP = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withDescription("Missing Response");
+ private final Request request;
+ private final RpcInvocation invocation;
+
+
+ public ClientStream(URL url, ChannelHandlerContext ctx, boolean needWrap, Request request) {
+ super(url, ctx, needWrap);
+ if (needWrap) {
+ setSerializeType((String) ((RpcInvocation) (request.getData())).getObjectAttachment(Constants.SERIALIZATION_KEY));
+ }
+ this.request = request;
+ this.invocation = (RpcInvocation) request.getData();
+ }
+
+ public static ConsumerModel getConsumerModel(Invocation invocation) {
+ Object o = invocation.get(CONSUMER_MODEL);
+ if (o instanceof ConsumerModel) {
+ return (ConsumerModel) o;
+ }
+ String serviceKey = invocation.getInvoker().getUrl().getServiceKey();
+ return ApplicationModel.getConsumerModel(serviceKey);
+ }
+
+ @Override
+ public void onError(GrpcStatus status) {
+ Response response = new Response(request.getId(), request.getVersion());
+ if (status.description != null) {
+ response.setErrorMessage(status.description);
+ } else {
+ response.setErrorMessage(status.cause.getMessage());
+ }
+ // TODO map grpc status to response status
+ response.setStatus(Response.BAD_REQUEST);
+ DefaultFuture2.received(Connection.getConnectionFromChannel(getCtx().channel()), response);
+ }
+
+ @Override
+ public void write(Object obj, ChannelPromise promise) throws IOException {
+ Http2Headers headers = new DefaultHttp2Headers()
+ .method(HttpMethod.POST.asciiName())
+ .path("/" + invocation.getObjectAttachment(CommonConstants.PATH_KEY) + "/" + invocation.getMethodName())
+ .set(HttpHeaderNames.CONTENT_TYPE, TripleConstant.CONTENT_PROTO)
+ .set(HttpHeaderNames.TE, HttpHeaderValues.TRAILERS);
+
+ final String version = (String) invocation.getObjectAttachment(CommonConstants.VERSION_KEY);
+ if (version != null) {
+ headers.set(TripleConstant.SERVICE_VERSION, version);
+ invocation.getObjectAttachments().remove(CommonConstants.VERSION_KEY);
+ }
+
+ final String app = (String) invocation.getObjectAttachment(CommonConstants.APPLICATION_KEY);
+ if (app != null) {
+ headers.set(TripleConstant.CONSUMER_APP_NAME_KEY, app);
+ invocation.getObjectAttachments().remove(CommonConstants.APPLICATION_KEY);
+ }
+
+ final String group = (String) invocation.getObjectAttachment(CommonConstants.GROUP_KEY);
+ if (group != null) {
+ headers.set(TripleConstant.SERVICE_GROUP, group);
+ invocation.getObjectAttachments().remove(CommonConstants.GROUP_KEY);
+ }
+ final Map<String, Object> attachments = invocation.getObjectAttachments();
+ if (attachments != null) {
+ convertAttachment(headers, attachments);
+ }
+ DefaultHttp2HeadersFrame frame = new DefaultHttp2HeadersFrame(headers);
+ final TripleHttp2ClientResponseHandler responseHandler = new TripleHttp2ClientResponseHandler();
+
+ final Http2StreamChannelBootstrap streamChannelBootstrap = new Http2StreamChannelBootstrap(getCtx().channel());
+ final Http2StreamChannel streamChannel = streamChannelBootstrap.open().syncUninterruptibly().getNow();
+ TripleUtil.setClientStream(streamChannel, this);
+ streamChannel.pipeline().addLast(responseHandler)
+ .addLast(new GrpcDataDecoder(Integer.MAX_VALUE))
+ .addLast(new TripleClientInboundHandler());
+ streamChannel.write(frame).addListener(future -> {
+ if (!future.isSuccess()) {
+ if (future.cause() instanceof Http2NoMoreStreamIdsException) {
+ getCtx().close();
+ }
+ promise.setFailure(future.cause());
+ }
+ });
+ final ByteBuf out;
+
+ ClassLoader tccl = Thread.currentThread().getContextClassLoader();
+ try {
+ final ConsumerModel model = getConsumerModel(invocation);
+ if (model != null) {
+ ClassLoadUtil.switchContextLoader(model.getClassLoader());
+ }
+ if (isNeedWrap()) {
+ final TripleWrapper.TripleRequestWrapper wrap = TripleUtil.wrapReq(getUrl(), invocation, getMultipleSerialization());
+ out = TripleUtil.pack(getCtx(), wrap);
+ } else {
+ out = TripleUtil.pack(getCtx(), invocation.getArguments()[0]);
+ }
+ } finally {
+ ClassLoadUtil.switchContextLoader(tccl);
+ }
+ streamChannel.write(new DefaultHttp2DataFrame(out, true));
+
+ }
+
+ public void halfClose() {
+ final int httpCode = HttpResponseStatus.parseLine(getHeaders().status()).code();
+ if (HttpResponseStatus.OK.code() != httpCode) {
+ final Integer code = getHeaders().getInt(TripleConstant.STATUS_KEY);
+ final GrpcStatus status = GrpcStatus.fromCode(code)
+ .withDescription(TripleUtil.percentDecode(getHeaders().get(TripleConstant.MESSAGE_KEY)));
+ onError(status);
+ return;
+ }
+ Http2Headers te = getTe();
+ if (te == null) {
+ te = getHeaders();
+ }
+ final Integer code = te.getInt(TripleConstant.STATUS_KEY);
+ if (!GrpcStatus.Code.isOk(code)) {
+ final GrpcStatus status = GrpcStatus.fromCode(code)
+ .withDescription(TripleUtil.percentDecode(getHeaders().get(TripleConstant.MESSAGE_KEY)));
+ onError(status);
+ return;
+ }
+ final InputStream data = getData();
+ if (data == null) {
+ responseErr(getCtx(), MISSING_RESP);
+ return;
+ }
+ final Invocation invocation = (Invocation) (request.getData());
+ ServiceRepository repo = ApplicationModel.getServiceRepository();
+ MethodDescriptor methodDescriptor = repo.lookupMethod(invocation.getServiceName(), invocation.getMethodName());
+ ClassLoader tccl = Thread.currentThread().getContextClassLoader();
+ try {
+ final Object resp;
+ final ConsumerModel model = getConsumerModel(invocation);
+ if (model != null) {
+ ClassLoadUtil.switchContextLoader(model.getClassLoader());
+ }
+ if (isNeedWrap()) {
+ final TripleWrapper.TripleResponseWrapper message = TripleUtil.unpack(data, TripleWrapper.TripleResponseWrapper.class);
+ resp = TripleUtil.unwrapResp(getUrl(), message, getMultipleSerialization());
+ } else {
+ resp = TripleUtil.unpack(data, methodDescriptor.getReturnClass());
+ }
+ Response response = new Response(request.getId(), request.getVersion());
+ final AppResponse result = new AppResponse(resp);
+ result.setObjectAttachments(parseHeadersToMap(te));
+ response.setResult(result);
+ DefaultFuture2.received(Connection.getConnectionFromChannel(getCtx().channel()), response);
+ } catch (Exception e) {
+ final GrpcStatus status = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withCause(e)
+ .withDescription("Failed to deserialize response");
+ onError(status);
+ } finally {
+ ClassLoadUtil.switchContextLoader(tccl);
+ }
+ }
+
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GracefulShutdown.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GracefulShutdown.java
new file mode 100644
index 0000000..60d8fbe
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GracefulShutdown.java
@@ -0,0 +1,81 @@
+/*
+ *
+ * * 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.buffer.ByteBufUtil;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.http2.DefaultHttp2GoAwayFrame;
+import io.netty.handler.codec.http2.DefaultHttp2PingFrame;
+import io.netty.handler.codec.http2.Http2Error;
+import io.netty.handler.codec.http2.Http2GoAwayFrame;
+import io.netty.handler.codec.http2.Http2PingFrame;
+import io.netty.util.concurrent.Future;
+
+import java.util.concurrent.TimeUnit;
+
+public class GracefulShutdown {
+ static final long GRACEFUL_SHUTDOWN_PING = 0x97ACEF001L;
+ private static final long GRACEFUL_SHUTDOWN_PING_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(10);
+ private final ChannelHandlerContext ctx;
+ private final ChannelPromise originPromise;
+ private boolean pingAckedOrTimeout;
+ private final String goAwayMessage;
+ private Future<?> pingFuture;
+
+ public GracefulShutdown(ChannelHandlerContext ctx, String goAwayMessage, ChannelPromise originPromise) {
+ this.ctx = ctx;
+ this.goAwayMessage = goAwayMessage;
+ this.originPromise = originPromise;
+ }
+
+ public void gracefulShutdown() {
+ Http2GoAwayFrame goAwayFrame = new DefaultHttp2GoAwayFrame(Http2Error.NO_ERROR, ByteBufUtil
+ .writeAscii(ctx.alloc(), goAwayMessage));
+ goAwayFrame.setExtraStreamIds(Integer.MAX_VALUE);
+ ctx.write(goAwayFrame);
+ pingFuture = ctx.executor().schedule(
+ () -> secondGoAwayAndClose(ctx),
+ GRACEFUL_SHUTDOWN_PING_TIMEOUT_NANOS,
+ TimeUnit.NANOSECONDS);
+
+ Http2PingFrame pingFrame = new DefaultHttp2PingFrame(GRACEFUL_SHUTDOWN_PING, false);
+ ctx.write(pingFrame);
+ }
+
+ void secondGoAwayAndClose(ChannelHandlerContext ctx) {
+ if (pingAckedOrTimeout) {
+ return;
+ }
+ pingAckedOrTimeout = true;
+
+ pingFuture.cancel(false);
+
+ try {
+ Http2GoAwayFrame goAwayFrame = new DefaultHttp2GoAwayFrame(Http2Error.NO_ERROR, ByteBufUtil.writeAscii(this.ctx.alloc(), this.goAwayMessage));
+ ctx.write(goAwayFrame);
+ ctx.flush();
+ //TODO support customize graceful shutdown timeout mills
+ ctx.close(originPromise);
+ } catch (Exception e) {
+ ctx.fireExceptionCaught(e);
+ }
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcDataDecoder.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcDataDecoder.java
new file mode 100644
index 0000000..1ce1137
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcDataDecoder.java
@@ -0,0 +1,81 @@
+/*
+ *
+ * * 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.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ReplayingDecoder;
+
+import java.util.List;
+
+public class GrpcDataDecoder extends ReplayingDecoder<GrpcDataDecoder.GrpcDecodeState> {
+ private static final int RESERVED_MASK = 0xFE;
+ private static final int COMPRESSED_FLAG_MASK = 1;
+ private final int maxDataSize;
+
+ private int len;
+ private boolean compressedFlag;
+
+ public GrpcDataDecoder(int maxDataSize) {
+ super(GrpcDecodeState.HEADER);
+ this.maxDataSize = maxDataSize;
+ }
+
+ @Override
+ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+ switch (state()) {
+ case HEADER:
+ int type = in.readByte();
+ if ((type & RESERVED_MASK) != 0) {
+ throw GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withDescription("gRPC frame header malformed: reserved bits not zero")
+ .asException();
+ }
+ // compression is not supported yet
+ // TODO support it
+ compressedFlag = (type & COMPRESSED_FLAG_MASK) != 0;
+ if (compressedFlag) {
+ throw GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withDescription("Compression is not supported ")
+ .asException();
+ }
+
+ len = in.readInt();
+ if (len < 0 || len > maxDataSize) {
+ throw GrpcStatus.fromCode(GrpcStatus.Code.RESOURCE_EXHAUSTED)
+ .withDescription(String.format("gRPC message exceeds maximum size %d: %d",
+ maxDataSize, len))
+ .asException();
+ }
+ checkpoint(GrpcDecodeState.PAYLOAD);
+ case PAYLOAD:
+ ByteBuf buf = in.readRetainedSlice(len);
+ out.add(buf);
+ checkpoint(GrpcDecodeState.HEADER);
+ break;
+ default:
+ throw new RuntimeException("Should not reach here");
+ }
+ }
+
+ enum GrpcDecodeState {
+ HEADER, PAYLOAD
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcStatus.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcStatus.java
new file mode 100644
index 0000000..b8074eb
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcStatus.java
@@ -0,0 +1,87 @@
+/*
+ *
+ * * 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;
+
+/**
+ * See https://github.com/grpc/grpc/blob/master/doc/statuscodes.md
+ */
+
+public class GrpcStatus {
+ public static final GrpcStatus OK = new GrpcStatus(Code.OK, null, "OK");
+ public final Code code;
+ public final Throwable cause;
+ public final String description;
+
+ public GrpcStatus(Code code, Throwable cause, String description) {
+ this.code = code;
+ this.cause = cause;
+ this.description = description;
+ }
+
+ public static GrpcStatus fromCode(int code) {
+ return fromCode(Code.fromCode(code));
+ }
+
+ public static GrpcStatus fromCode(Code code) {
+ return new GrpcStatus(code, null, null);
+ }
+
+ public GrpcStatus withCause(Throwable cause) {
+ return new GrpcStatus(this.code, cause, this.description);
+ }
+
+ public GrpcStatus withDescription(String description) {
+ return new GrpcStatus(this.code, this.cause, description);
+ }
+
+ public TripleRpcException asException() {
+ return new TripleRpcException(this);
+ }
+
+ enum Code {
+ OK(0),
+ UNKNOWN(2),
+ DEADLINE_EXCEEDED(4),
+ NOT_FOUND(5),
+ RESOURCE_EXHAUSTED(8),
+ UNIMPLEMENTED(12),
+ INTERNAL(13);
+
+ final int code;
+
+ Code(int code) {
+ this.code = code;
+ }
+
+ public static boolean isOk(Integer status) {
+ return status == OK.code;
+ }
+
+ public static Code fromCode(int code) {
+ for (Code value : Code.values()) {
+ if (value.code == code) {
+ return value;
+ }
+ }
+ throw new IllegalStateException("Can not find status for code: " + code);
+ }
+ }
+
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/PathResolver.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/PathResolver.java
new file mode 100644
index 0000000..9b87129
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/PathResolver.java
@@ -0,0 +1,36 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.extension.SPI;
+import org.apache.dubbo.rpc.Invoker;
+
+@SPI(CommonConstants.TRIPLE)
+public interface PathResolver {
+
+ void add(String path, Invoker<?> invoker);
+
+ Invoker<?> resolve(String path);
+
+ void remove(String path);
+
+ void destroy();
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStream.java
new file mode 100644
index 0000000..8b2140f
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStream.java
@@ -0,0 +1,279 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
+import org.apache.dubbo.common.utils.ExecutorUtil;
+import org.apache.dubbo.remoting.TimeoutException;
+import org.apache.dubbo.rpc.AppResponse;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.MethodDescriptor;
+import org.apache.dubbo.rpc.model.ProviderModel;
+import org.apache.dubbo.rpc.model.ServiceDescriptor;
+import org.apache.dubbo.rpc.model.ServiceRepository;
+import org.apache.dubbo.rpc.protocol.tri.GrpcStatus.Code;
+import org.apache.dubbo.rpc.service.GenericService;
+import org.apache.dubbo.triple.TripleWrapper;
+
+import com.google.protobuf.Message;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
+import io.netty.handler.codec.http2.Http2Headers;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static org.apache.dubbo.rpc.protocol.tri.TripleUtil.responseErr;
+
+public class ServerStream extends AbstractStream implements Stream {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ServerStream.class);
+ private static final String TOO_MANY_REQ = "Too many requests";
+ private static final String MISSING_REQ = "Missing request";
+ private static final ExecutorRepository EXECUTOR_REPOSITORY =
+ ExtensionLoader.getExtensionLoader(ExecutorRepository.class).getDefaultExtension();
+ private final Invoker<?> invoker;
+ private final ChannelHandlerContext ctx;
+ private final ServiceDescriptor serviceDescriptor;
+ private final ProviderModel providerModel;
+ private final String methodName;
+ private MethodDescriptor methodDescriptor;
+
+
+ public ServerStream(Invoker<?> invoker, ServiceDescriptor serviceDescriptor, String methodName, ChannelHandlerContext ctx) {
+ super(ExecutorUtil.setThreadName(invoker.getUrl(), "DubboPUServerHandler"), ctx);
+ this.invoker = invoker;
+ ServiceRepository repo = ApplicationModel.getServiceRepository();
+ this.providerModel = repo.lookupExportedService(getUrl().getServiceKey());
+ this.methodName = methodName;
+ this.serviceDescriptor = serviceDescriptor;
+ this.ctx = ctx;
+ }
+
+
+ @Override
+ public void onError(GrpcStatus status) {
+ }
+
+ @Override
+ public void write(Object obj, ChannelPromise promise) throws Exception {
+
+ }
+
+ public void halfClose() throws Exception {
+ if (getData() == null) {
+ responseErr(ctx, GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withDescription(MISSING_REQ));
+ return;
+ }
+ ExecutorService executor = null;
+ if (providerModel != null) {
+ executor = (ExecutorService) providerModel.getServiceMetadata().getAttribute(CommonConstants.THREADPOOL_KEY);
+ }
+ if (executor == null) {
+ executor = EXECUTOR_REPOSITORY.getExecutor(getUrl());
+ }
+ if (executor == null) {
+ executor = EXECUTOR_REPOSITORY.createExecutorIfAbsent(getUrl());
+ }
+
+ try {
+ executor.execute(this::unaryInvoke);
+ } catch (RejectedExecutionException e) {
+ LOGGER.error("Provider's thread pool is full", e);
+ responseErr(ctx, GrpcStatus.fromCode(Code.RESOURCE_EXHAUSTED)
+ .withDescription("Provider's thread pool is full"));
+ } catch (Throwable t) {
+ LOGGER.error("Provider submit request to thread pool error ", t);
+ responseErr(ctx, GrpcStatus.fromCode(Code.INTERNAL)
+ .withCause(t)
+ .withDescription("Provider's error"));
+ }
+ }
+
+ private void unaryInvoke() {
+
+ Invocation invocation;
+ try {
+ invocation = buildInvocation();
+ } catch (Throwable t) {
+ LOGGER.warn("Exception processing triple message", t);
+ responseErr(ctx, GrpcStatus.fromCode(Code.INTERNAL).withDescription("Decode request failed:" + t.getMessage()));
+ return;
+ }
+ if (invocation == null) {
+ return;
+ }
+
+ final Result result = this.invoker.invoke(invocation);
+ CompletionStage<Object> future = result.thenApply(Function.identity());
+
+ BiConsumer<Object, Throwable> onComplete = (appResult, t) -> {
+ try {
+ if (t != null) {
+ if (t instanceof TimeoutException) {
+ responseErr(ctx, GrpcStatus.fromCode(Code.DEADLINE_EXCEEDED).withCause(t));
+ } else {
+ responseErr(ctx, GrpcStatus.fromCode(GrpcStatus.Code.UNKNOWN).withCause(t));
+ }
+ return;
+ }
+ AppResponse response = (AppResponse) appResult;
+ if (response.hasException()) {
+ final Throwable exception = response.getException();
+ if (exception instanceof TripleRpcException) {
+ responseErr(ctx, ((TripleRpcException) exception).getStatus());
+ } else {
+ responseErr(ctx, GrpcStatus.fromCode(GrpcStatus.Code.UNKNOWN)
+ .withCause(exception));
+ }
+ return;
+ }
+ Http2Headers http2Headers = new DefaultHttp2Headers()
+ .status(OK.codeAsText())
+ .set(HttpHeaderNames.CONTENT_TYPE, TripleConstant.CONTENT_PROTO);
+ final Message message;
+
+ ClassLoader tccl = Thread.currentThread().getContextClassLoader();
+
+ final ByteBuf buf;
+ try {
+ ClassLoadUtil.switchContextLoader(providerModel.getServiceInterfaceClass().getClassLoader());
+ if (isNeedWrap()) {
+ message = TripleUtil.wrapResp(getUrl(), getSerializeType(), response.getValue(), methodDescriptor, getMultipleSerialization());
+ } else {
+ message = (Message) response.getValue();
+ }
+ buf = TripleUtil.pack(ctx, message);
+ } finally {
+ ClassLoadUtil.switchContextLoader(tccl);
+ }
+
+ final Http2Headers trailers = new DefaultHttp2Headers()
+ .setInt(TripleConstant.STATUS_KEY, GrpcStatus.Code.OK.code);
+ final Map<String, Object> attachments = response.getObjectAttachments();
+ if (attachments != null) {
+ convertAttachment(trailers, attachments);
+ }
+ ctx.write(new DefaultHttp2HeadersFrame(http2Headers));
+ ctx.write(new DefaultHttp2DataFrame(buf));
+ ctx.writeAndFlush(new DefaultHttp2HeadersFrame(trailers, true));
+ } catch (Throwable e) {
+ LOGGER.warn("Exception processing triple message", e);
+ if (e instanceof TripleRpcException) {
+ responseErr(ctx, ((TripleRpcException) e).getStatus());
+ } else {
+ responseErr(ctx, GrpcStatus.fromCode(GrpcStatus.Code.UNKNOWN)
+ .withDescription("Exception occurred in provider's execution:" + e.getMessage())
+ .withCause(e));
+ }
+ }
+ };
+
+ future.whenComplete(onComplete);
+ }
+
+
+ private Invocation buildInvocation() {
+
+ RpcInvocation inv = new RpcInvocation();
+ ClassLoader tccl = Thread.currentThread().getContextClassLoader();
+ ServiceRepository repo = ApplicationModel.getServiceRepository();
+ final List<MethodDescriptor> methods = serviceDescriptor.getMethods(methodName);
+ if (methods == null || methods.isEmpty()) {
+ responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED)
+ .withDescription("Method not found:" + methodName + " of service:" + serviceDescriptor.getServiceName()));
+ return null;
+ }
+ if (methods.size() == 1) {
+ this.methodDescriptor = methods.get(0);
+ setNeedWrap(TripleUtil.needWrapper(this.methodDescriptor.getParameterClasses()));
+ } else {
+ // can not determine which one to invoke when same protobuf method name is used, force wrap it
+ setNeedWrap(true);
+ }
+ if (isNeedWrap()) {
+ loadFromURL(getUrl());
+ }
+
+ try {
+ if (providerModel != null) {
+ ClassLoadUtil.switchContextLoader(providerModel.getServiceInterfaceClass().getClassLoader());
+ }
+ if (isNeedWrap()) {
+ final TripleWrapper.TripleRequestWrapper req = TripleUtil.unpack(getData(), TripleWrapper.TripleRequestWrapper.class);
+ setSerializeType(req.getSerializeType());
+ if (CommonConstants.$INVOKE.equals(methodName) || CommonConstants.$INVOKE_ASYNC.equals(methodName)) {
+ this.methodDescriptor = repo.lookupMethod(GenericService.class.getName(), methodName);
+ } else {
+ String[] paramTypes = req.getArgTypesList().toArray(new String[req.getArgsCount()]);
+ for (MethodDescriptor method : methods) {
+ if (Arrays.equals(method.getCompatibleParamSignatures(), paramTypes)) {
+ this.methodDescriptor = method;
+ break;
+ }
+ }
+ if (this.methodDescriptor == null) {
+ responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED)
+ .withDescription("Method not found:" + methodName +
+ " args:" + Arrays.toString(paramTypes) + " of service:" + serviceDescriptor.getServiceName()));
+ return null;
+ }
+ }
+ final Object[] arguments = TripleUtil.unwrapReq(getUrl(), req, getMultipleSerialization());
+ inv.setArguments(arguments);
+ } else {
+
+ final Object req = TripleUtil.unpack(getData(), methodDescriptor.getParameterClasses()[0]);
+ inv.setArguments(new Object[]{req});
+ }
+ } finally {
+ ClassLoadUtil.switchContextLoader(tccl);
+ }
+ inv.setMethodName(methodDescriptor.getMethodName());
+ inv.setServiceName(serviceDescriptor.getServiceName());
+ inv.setTargetServiceUniqueName(getUrl().getServiceKey());
+ inv.setParameterTypes(methodDescriptor.getParameterClasses());
+ inv.setReturnTypes(methodDescriptor.getReturnTypes());
+ final Map<String, Object> attachments = parseHeadersToMap(getHeaders());
+ attachments.put(TripleConstant.TRI_CHANNEL_CTX_KEY, ctx);
+ inv.setObjectAttachments(attachments);
+ return inv;
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufSerialization.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufSerialization.java
new file mode 100644
index 0000000..2f9a20d
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufSerialization.java
@@ -0,0 +1,122 @@
+/*
+ *
+ * * 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 com.google.protobuf.CodedInputStream;
+import com.google.protobuf.ExtensionRegistryLite;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.reflect.InvocationTargetException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+public class SingleProtobufSerialization {
+ private static final ConcurrentHashMap<Class<?>, Message> instCache = new ConcurrentHashMap<>();
+ private static final ExtensionRegistryLite globalRegistry =
+ ExtensionRegistryLite.getEmptyRegistry();
+ private final ConcurrentMap<Class<?>, SingleMessageMarshaller<?>> marshallers = new ConcurrentHashMap<>();
+
+ @SuppressWarnings("all")
+ public static Message defaultInst(Class<?> clz) {
+ Message defaultInst = instCache.get(clz);
+ if (defaultInst != null) {
+ return defaultInst;
+ }
+ try {
+ defaultInst = (Message) clz.getMethod("getDefaultInstance").invoke(null);
+ } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
+ throw new RuntimeException("Create default protobuf instance failed ", e);
+ }
+ instCache.put(clz, defaultInst);
+ return defaultInst;
+ }
+
+ @SuppressWarnings("all")
+ public static <T> Parser<T> getParser(Class<T> clz) {
+ Message defaultInst = defaultInst(clz);
+ return (Parser<T>) defaultInst.getParserForType();
+ }
+
+ public Object deserialize(InputStream in, Class<?> clz) throws IOException {
+ try {
+ return getMarshaller(clz).parse(in);
+ } catch (InvalidProtocolBufferException e) {
+ throw new IOException(e);
+ }
+ }
+
+ public int serialize(Object obj, OutputStream os) throws IOException {
+ final MessageLite msg = (MessageLite) obj;
+ msg.writeTo(os);
+ return msg.getSerializedSize();
+ }
+
+ private SingleMessageMarshaller<?> getMarshaller(Class<?> clz) {
+ return marshallers.computeIfAbsent(clz, k -> new SingleMessageMarshaller(k));
+ }
+
+ private SingleMessageMarshaller<?> getMarshaller(Object obj) {
+ return getMarshaller(obj.getClass());
+ }
+
+ public static final class SingleMessageMarshaller<T extends MessageLite> {
+ private final Parser<T> parser;
+ private final T defaultInstance;
+
+ @SuppressWarnings("unchecked")
+ SingleMessageMarshaller(Class<T> clz) {
+ this.defaultInstance = (T) defaultInst(clz);
+ this.parser = (Parser<T>) defaultInstance.getParserForType();
+ }
+
+ @SuppressWarnings("unchecked")
+ public Class<T> getMessageClass() {
+ // Precisely T since protobuf doesn't let messages extend other messages.
+ return (Class<T>) defaultInstance.getClass();
+ }
+
+ public T getMessagePrototype() {
+ return defaultInstance;
+ }
+
+ public T parse(InputStream stream) throws InvalidProtocolBufferException {
+ return parser.parseFrom(stream, globalRegistry);
+ }
+
+ private T parseFrom(CodedInputStream stream) throws InvalidProtocolBufferException {
+ T message = parser.parseFrom(stream, globalRegistry);
+ try {
+ stream.checkLastTagWas(0);
+ return message;
+ } catch (InvalidProtocolBufferException e) {
+ e.setUnfinishedMessage(message);
+ throw e;
+ }
+ }
+ }
+
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/Stream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/Stream.java
new file mode 100644
index 0000000..22b7921
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/Stream.java
@@ -0,0 +1,38 @@
+/*
+ *
+ * * 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.channel.ChannelPromise;
+import io.netty.handler.codec.http2.Http2Headers;
+
+import java.io.InputStream;
+
+public interface Stream {
+
+ void onHeaders(Http2Headers headers);
+
+ void onData(InputStream in);
+
+ void onError(GrpcStatus status);
+
+ void write(Object obj, ChannelPromise promise) throws Exception;
+
+ void halfClose() throws Exception;
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleClientHandler.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleClientHandler.java
new file mode 100644
index 0000000..e009554
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleClientHandler.java
@@ -0,0 +1,68 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.URL;
+import org.apache.dubbo.remoting.api.ConnectionHandler;
+import org.apache.dubbo.remoting.exchange.Request;
+import org.apache.dubbo.rpc.RpcInvocation;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.http2.Http2GoAwayFrame;
+import io.netty.handler.codec.http2.Http2SettingsFrame;
+
+import java.io.IOException;
+
+public class TripleClientHandler extends ChannelDuplexHandler {
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+ if (msg instanceof Request) {
+ writeRequest(ctx, (Request) msg, promise);
+ } else {
+ super.write(ctx, msg, promise);
+ }
+ }
+
+ @Override
+ public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+ super.close(ctx, promise);
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof Http2SettingsFrame) {
+ // already handled
+ } else if (msg instanceof Http2GoAwayFrame) {
+ final ConnectionHandler connectionHandler = ctx.pipeline().get(ConnectionHandler.class);
+ connectionHandler.onGoAway(ctx.channel());
+ }
+ }
+
+ private void writeRequest(ChannelHandlerContext ctx, final Request req, ChannelPromise promise) throws IOException {
+ final RpcInvocation inv = (RpcInvocation) req.getData();
+ final boolean needWrapper = TripleUtil.needWrapper(inv.getParameterTypes());
+ final URL url = inv.getInvoker().getUrl();
+ ClientStream clientStream = new ClientStream(url, ctx, needWrapper, req);
+ clientStream.write(req, promise);
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleClientInboundHandler.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleClientInboundHandler.java
new file mode 100644
index 0000000..eb6e6a4
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleClientInboundHandler.java
@@ -0,0 +1,35 @@
+/*
+ *
+ * * 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.buffer.ByteBuf;
+import io.netty.buffer.ByteBufInputStream;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+public class TripleClientInboundHandler extends ChannelInboundHandlerAdapter {
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ final ClientStream invoker = TripleUtil.getClientStream(ctx);
+ if (invoker != null) {
+ invoker.onData(new ByteBufInputStream((ByteBuf) msg));
+ }
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleConstant.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleConstant.java
new file mode 100644
index 0000000..703ce91
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleConstant.java
@@ -0,0 +1,36 @@
+/*
+ *
+ * * 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;
+
+public interface TripleConstant {
+ String TRI_CHANNEL_CTX_KEY = "tri-ctx-channel";
+ String STATUS_KEY = "grpc-status";
+ String MESSAGE_KEY = "grpc-message";
+ String CONTENT_TYPE_KEY = "content-type";
+ String CONTENT_PROTO = "application/grpc+proto";
+ String APPLICATION_GRPC = "application/grpc";
+ String TRICE_ID_KEY = "tri-trace-traceid";
+ String RPC_ID_KEY = "tri-trace-rpcid";
+ String CONSUMER_APP_NAME_KEY = "tri-consumer-appname";
+ String UNIT_INFO_KEY = "tri-unit-info";
+ String SERVICE_VERSION = "tri-service-version";
+ String SERVICE_GROUP = "tri-service-group";
+
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2ClientResponseHandler.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2ClientResponseHandler.java
new file mode 100644
index 0000000..56e6ac2
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2ClientResponseHandler.java
@@ -0,0 +1,85 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http2.Http2DataFrame;
+import io.netty.handler.codec.http2.Http2GoAwayFrame;
+import io.netty.handler.codec.http2.Http2HeadersFrame;
+import io.netty.handler.codec.http2.Http2StreamFrame;
+
+public final class TripleHttp2ClientResponseHandler extends SimpleChannelInboundHandler<Http2StreamFrame> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(TripleHttp2ClientResponseHandler.class);
+
+ public TripleHttp2ClientResponseHandler() {
+ super(false);
+ }
+
+ @Override
+ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+ super.userEventTriggered(ctx, evt);
+ if (evt instanceof Http2GoAwayFrame) {
+ ctx.close();
+ }
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, Http2StreamFrame msg) throws Exception {
+ if (msg instanceof Http2HeadersFrame) {
+ onHeadersRead(ctx, (Http2HeadersFrame) msg);
+ } else if (msg instanceof Http2DataFrame) {
+ onDataRead(ctx, (Http2DataFrame) msg);
+ } else {
+ super.channelRead(ctx, msg);
+ }
+ }
+
+ private void onHeadersRead(ChannelHandlerContext ctx, Http2HeadersFrame msg) {
+ TripleUtil.getClientStream(ctx).onHeaders(msg.headers());
+ if (msg.isEndStream()) {
+ final ClientStream clientStream = TripleUtil.getClientStream(ctx);
+ clientStream.halfClose();
+ }
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ final ClientStream clientStream = TripleUtil.getClientStream(ctx);
+ final GrpcStatus status = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withCause(cause);
+ clientStream.onError(status);
+ ctx.close();
+ }
+
+ public void onDataRead(ChannelHandlerContext ctx, Http2DataFrame msg) throws Exception {
+ super.channelRead(ctx, msg.content());
+ if (msg.isEndStream()) {
+ final ClientStream clientStream = TripleUtil.getClientStream(ctx);
+ // stream already closed;
+ if (clientStream != null) {
+ clientStream.halfClose();
+ }
+ }
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameServerHandler.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameServerHandler.java
new file mode 100644
index 0000000..27b1b0d
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameServerHandler.java
@@ -0,0 +1,158 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.ServiceDescriptor;
+import org.apache.dubbo.rpc.model.ServiceRepository;
+import org.apache.dubbo.rpc.protocol.tri.GrpcStatus.Code;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http2.Http2DataFrame;
+import io.netty.handler.codec.http2.Http2Frame;
+import io.netty.handler.codec.http2.Http2Headers;
+import io.netty.handler.codec.http2.Http2HeadersFrame;
+
+import static org.apache.dubbo.rpc.protocol.tri.TripleUtil.responseErr;
+import static org.apache.dubbo.rpc.protocol.tri.TripleUtil.responsePlainTextError;
+
+public class TripleHttp2FrameServerHandler extends ChannelDuplexHandler {
+ private static final Logger LOGGER = LoggerFactory.getLogger(TripleHttp2FrameServerHandler.class);
+ private static final PathResolver PATH_RESOLVER = ExtensionLoader.getExtensionLoader(PathResolver.class).getDefaultExtension();
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof Http2HeadersFrame) {
+ onHeadersRead(ctx, (Http2HeadersFrame) msg);
+ } else if (msg instanceof Http2DataFrame) {
+ onDataRead(ctx, (Http2DataFrame) msg);
+ } else if (msg instanceof Http2Frame) {
+ // ignored
+ } else {
+ super.channelRead(ctx, msg);
+ }
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ if (LOGGER.isWarnEnabled()) {
+ LOGGER.warn("Exception in processing triple message", cause);
+ }
+ if (cause instanceof TripleRpcException) {
+ TripleUtil.responseErr(ctx, ((TripleRpcException) cause).getStatus());
+ } else {
+ TripleUtil.responseErr(ctx, GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withDescription("Provider's error:\n" + cause.getMessage()));
+ }
+ }
+
+ public void onDataRead(ChannelHandlerContext ctx, Http2DataFrame msg) throws Exception {
+ super.channelRead(ctx, msg.content());
+ if (msg.isEndStream()) {
+ final ServerStream serverStream = TripleUtil.getServerStream(ctx);
+ // stream already closed;
+ if (serverStream != null) {
+ serverStream.halfClose();
+ }
+ }
+ }
+
+ private Invoker<?> getInvoker(Http2Headers headers, String serviceName) {
+ final String version = headers.contains(TripleConstant.SERVICE_VERSION) ? headers.get(TripleConstant.SERVICE_VERSION).toString() : null;
+ final String group = headers.contains(TripleConstant.SERVICE_GROUP) ? headers.get(TripleConstant.SERVICE_GROUP).toString() : null;
+ final String key = URL.buildKey(serviceName, group, version);
+ Invoker<?> invoker = PATH_RESOLVER.resolve(key);
+ if (invoker == null) {
+ invoker = PATH_RESOLVER.resolve(serviceName);
+ }
+ return invoker;
+ }
+
+ public void onHeadersRead(ChannelHandlerContext ctx, Http2HeadersFrame msg) throws Exception {
+ final Http2Headers headers = msg.headers();
+
+ if (!HttpMethod.POST.asciiName().contentEquals(headers.method())) {
+ responsePlainTextError(ctx, HttpResponseStatus.METHOD_NOT_ALLOWED.code(), GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withDescription(String.format("Method '%s' is not supported", headers.method())));
+ return;
+ }
+
+ if (headers.path() == null) {
+ responsePlainTextError(ctx, HttpResponseStatus.NOT_FOUND.code(), GrpcStatus.fromCode(Code.UNIMPLEMENTED.code).withDescription("Expected path but is missing"));
+ return;
+ }
+
+ final String path = headers.path().toString();
+ if (path.charAt(0) != '/') {
+ responsePlainTextError(ctx, HttpResponseStatus.NOT_FOUND.code(), GrpcStatus.fromCode(Code.UNIMPLEMENTED.code)
+ .withDescription(String.format("Expected path to start with /: %s", path)));
+ return;
+ }
+
+ final CharSequence contentType = HttpUtil.getMimeType(headers.get(HttpHeaderNames.CONTENT_TYPE));
+ if (contentType == null) {
+ responsePlainTextError(ctx, HttpResponseStatus.UNSUPPORTED_MEDIA_TYPE.code(), GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL.code)
+ .withDescription("Content-Type is missing from the request"));
+ return;
+ }
+
+ final String contentString = contentType.toString();
+ if (!TripleUtil.supportContentType(contentString)) {
+ responsePlainTextError(ctx, HttpResponseStatus.UNSUPPORTED_MEDIA_TYPE.code(), GrpcStatus.fromCode(Code.INTERNAL.code)
+ .withDescription(String.format("Content-Type '%s' is not supported", contentString)));
+ return;
+ }
+
+ String[] parts = path.split("/");
+ String serviceName = parts[1];
+ String originalMethodName = parts[2];
+ String methodName = Character.toLowerCase(originalMethodName.charAt(0)) + originalMethodName.substring(1);
+
+ final Invoker<?> delegateInvoker = getInvoker(headers, serviceName);
+ if (delegateInvoker == null) {
+ responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED).withDescription("Service not found:" + serviceName));
+ return;
+ }
+ ServiceRepository repo = ApplicationModel.getServiceRepository();
+ final ServiceDescriptor descriptor = repo.lookupService(delegateInvoker.getUrl().getServiceKey());
+ if (descriptor == null) {
+ responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED).withDescription("Service not found:" + serviceName));
+ return;
+ }
+
+
+ final ServerStream serverStream = new ServerStream(delegateInvoker, descriptor, methodName, ctx);
+ serverStream.onHeaders(headers);
+ ctx.channel().attr(TripleUtil.SERVER_STREAM_KEY).set(serverStream);
+ if (msg.isEndStream()) {
+ serverStream.halfClose();
+ }
+ }
+}
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
new file mode 100644
index 0000000..617b010
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
@@ -0,0 +1,68 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.remoting.api.Http2WireProtocol;
+
+import io.netty.channel.ChannelInboundHandlerAdapter;
+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.Http2MultiplexHandler;
+import io.netty.handler.codec.http2.Http2Settings;
+import io.netty.handler.ssl.SslContext;
+
+@Activate
+public class TripleHttp2Protocol extends Http2WireProtocol {
+
+
+ @Override
+ public void close() {
+ super.close();
+ }
+
+ @Override
+ public void configServerPipeline(ChannelPipeline pipeline, SslContext sslContext) {
+ final Http2FrameCodec codec = Http2FrameCodecBuilder.forServer()
+ .gracefulShutdownTimeoutMillis(10000)
+ .initialSettings(new Http2Settings()
+ .maxHeaderListSize(8192)
+ .initialWindowSize(1048576))
+ .frameLogger(SERVER_LOGGER)
+ .build();
+ final Http2MultiplexHandler handler = new Http2MultiplexHandler(new TripleServerInitializer());
+ pipeline.addLast(codec, new TripleServerConnectionHandler(), handler);
+ }
+
+ @Override
+ public void configClientPipeline(ChannelPipeline pipeline, SslContext sslContext) {
+ final Http2FrameCodec codec = Http2FrameCodecBuilder.forClient()
+ .initialSettings(new Http2Settings()
+ .maxHeaderListSize(8192)
+ .initialWindowSize(1048576))
+ .gracefulShutdownTimeoutMillis(10000)
+ .frameLogger(CLIENT_LOGGER)
+ .build();
+ final Http2MultiplexHandler handler = new Http2MultiplexHandler(new ChannelInboundHandlerAdapter());
+ pipeline.addLast(codec, handler, new TripleClientHandler());
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
new file mode 100644
index 0000000..c874455
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
@@ -0,0 +1,182 @@
+/*
+ * 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 org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.Version;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.RemotingException;
+import org.apache.dubbo.remoting.TimeoutException;
+import org.apache.dubbo.remoting.api.Connection;
+import org.apache.dubbo.remoting.api.ConnectionManager;
+import org.apache.dubbo.remoting.exchange.Request;
+import org.apache.dubbo.remoting.exchange.Response;
+import org.apache.dubbo.remoting.exchange.support.DefaultFuture2;
+import org.apache.dubbo.rpc.AppResponse;
+import org.apache.dubbo.rpc.AsyncRpcResult;
+import org.apache.dubbo.rpc.FutureContext;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.TimeoutCountDown;
+import org.apache.dubbo.rpc.protocol.AbstractInvoker;
+import org.apache.dubbo.rpc.support.RpcUtils;
+
+import io.netty.channel.ChannelFuture;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT;
+import static org.apache.dubbo.common.constants.CommonConstants.ENABLE_TIMEOUT_COUNTDOWN_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_ATTACHMENT_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.TIME_COUNTDOWN_KEY;
+import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
+
+/**
+ * TripleInvoker
+ */
+public class TripleInvoker<T> extends AbstractInvoker<T> {
+
+ private static final ConnectionManager CONNECTION_MANAGER = ExtensionLoader.getExtensionLoader(ConnectionManager.class).getExtension("multiple");
+ private final Connection connection;
+ private final ReentrantLock destroyLock = new ReentrantLock();
+
+ private final Set<Invoker<?>> invokers;
+
+
+ public TripleInvoker(Class<T> serviceType, URL url, Set<Invoker<?>> invokers) throws RemotingException {
+ super(serviceType, url, new String[]{INTERFACE_KEY, GROUP_KEY, TOKEN_KEY});
+ this.invokers = invokers;
+ this.connection = CONNECTION_MANAGER.connect(url);
+ }
+
+ @Override
+ protected Result doInvoke(final Invocation invocation) throws Throwable {
+ RpcInvocation inv = (RpcInvocation) invocation;
+ final String methodName = RpcUtils.getMethodName(invocation);
+ inv.setAttachment(PATH_KEY, getUrl().getPath());
+ inv.setAttachment(Constants.SERIALIZATION_KEY, getUrl().getParameter(Constants.SERIALIZATION_KEY, Constants.DEFAULT_REMOTING_SERIALIZATION));
+ try {
+ int timeout = calculateTimeout(invocation, methodName);
+ invocation.put(TIMEOUT_KEY, timeout);
+ ExecutorService executor = getCallbackExecutor(getUrl(), inv);
+ // create request.
+ Request req = new Request();
+ req.setVersion(Version.getProtocolVersion());
+ req.setTwoWay(true);
+ req.setData(inv);
+ this.connection.init();
+
+ DefaultFuture2 future = DefaultFuture2.newFuture(this.connection, req, timeout, executor);
+ final CompletableFuture<AppResponse> respFuture = future.thenApply(obj -> (AppResponse) obj);
+ // save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter
+ FutureContext.getContext().setCompatibleFuture(respFuture);
+ AsyncRpcResult result = new AsyncRpcResult(respFuture, inv);
+ result.setExecutor(executor);
+
+ if (!connection.isAvailable()) {
+ Response response = new Response(req.getId(), req.getVersion());
+ response.setStatus(Response.CHANNEL_INACTIVE);
+ response.setErrorMessage(String.format("Connect to %s failed", this));
+ DefaultFuture2.received(connection, response);
+ } else {
+ final ChannelFuture writeFuture = this.connection.write(req);
+ writeFuture.addListener(future1 -> {
+ if (future1.isSuccess()) {
+ DefaultFuture2.sent(req);
+ } else {
+ Response response = new Response(req.getId(), req.getVersion());
+ response.setStatus(Response.CHANNEL_INACTIVE);
+ response.setErrorMessage(StringUtils.toString(future1.cause()));
+ DefaultFuture2.received(connection, response);
+ }
+ });
+ }
+
+ return result;
+ } catch (TimeoutException e) {
+ throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
+ } catch (RemotingException e) {
+ throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public boolean isAvailable() {
+ if (!super.isAvailable()) {
+ return false;
+ }
+ return connection.isAvailable();
+ }
+
+ @Override
+ public void destroy() {
+ // in order to avoid closing a client multiple times, a counter is used in case of connection per jvm, every
+ // time when client.close() is called, counter counts down once, and when counter reaches zero, client will be
+ // closed.
+ if (!super.isDestroyed()) {
+ // double check to avoid dup close
+ destroyLock.lock();
+ try {
+ if (super.isDestroyed()) {
+ return;
+ }
+ super.destroy();
+ if (invokers != null) {
+ invokers.remove(this);
+ }
+ try {
+ connection.release();
+ } catch (Throwable t) {
+ logger.warn(t.getMessage(), t);
+ }
+
+ } finally {
+ destroyLock.unlock();
+ }
+ }
+ }
+
+ private int calculateTimeout(Invocation invocation, String methodName) {
+ Object countdown = RpcContext.getContext().get(TIME_COUNTDOWN_KEY);
+ int timeout;
+ if (countdown == null) {
+ timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getContext(), DEFAULT_TIMEOUT);
+ if (getUrl().getParameter(ENABLE_TIMEOUT_COUNTDOWN_KEY, false)) {
+ invocation.setObjectAttachment(TIMEOUT_ATTACHMENT_KEY, timeout); // pass timeout to remote server
+ }
+ } else {
+ TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countdown;
+ timeout = (int) timeoutCountDown.timeRemaining(TimeUnit.MILLISECONDS);
+ invocation.setObjectAttachment(TIMEOUT_ATTACHMENT_KEY, timeout);// pass timeout to remote server
+ }
+ return timeout;
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TriplePathResolver.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TriplePathResolver.java
new file mode 100644
index 0000000..d57b140
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TriplePathResolver.java
@@ -0,0 +1,54 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.rpc.Invoker;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+public class TriplePathResolver implements PathResolver {
+ private static final Logger logger = LoggerFactory.getLogger(TriplePathResolver.class);
+
+ private final ConcurrentHashMap<String, Invoker<?>> path2Invoker = new ConcurrentHashMap<>();
+
+ @Override
+ public void add(String path, Invoker<?> invoker) {
+ path2Invoker.put(path, invoker);
+ }
+
+ @Override
+ public Invoker<?> resolve(String path) {
+ return path2Invoker.get(path);
+ }
+
+ @Override
+ public void remove(String path) {
+ path2Invoker.remove(path);
+ }
+
+ @Override
+ public void destroy() {
+ path2Invoker.clear();
+ }
+
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java
new file mode 100644
index 0000000..4d6f16c
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java
@@ -0,0 +1,107 @@
+/*
+ * 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 org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.remoting.RemotingException;
+import org.apache.dubbo.remoting.exchange.PortUnificationExchanger;
+import org.apache.dubbo.rpc.Exporter;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Protocol;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.protocol.AbstractExporter;
+import org.apache.dubbo.rpc.protocol.AbstractProtocol;
+
+import java.util.ArrayList;
+
+/**
+ *
+ */
+public class TripleProtocol extends AbstractProtocol implements Protocol {
+
+ private static final Logger logger = LoggerFactory.getLogger(TripleProtocol.class);
+ private final PathResolver pathResolver = ExtensionLoader.getExtensionLoader(PathResolver.class).getDefaultExtension();
+
+ @Override
+ public int getDefaultPort() {
+ return 50051;
+ }
+
+ @Override
+ public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
+ URL url = invoker.getUrl();
+ String key = serviceKey(url);
+ final AbstractExporter<T> exporter = new AbstractExporter<T>(invoker) {
+ @Override
+ public void unexport() {
+ super.unexport();
+ pathResolver.remove(url.getServiceKey());
+ pathResolver.remove(url.getServiceInterface());
+ exporterMap.remove(key);
+ }
+ };
+
+ exporterMap.put(key, exporter);
+
+ invokers.add(invoker);
+
+ pathResolver.add(url.getServiceKey(), invoker);
+ pathResolver.add(url.getServiceInterface(), invoker);
+ PortUnificationExchanger.bind(invoker.getUrl());
+ return exporter;
+ }
+
+ @Override
+ public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
+ TripleInvoker<T> invoker;
+ try {
+ invoker = new TripleInvoker<T>(type, url, invokers);
+ } catch (RemotingException e) {
+ throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
+ }
+ invokers.add(invoker);
+ return invoker;
+ }
+
+ @Override
+ protected <T> Invoker<T> protocolBindingRefer(Class<T> type, URL url) throws RpcException {
+ return null;
+ }
+
+ @Override
+ public void destroy() {
+ PortUnificationExchanger.close();
+ pathResolver.destroy();
+ for (String key : new ArrayList<>(exporterMap.keySet())) {
+ Exporter<?> exporter = exporterMap.remove(key);
+ if (exporter != null) {
+ try {
+ if (logger.isInfoEnabled()) {
+ logger.info("Unexport service: " + exporter.getInvoker().getUrl());
+ }
+ exporter.unexport();
+ } catch (Throwable t) {
+ logger.warn(t.getMessage(), t);
+ }
+ }
+ }
+
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleRpcException.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleRpcException.java
new file mode 100644
index 0000000..0056bdd
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleRpcException.java
@@ -0,0 +1,35 @@
+/*
+ *
+ * * 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 org.apache.dubbo.rpc.RpcException;
+
+public class TripleRpcException extends RpcException {
+ private final GrpcStatus status;
+
+ public TripleRpcException(GrpcStatus status) {
+ super(status.description, status.cause);
+ this.status = status;
+ }
+
+ public GrpcStatus getStatus() {
+ return status;
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerConnectionHandler.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerConnectionHandler.java
new file mode 100644
index 0000000..1e4369e
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerConnectionHandler.java
@@ -0,0 +1,68 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.http2.Http2ChannelDuplexHandler;
+import io.netty.handler.codec.http2.Http2GoAwayFrame;
+import io.netty.handler.codec.http2.Http2PingFrame;
+
+import static org.apache.dubbo.rpc.protocol.tri.GracefulShutdown.GRACEFUL_SHUTDOWN_PING;
+
+public class TripleServerConnectionHandler extends Http2ChannelDuplexHandler {
+ private static final Logger logger = LoggerFactory.getLogger(TripleServerConnectionHandler.class);
+ private GracefulShutdown gracefulShutdown;
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof Http2PingFrame) {
+ if (((Http2PingFrame) msg).content() == GRACEFUL_SHUTDOWN_PING) {
+ if (gracefulShutdown == null) {
+ // this should never happen
+ logger.warn("Received GRACEFUL_SHUTDOWN_PING Ack but gracefulShutdown is null");
+ } else {
+ gracefulShutdown.secondGoAwayAndClose(ctx);
+ }
+ }
+ } else if (msg instanceof Http2GoAwayFrame) {
+ } else {
+ super.channelRead(ctx, msg);
+ }
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ logger.warn(String.format("Channel:%s Error", ctx.channel()), cause);
+ ctx.close();
+ }
+
+ @Override
+ public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+ if (gracefulShutdown == null) {
+ gracefulShutdown = new GracefulShutdown(ctx, "app_requested", promise);
+ }
+ gracefulShutdown.gracefulShutdown();
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerInboundHandler.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerInboundHandler.java
new file mode 100644
index 0000000..98ec646
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerInboundHandler.java
@@ -0,0 +1,35 @@
+/*
+ *
+ * * 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.buffer.ByteBuf;
+import io.netty.buffer.ByteBufInputStream;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+public class TripleServerInboundHandler extends ChannelInboundHandlerAdapter {
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ final ServerStream serverStream = TripleUtil.getServerStream(ctx);
+ if (serverStream != null) {
+ serverStream.onData(new ByteBufInputStream((ByteBuf) msg));
+ }
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerInitializer.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerInitializer.java
new file mode 100644
index 0000000..cb002f7
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerInitializer.java
@@ -0,0 +1,36 @@
+/*
+ *
+ * * 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.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+
+public class TripleServerInitializer extends ChannelInitializer<Channel> {
+
+ @Override
+ protected void initChannel(Channel ch) throws Exception {
+ final ChannelPipeline p = ch.pipeline();
+ p.addLast(new TripleHttp2FrameServerHandler());
+ // TODO constraint MAX DATA_SIZE
+ p.addLast(new GrpcDataDecoder(Integer.MAX_VALUE));
+ p.addLast(new TripleServerInboundHandler());
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleUtil.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleUtil.java
new file mode 100644
index 0000000..1a1fd0c
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleUtil.java
@@ -0,0 +1,309 @@
+/*
+ *
+ * * 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 org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.serialize.MultipleSerialization;
+import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.model.MethodDescriptor;
+import org.apache.dubbo.triple.TripleWrapper;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufOutputStream;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.codec.http.QueryStringEncoder;
+import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
+import io.netty.handler.codec.http2.Http2Headers;
+import io.netty.util.AttributeKey;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+
+public class TripleUtil {
+
+
+ public static final AttributeKey<ServerStream> SERVER_STREAM_KEY = AttributeKey.newInstance("tri_server_stream");
+ public static final AttributeKey<ClientStream> CLIENT_STREAM_KEY = AttributeKey.newInstance("tri_client_stream");
+ private static final SingleProtobufSerialization pbSerialization = new SingleProtobufSerialization();
+ private static final Base64.Decoder BASE64_DECODER = Base64.getDecoder();
+ private static final Base64.Encoder BASE64_ENCODER = Base64.getEncoder().withoutPadding();
+
+ public static ServerStream getServerStream(ChannelHandlerContext ctx) {
+ return ctx.channel().attr(TripleUtil.SERVER_STREAM_KEY).get();
+ }
+
+ public static void setClientStream(Channel channel, ClientStream clientStream) {
+ channel.attr(TripleUtil.CLIENT_STREAM_KEY).set(clientStream);
+ }
+
+ public static void setClientStream(ChannelHandlerContext ctx, ClientStream clientStream) {
+ setClientStream(ctx.channel(), clientStream);
+ }
+
+ public static ClientStream getClientStream(ChannelHandlerContext ctx) {
+ return ctx.channel().attr(TripleUtil.CLIENT_STREAM_KEY).get();
+ }
+
+ /**
+ * must starts from application/grpc
+ */
+ public static boolean supportContentType(String contentType) {
+ if (contentType == null) {
+ return false;
+ }
+ return contentType.startsWith(TripleConstant.APPLICATION_GRPC);
+ }
+
+ public static void responseErr(ChannelHandlerContext ctx, GrpcStatus status) {
+ Http2Headers trailers = new DefaultHttp2Headers()
+ .status(OK.codeAsText())
+ .set(HttpHeaderNames.CONTENT_TYPE, TripleConstant.CONTENT_PROTO)
+ .setInt(TripleConstant.STATUS_KEY, status.code.code)
+ .set(TripleConstant.MESSAGE_KEY, getErrorMsg(status));
+ ctx.write(new DefaultHttp2HeadersFrame(trailers, true));
+ }
+
+ public static String getErrorMsg(GrpcStatus status) {
+ final String msg;
+ if (status.cause == null) {
+ msg = status.description;
+ } else {
+ msg = StringUtils.toString(status.description, status.cause);
+ }
+ return percentEncode(msg);
+ }
+
+ public static String limitSizeTo4KB(String desc) {
+ if (desc.length() < 4096) {
+ return desc;
+ } else {
+ return desc.substring(0, 4086);
+ }
+ }
+
+ public static String percentDecode(CharSequence corpus) {
+ if (corpus == null) {
+ return "";
+ }
+ QueryStringDecoder decoder = new QueryStringDecoder("?=" + corpus);
+ for (Map.Entry<String, List<String>> e : decoder.parameters().entrySet()) {
+ return e.getKey();
+ }
+ return "";
+ }
+
+ public static String percentEncode(String corpus) {
+ if (corpus == null) {
+ return "";
+ }
+ corpus = limitSizeTo4KB(corpus);
+ QueryStringEncoder encoder = new QueryStringEncoder("");
+ encoder.addParam("", corpus);
+ // ?=
+ return encoder.toString().substring(2);
+ }
+
+ public static void responsePlainTextError(ChannelHandlerContext ctx, int code, GrpcStatus status) {
+ Http2Headers headers = new DefaultHttp2Headers(true)
+ .status("" + code)
+ .setInt(TripleConstant.STATUS_KEY, status.code.code)
+ .set(TripleConstant.MESSAGE_KEY, status.description)
+ .set(TripleConstant.CONTENT_TYPE_KEY, "text/plain; encoding=utf-8");
+ ctx.write(new DefaultHttp2HeadersFrame(headers));
+ ByteBuf buf = ByteBufUtil.writeUtf8(ctx.alloc(), status.description);
+ ctx.write(new DefaultHttp2DataFrame(buf, true));
+ }
+
+ public static boolean needWrapper(Class<?>[] parameterTypes) {
+ if (parameterTypes.length != 1) {
+ return true;
+ }
+ return !Message.class.isAssignableFrom(parameterTypes[0]);
+ }
+
+ public static Object unwrapResp(URL url, TripleWrapper.TripleResponseWrapper wrap, MultipleSerialization serialization) {
+ String serializeType = convertHessianFromWrapper(wrap.getSerializeType());
+ try {
+ final ByteArrayInputStream bais = new ByteArrayInputStream(wrap.getData().toByteArray());
+ return serialization.deserialize(url, serializeType, wrap.getType(), bais);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to unwrap resp", e);
+ }
+ }
+
+ public static Object[] unwrapReq(URL url, TripleWrapper.TripleRequestWrapper wrap, MultipleSerialization multipleSerialization) {
+ String serializeType = convertHessianFromWrapper(wrap.getSerializeType());
+ try {
+ Object[] arguments = new Object[wrap.getArgsCount()];
+ for (int i = 0; i < arguments.length; i++) {
+ final ByteArrayInputStream bais = new ByteArrayInputStream(wrap.getArgs(i).toByteArray());
+ Object obj = multipleSerialization.deserialize(url,
+ serializeType, wrap.getArgTypes(i), bais);
+ arguments[i] = obj;
+ }
+ return arguments;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to unwrap req: " + e.getMessage(), e);
+ }
+ }
+
+ public static TripleWrapper.TripleResponseWrapper wrapResp(URL url, String serializeType, Object resp, MethodDescriptor desc,
+ MultipleSerialization multipleSerialization) {
+ try {
+ final TripleWrapper.TripleResponseWrapper.Builder builder = TripleWrapper.TripleResponseWrapper.newBuilder()
+ .setType(desc.getReturnClass().getName())
+ .setSerializeType(convertHessianToWrapper(serializeType));
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ multipleSerialization.serialize(url, serializeType, desc.getReturnClass().getName(), resp, bos);
+ builder.setData(ByteString.copyFrom(bos.toByteArray()));
+ return builder.build();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to pack wrapper req", e);
+ }
+ }
+
+ public static <T> T unpack(InputStream is, Class<T> clz) {
+ try {
+ final T req = (T) pbSerialization.deserialize(is, clz);
+ return req;
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to unpack req", e);
+ }finally {
+ closeQuietly(is);
+ }
+ }
+ private static void closeQuietly(Closeable c) {
+ if (c != null) {
+ try {
+ c.close();
+ } catch (IOException ignore) {
+ // ignored
+ }
+ }
+ }
+
+
+ public static ByteBuf pack(ChannelHandlerContext ctx, Object obj) {
+ try {
+ final ByteBuf buf = ctx.alloc().buffer();
+ buf.writeByte(0);
+ buf.writeInt(0);
+ final ByteBufOutputStream bos = new ByteBufOutputStream(buf);
+ final int size = pbSerialization.serialize(obj, bos);
+ buf.setInt(1, size);
+ return buf;
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to pack req", e);
+ }
+ }
+
+ public static TripleWrapper.TripleRequestWrapper wrapReq(URL url, RpcInvocation invocation, MultipleSerialization serialization) {
+ try {
+ String serializationName = (String) invocation.getObjectAttachment(Constants.SERIALIZATION_KEY);
+ final TripleWrapper.TripleRequestWrapper.Builder builder = TripleWrapper.TripleRequestWrapper.newBuilder()
+ .setSerializeType(convertHessianToWrapper(serializationName));
+ for (int i = 0; i < invocation.getArguments().length; i++) {
+ final String clz = invocation.getParameterTypes()[i].getName();
+ builder.addArgTypes(clz);
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ serialization.serialize(url, serializationName, clz, invocation.getArguments()[i], bos);
+ builder.addArgs(ByteString.copyFrom(bos.toByteArray()));
+ }
+ return builder.build();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to pack wrapper req", e);
+ }
+ }
+
+ public static String encodeWrapper(URL url, Object obj, String serializeType, MultipleSerialization serialization) throws IOException {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ serialization.serialize(url, serializeType, obj.getClass().getName(), obj, bos);
+ final TripleWrapper.TripleRequestWrapper wrap = TripleWrapper.TripleRequestWrapper.newBuilder()
+ .setSerializeType(convertHessianToWrapper(serializeType))
+ .addArgTypes(obj.getClass().getName())
+ .addArgs(ByteString.copyFrom(bos.toByteArray()))
+ .build();
+ return encodeBase64ASCII(wrap.toByteArray());
+ }
+
+ public static String encodeBase64ASCII(byte[] in) {
+ byte[] bytes = encodeBase64(in);
+ return new String(bytes, StandardCharsets.US_ASCII);
+ }
+
+ public static byte[] encodeBase64(byte[] in) {
+ return BASE64_ENCODER.encode(in);
+ }
+
+ public static byte[] decodeBase64(byte[] in) {
+ return BASE64_DECODER.decode(in);
+ }
+
+ public static byte[] decodeBase64(String in) {
+ return BASE64_DECODER.decode(in.getBytes(StandardCharsets.UTF_8));
+ }
+
+ public static Object decodeObjFromHeader(URL url, CharSequence value, MultipleSerialization serialization) throws InvalidProtocolBufferException {
+ final byte[] decode = decodeASCIIByte(value);
+ final TripleWrapper.TripleRequestWrapper wrapper = TripleWrapper.TripleRequestWrapper.parseFrom(decode);
+ final Object[] objects = TripleUtil.unwrapReq(url, wrapper, serialization);
+ return objects[0];
+ }
+
+ public static byte[] decodeASCIIByte(CharSequence value) {
+ return BASE64_DECODER.decode(value.toString().getBytes(StandardCharsets.US_ASCII));
+ }
+
+ public static String convertHessianToWrapper(String serializeType) {
+ if (serializeType.equals("hessian2")) {
+ return "hessian4";
+ }
+ return serializeType;
+ }
+
+ public static String convertHessianFromWrapper(String serializeType) {
+ if (serializeType.equals("hessian4")) {
+ return "hessian2";
+ }
+ return serializeType;
+ }
+
+
+}
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/proto/triple_wrapper.proto b/dubbo-rpc/dubbo-rpc-triple/src/main/proto/triple_wrapper.proto
new file mode 100644
index 0000000..a2d8703
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/proto/triple_wrapper.proto
@@ -0,0 +1,17 @@
+syntax = "proto3";
+
+package org.apache.dubbo.triple;
+
+message TripleRequestWrapper {
+ // hessian4
+ // json
+ string serializeType = 1;
+ repeated bytes args = 2;
+ repeated string argTypes = 3;
+}
+
+message TripleResponseWrapper {
+ string serializeType = 1;
+ bytes data = 2;
+ string type = 3;
+}
\ No newline at end of file
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.api.WireProtocol b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.api.WireProtocol
new file mode 100644
index 0000000..2280d84
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.api.WireProtocol
@@ -0,0 +1 @@
+tri=org.apache.dubbo.rpc.protocol.tri.TripleHttp2Protocol
\ No newline at end of file
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol
new file mode 100644
index 0000000..cdcb95f
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol
@@ -0,0 +1 @@
+tri=org.apache.dubbo.rpc.protocol.tri.TripleProtocol
\ No newline at end of file
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.protocol.tri.PathResolver b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.protocol.tri.PathResolver
new file mode 100644
index 0000000..89232d5
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.protocol.tri.PathResolver
@@ -0,0 +1 @@
+tri=org.apache.dubbo.rpc.protocol.tri.TriplePathResolver
\ No newline at end of file
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleUtilTest.java b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleUtilTest.java
new file mode 100644
index 0000000..16186da
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleUtilTest.java
@@ -0,0 +1,34 @@
+/*
+ *
+ * * 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 org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TripleUtilTest {
+
+ @Test
+ void percentEncoding() {
+ String content="\t\ntest with whitespace\r\nand Unicode BMP ☺ and non-BMP 😈\t\n";
+ final String encoded = TripleUtil.percentEncode(content);
+ final String decoded = TripleUtil.percentDecode(encoded);
+ Assertions.assertEquals(content,decoded);
+ }
+}
\ No newline at end of file
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.telnet.TelnetHandler b/dubbo-rpc/dubbo-rpc-triple/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.telnet.TelnetHandler
new file mode 100644
index 0000000..60d4bee
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.telnet.TelnetHandler
@@ -0,0 +1 @@
+test=org.apache.dubbo.rpc.protocol.dubbo.decode.telnet.TestTelnetHandler
\ No newline at end of file
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/test/resources/log4j.xml b/dubbo-rpc/dubbo-rpc-triple/src/test/resources/log4j.xml
new file mode 100644
index 0000000..09dba05
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/test/resources/log4j.xml
@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+ -->
+<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
+<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
+ <!-- ===================================================================== -->
+ <!-- 以下是appender的定义 -->
+ <!-- ===================================================================== -->
+ <appender name="dubbo" class="org.apache.dubbo.common.utils.DubboAppender">
+ <param name="File" value="../../dubbo.log"/>
+ <param name="encoding" value="GBK"/>
+ <layout class="org.apache.log4j.PatternLayout">
+ <param name="ConversionPattern" value="%d %p [%c:%M] - %m%n"/>
+ </layout>
+ <!-- <filter class="org.apache.log4j.varia.LevelRangeFilter">
+ <param name="LevelMin" value="DEBUG" />
+ <param name="LevelMax" value="DEBUG" />
+ </filter> -->
+ </appender>
+ <root>
+ <level value="INFO"/>
+ <appender-ref ref="dubbo"/>
+ </root>
+</log4j:configuration>
diff --git a/dubbo-rpc/pom.xml b/dubbo-rpc/pom.xml
index b028343..ef384d5 100644
--- a/dubbo-rpc/pom.xml
+++ b/dubbo-rpc/pom.xml
@@ -35,5 +35,6 @@
<module>dubbo-rpc-injvm</module>
<module>dubbo-rpc-rest</module>
<module>dubbo-rpc-grpc</module>
+ <module>dubbo-rpc-triple</module>
</modules>
</project>
diff --git a/dubbo-serialization/dubbo-serialization-api/src/main/java/org/apache/dubbo/common/serialize/Constants.java b/dubbo-serialization/dubbo-serialization-api/src/main/java/org/apache/dubbo/common/serialize/Constants.java
index 66f152a..3715437 100644
--- a/dubbo-serialization/dubbo-serialization-api/src/main/java/org/apache/dubbo/common/serialize/Constants.java
+++ b/dubbo-serialization/dubbo-serialization-api/src/main/java/org/apache/dubbo/common/serialize/Constants.java
@@ -33,4 +33,5 @@ public interface Constants {
byte PROTOBUF_SERIALIZATION_ID = 22;
byte KRYO_SERIALIZATION2_ID = 25;
+ byte CUSTOM_MESSAGE_PACK_ID = 31;
}
diff --git a/dubbo-serialization/dubbo-serialization-api/src/main/java/org/apache/dubbo/common/serialize/DefaultMultipleSerialization.java b/dubbo-serialization/dubbo-serialization-api/src/main/java/org/apache/dubbo/common/serialize/DefaultMultipleSerialization.java
new file mode 100644
index 0000000..22f891f
--- /dev/null
+++ b/dubbo-serialization/dubbo-serialization-api/src/main/java/org/apache/dubbo/common/serialize/DefaultMultipleSerialization.java
@@ -0,0 +1,56 @@
+/*
+ *
+ * * 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.common.serialize;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.utils.ClassUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+public class DefaultMultipleSerialization implements MultipleSerialization {
+
+ @Override
+ public void serialize(URL url, String serializeType, String clz, Object obj, OutputStream os) throws IOException {
+ serializeType = convertHessian(serializeType);
+ final Serialization serialization = ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(serializeType);
+ final ObjectOutput serialize = serialization.serialize(null, os);
+ serialize.writeObject(obj);
+ serialize.flushBuffer();
+ }
+
+ @Override
+ public Object deserialize(URL url, String serializeType, String clz, InputStream os) throws IOException, ClassNotFoundException {
+ serializeType = convertHessian(serializeType);
+ final Serialization serialization = ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(serializeType);
+ final Class<?> aClass = ClassUtils.forName(clz);
+ final ObjectInput in = serialization.deserialize(null, os);
+ return in.readObject(aClass);
+ }
+
+ private String convertHessian(String ser) {
+ if (ser.equals("hessian4")) {
+ return "hessian2";
+ }
+ return ser;
+ }
+}
diff --git a/dubbo-serialization/dubbo-serialization-api/src/main/java/org/apache/dubbo/common/serialize/MultipleSerialization.java b/dubbo-serialization/dubbo-serialization-api/src/main/java/org/apache/dubbo/common/serialize/MultipleSerialization.java
new file mode 100644
index 0000000..e4b2aa5
--- /dev/null
+++ b/dubbo-serialization/dubbo-serialization-api/src/main/java/org/apache/dubbo/common/serialize/MultipleSerialization.java
@@ -0,0 +1,36 @@
+/*
+ *
+ * * 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.common.serialize;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.SPI;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+@SPI
+public interface MultipleSerialization {
+
+ void serialize(URL url, String serializeType, String clz, Object obj, OutputStream os) throws IOException;
+
+ Object deserialize(URL url, String serializeType, String clz, InputStream os) throws IOException, ClassNotFoundException;
+
+}
diff --git a/dubbo-serialization/dubbo-serialization-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.serialize.MultipleSerialization b/dubbo-serialization/dubbo-serialization-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.serialize.MultipleSerialization
new file mode 100644
index 0000000..3e0db4c
--- /dev/null
+++ b/dubbo-serialization/dubbo-serialization-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.serialize.MultipleSerialization
@@ -0,0 +1 @@
+default=org.apache.dubbo.common.serialize.DefaultMultipleSerialization
\ No newline at end of file