natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'apache-3.2' into apache-3.3 

# Conflicts:
#	dubbo-cluster/pom.xml
#	dubbo-dependencies-bom/pom.xml
#	dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java
This commit is contained in:
Albumen Kevin 2023-10-18 19:02:54 +08:00
parent 7696e1b42c fa11b68a16
commit 0562b64828
36 changed files with 303 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
dubbo-cluster/pom.xml
View File

@ -29,6 +29,7 @@
<description>The cluster module of dubbo project</description>
<properties>
<skip_maven_deploy>false</skip_maven_deploy>
<nashorn-core.version>15.4</nashorn-core.version>
</properties>
<dependencies>
<dependency>
@ -86,5 +87,11 @@
<version>${project.parent.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.openjdk.nashorn</groupId>
<artifactId>nashorn-core</artifactId>
<version>${nashorn-core.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

7
dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java
View File

@ -26,6 +26,7 @@ import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadlocal.InternalThreadLocalMap;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_ERROR_RUN_THREAD_TASK;
import static org.apache.dubbo.common.utils.ExecutorUtil.isShutdown;
/**
* Executor ensuring that all {@link Runnable} tasks submitted are executed in order
@ -68,8 +69,10 @@ public final class SerializingExecutor implements Executor, Runnable {
if (atomicBoolean.compareAndSet(false, true)) {
boolean success = false;
try {
executor.execute(this);
success = true;
if (!isShutdown(executor)) {
executor.execute(this);
success = true;
}
} finally {
// It is possible that at this point that there are still tasks in
// the queue, it would be nice to keep trying but the error may not

17
dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/support/cached/CachedThreadPool.java
View File

@ -22,6 +22,7 @@ import org.apache.dubbo.common.threadpool.MemorySafeLinkedBlockingQueue;
import org.apache.dubbo.common.threadpool.ThreadPool;
import org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
@ -53,10 +54,18 @@ public class CachedThreadPool implements ThreadPool {
int threads = url.getParameter(THREADS_KEY, Integer.MAX_VALUE);
int queues = url.getParameter(QUEUES_KEY, DEFAULT_QUEUES);
int alive = url.getParameter(ALIVE_KEY, DEFAULT_ALIVE);
return new ThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS,
queues == 0 ? new SynchronousQueue<Runnable>() :
(queues < 0 ? new MemorySafeLinkedBlockingQueue<Runnable>()
: new LinkedBlockingQueue<Runnable>(queues)),
BlockingQueue<Runnable> blockingQueue;
if (queues == 0) {
blockingQueue = new SynchronousQueue<>();
} else if (queues < 0) {
blockingQueue = new MemorySafeLinkedBlockingQueue<>();
} else {
blockingQueue = new LinkedBlockingQueue<>(queues);
}
return new ThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS, blockingQueue,
new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}
}

17
dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/support/limited/LimitedThreadPool.java
View File

@ -23,6 +23,7 @@ import org.apache.dubbo.common.threadpool.MemorySafeLinkedBlockingQueue;
import org.apache.dubbo.common.threadpool.ThreadPool;
import org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
@ -50,10 +51,18 @@ public class LimitedThreadPool implements ThreadPool {
int cores = url.getParameter(CORE_THREADS_KEY, DEFAULT_CORE_THREADS);
int threads = url.getParameter(THREADS_KEY, DEFAULT_THREADS);
int queues = url.getParameter(QUEUES_KEY, DEFAULT_QUEUES);
return new ThreadPoolExecutor(cores, threads, Long.MAX_VALUE, TimeUnit.MILLISECONDS,
queues == 0 ? new SynchronousQueue<Runnable>() :
(queues < 0 ? new MemorySafeLinkedBlockingQueue<Runnable>()
: new LinkedBlockingQueue<Runnable>(queues)),
BlockingQueue<Runnable> blockingQueue;
if (queues == 0) {
blockingQueue = new SynchronousQueue<>();
} else if (queues < 0) {
blockingQueue = new MemorySafeLinkedBlockingQueue<>();
} else {
blockingQueue = new LinkedBlockingQueue<>(queues);
}
return new ThreadPoolExecutor(cores, threads, Long.MAX_VALUE, TimeUnit.MILLISECONDS, blockingQueue,
new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}

15
dubbo-common/src/main/java/org/apache/dubbo/common/utils/ExecutorUtil.java
View File

@ -38,12 +38,17 @@ public class ExecutorUtil {
new NamedThreadFactory("Close-ExecutorService-Timer", true));
public static boolean isTerminated(Executor executor) {
if (executor instanceof ExecutorService) {
if (((ExecutorService) executor).isTerminated()) {
return true;
}
if (!(executor instanceof ExecutorService)) {
return false;
}
return false;
return ((ExecutorService) executor).isTerminated();
}
public static boolean isShutdown(Executor executor) {
if (!(executor instanceof ExecutorService)) {
return false;
}
return ((ExecutorService) executor).isShutdown();
}
/**

1
dubbo-common/src/main/java/org/apache/dubbo/rpc/executor/AbstractIsolationExecutorSupport.java
View File

@ -36,6 +36,7 @@ public abstract class AbstractIsolationExecutorSupport implements ExecutorSuppor
this.frameworkServiceRepository = url.getOrDefaultFrameworkModel().getServiceRepository();
}
@Override
public Executor getExecutor(Object data) {
ProviderModel providerModel = getProviderModel(data);

1
dubbo-common/src/main/java/org/apache/dubbo/rpc/executor/DefaultExecutorSupport.java
View File

@ -30,6 +30,7 @@ public class DefaultExecutorSupport implements ExecutorSupport {
this.executorRepository = ExecutorRepository.getInstance(url.getOrDefaultApplicationModel());
}
@Override
public Executor getExecutor(Object data) {
return executorRepository.getExecutor(url);
}

6
dubbo-config/dubbo-config-spring/src/main/java/org/apache/dubbo/config/spring/util/LazyTargetInvocationHandler.java
View File

@ -47,11 +47,7 @@ public class LazyTargetInvocationHandler implements InvocationHandler {
}
if (target == null) {
synchronized (this) {
if (target == null) {
target = lazyTargetSource.getTarget();
}
}
target = lazyTargetSource.getTarget();
}
if (method.getDeclaringClass().isInstance(target)) {
try {

2
dubbo-demo/dubbo-demo-native/dubbo-demo-native-consumer/pom.xml
View File

@ -31,7 +31,7 @@
<properties>
<skip_maven_deploy>true</skip_maven_deploy>
<curator5_version>5.1.0</curator5_version>
<zookeeper_version>3.8.1</zookeeper_version>
<zookeeper_version>3.8.3</zookeeper_version>
</properties>
<dependencies>

2
dubbo-demo/dubbo-demo-native/dubbo-demo-native-provider/pom.xml
View File

@ -31,7 +31,7 @@
<properties>
<skip_maven_deploy>true</skip_maven_deploy>
<curator5_version>5.1.0</curator5_version>
<zookeeper_version>3.8.1</zookeeper_version>
<zookeeper_version>3.8.3</zookeeper_version>
</properties>
<dependencies>

2
dubbo-demo/dubbo-demo-spring-boot/pom.xml
View File

@ -38,7 +38,7 @@
<skip_maven_deploy>true</skip_maven_deploy>
<spring-boot.version>2.7.16</spring-boot.version>
<spring-boot-maven-plugin.version>2.7.16</spring-boot-maven-plugin.version>
<micrometer-core.version>1.11.4</micrometer-core.version>
<micrometer-core.version>1.11.5</micrometer-core.version>
</properties>
<dependencyManagement>
<dependencies>

12
dubbo-dependencies-bom/pom.xml
View File

@ -94,7 +94,7 @@
<spring_version>5.3.25</spring_version>
<spring_security_version>5.8.7</spring_security_version>
<javassist_version>3.29.2-GA</javassist_version>
<bytebuddy.version>1.14.8</bytebuddy.version>
<bytebuddy.version>1.14.9</bytebuddy.version>
<netty_version>3.2.10.Final</netty_version>
<netty4_version>4.1.99.Final</netty4_version>
<mina_version>2.2.1</mina_version>
@ -117,7 +117,7 @@
<protobuf-java_version>3.24.4</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.52.v20230823</jetty_version>
<jetty_version>9.4.53.v20231009</jetty_version>
<validation_new_version>3.0.2</validation_new_version>
<validation_version>1.1.0.Final</validation_version>
<hibernate_validator_version>5.4.3.Final</hibernate_validator_version>
@ -133,14 +133,14 @@
<commons_lang3_version>3.12.0</commons_lang3_version>
<protostuff_version>1.8.0</protostuff_version>
<envoy_api_version>0.1.35</envoy_api_version>
<micrometer.version>1.11.4</micrometer.version>
<micrometer.version>1.11.5</micrometer.version>
<opentelemetry.version>1.26.0</opentelemetry.version>
<zipkin-reporter.version>2.16.4</zipkin-reporter.version>
<micrometer-tracing.version>1.1.5</micrometer-tracing.version>
<micrometer-tracing.version>1.1.6</micrometer-tracing.version>
<t_digest.version>3.3</t_digest.version>
<prometheus_client.version>0.16.0</prometheus_client.version>
<reactive.version>1.0.4</reactive.version>
<reactor.version>3.5.10</reactor.version>
<reactor.version>3.5.11</reactor.version>
<rxjava.version>2.2.21</rxjava.version>
<okhttp_version>3.14.9</okhttp_version>
@ -187,7 +187,7 @@
<metrics_version>2.0.6</metrics_version>
<sofa_registry_version>5.4.3</sofa_registry_version>
<gson_version>2.10.1</gson_version>
<jackson_version>2.15.2</jackson_version>
<jackson_version>2.15.3</jackson_version>
<mortbay_jetty_version>6.1.26</mortbay_jetty_version>
<portlet_version>2.0</portlet_version>
<maven_flatten_version>1.5.0</maven_flatten_version>

8
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/model/MetricsSupport.java
View File

@ -20,6 +20,7 @@ package org.apache.dubbo.metrics.model;
import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.lang.Nullable;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metrics.collector.MethodMetricsCollector;
import org.apache.dubbo.metrics.collector.ServiceMetricsCollector;
import org.apache.dubbo.metrics.event.MetricsEvent;
@ -32,10 +33,10 @@ import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.HashMap;
import java.util.Set;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
@ -162,6 +163,9 @@ public class MetricsSupport {
} else {
String serviceUniqueName = invocation.getTargetServiceUniqueName();
String interfaceAndVersion;
if (StringUtils.isBlank(serviceUniqueName)) {
return "";
}
String[] arr = serviceUniqueName.split(PATH_SEPARATOR);
if (arr.length == 2) {
interfaceAndVersion = arr[1];

23
dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/protocol/QosProtocolWrapper.java
View File

@ -75,13 +75,13 @@ public class QosProtocolWrapper implements Protocol, ScopeModelAware {
@Override
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
startQosServer(invoker.getUrl());
startQosServer(invoker.getUrl(), true);
return protocol.export(invoker);
}
@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
startQosServer(url);
startQosServer(url, false);
return protocol.refer(type, url);
}
@ -96,7 +96,7 @@ public class QosProtocolWrapper implements Protocol, ScopeModelAware {
return protocol.getServers();
}
private void startQosServer(URL url) throws RpcException {
private void startQosServer(URL url, boolean isServer) throws RpcException {
boolean qosCheck = url.getParameter(QOS_CHECK, false);
try {
@ -134,13 +134,18 @@ public class QosProtocolWrapper implements Protocol, ScopeModelAware {
} catch (Throwable throwable) {
logger.warn(QOS_FAILED_START_SERVER, "", "", "Fail to start qos server: ", throwable);
try {
stopServer();
} catch (Throwable stop) {
logger.warn(QOS_FAILED_START_SERVER, "", "", "Fail to stop qos server: ", stop);
}
if (qosCheck) {
throw new RpcException(throwable);
try {
// Stop QoS Server to support re-start if Qos-Check is enabled
stopServer();
} catch (Throwable stop) {
logger.warn(QOS_FAILED_START_SERVER, "", "", "Fail to stop qos server: ", stop);
}
if (isServer) {
// Only throws exception when export services
throw new RpcException(throwable);
}
}
}
}

2
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java
View File

@ -183,4 +183,6 @@ public interface Constants {
String CONTENT_LENGTH_KEY = "content-length";
String USE_SECURE_RANDOM_ID = "dubbo.application.use-secure-random-request-id";
String CONNECTION_HANDLER_NAME = "connectionHandler";
}

3
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/AbstractClient.java
View File

@ -40,14 +40,13 @@ import static org.apache.dubbo.common.constants.CommonConstants.THREAD_NAME_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.LAZY_CONNECT_KEY;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CLOSE;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CONNECT_PROVIDER;
import static org.apache.dubbo.config.Constants.CLIENT_THREAD_POOL_NAME;
/**
* AbstractClient
*/
public abstract class AbstractClient extends AbstractEndpoint implements Client {
protected static final String CLIENT_THREAD_POOL_NAME = "DubboClientHandler";
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractClient.class);
private final Lock connectLock = new ReentrantLock();

25
dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionClient.java
View File

@ -20,15 +20,16 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
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.remoting.Channel;
import org.apache.dubbo.remoting.ChannelHandler;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.api.WireProtocol;
import org.apache.dubbo.remoting.api.connection.AbstractConnectionClient;
import org.apache.dubbo.remoting.transport.netty4.ssl.SslClientTlsHandler;
import org.apache.dubbo.remoting.transport.netty4.ssl.SslContexts;
import org.apache.dubbo.remoting.utils.UrlUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
@ -40,6 +41,7 @@ import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoop;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.GlobalEventExecutor;
@ -49,8 +51,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_CLIENT_THREADPOOL;
import static org.apache.dubbo.common.constants.CommonConstants.THREADPOOL_KEY;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_CLIENT_CONNECT_TIMEOUT;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CONNECT_PROVIDER;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RECONNECT;
@ -79,10 +80,7 @@ public class NettyConnectionClient extends AbstractConnectionClient {
@Override
protected void initConnectionClient() {
URL url = ExecutorUtil.setThreadName(getUrl(), "DubboClientHandler");
url = url.addParameterIfAbsent(THREADPOOL_KEY, DEFAULT_CLIENT_THREADPOOL);
setUrl(url);
this.protocol = url.getOrDefaultFrameworkModel().getExtensionLoader(WireProtocol.class).getExtension(url.getProtocol());
this.protocol = getUrl().getOrDefaultFrameworkModel().getExtensionLoader(WireProtocol.class).getExtension(getUrl().getProtocol());
this.remote = getConnectAddress();
this.connectingPromise = new AtomicReference<>();
this.connectionListener = new ConnectionListener();
@ -122,12 +120,15 @@ public class NettyConnectionClient extends AbstractConnectionClient {
}
// pipeline.addLast("logging", new LoggingHandler(LogLevel.INFO)); //for debug
// TODO support IDLE
// int heartbeatInterval = UrlUtils.getHeartbeat(getUrl());
pipeline.addLast("connectionHandler", connectionHandler);
int heartbeat = UrlUtils.getHeartbeat(getUrl());
pipeline.addLast("client-idle-handler", new IdleStateHandler(heartbeat, 0, 0, MILLISECONDS));
pipeline.addLast(Constants.CONNECTION_HANDLER_NAME, connectionHandler);
NettyConfigOperator operator = new NettyConfigOperator(nettyChannel, getChannelHandler());
protocol.configClientPipeline(getUrl(), operator, nettySslContextOperator);
ch.closeFuture().addListener(channelFuture -> doClose());
// TODO support Socks5
}
});
@ -271,7 +272,7 @@ public class NettyConnectionClient extends AbstractConnectionClient {
try {
doConnect();
} catch (RemotingException e) {
LOGGER.error(TRANSPORT_FAILED_RECONNECT, "", "", "Failed to connect to server: " + getConnectAddress());
LOGGER.error(TRANSPORT_FAILED_RECONNECT, "", "", "Failed to connect to server: " + getConnectAddress());
}
}
@ -349,7 +350,7 @@ public class NettyConnectionClient extends AbstractConnectionClient {
try {
connectionClient.doConnect();
} catch (RemotingException e) {
LOGGER.error(TRANSPORT_FAILED_RECONNECT, "", "", "Failed to connect to server: " + getConnectAddress());
LOGGER.error(TRANSPORT_FAILED_RECONNECT, "", "", "Failed to connect to server: " + getConnectAddress());
}
}, 1L, TimeUnit.SECONDS);
}

2
dubbo-remoting/dubbo-remoting-zookeeper-curator5/pom.xml
View File

@ -30,7 +30,7 @@
<properties>
<skip_maven_deploy>false</skip_maven_deploy>
<curator5_version>5.1.0</curator5_version>
<zookeeper_version>3.7.0</zookeeper_version>
<zookeeper_version>3.7.2</zookeeper_version>
</properties>
<dependencies>

4
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java
View File

@ -180,7 +180,7 @@ public class AsyncRpcResult implements Result {
*/
@Override
public Result get() throws InterruptedException, ExecutionException {
if (executor != null && executor instanceof ThreadlessExecutor) {
if (executor instanceof ThreadlessExecutor) {
ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor;
try {
while (!responseFuture.isDone() && !threadlessExecutor.isShutdown()) {
@ -196,7 +196,7 @@ public class AsyncRpcResult implements Result {
@Override
public Result get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
long deadline = System.nanoTime() + unit.toNanos(timeout);
if (executor != null && executor instanceof ThreadlessExecutor) {
if (executor instanceof ThreadlessExecutor) {
ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor;
try {
while (!responseFuture.isDone() && !threadlessExecutor.isShutdown()) {

34
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/EchoFilterTest.java
View File

@ -38,20 +38,9 @@ class EchoFilterTest {
@SuppressWarnings("unchecked")
@Test
void testEcho() {
Invocation invocation = mock(RpcInvocation.class);
Invocation invocation = createMockRpcInvocation();
Invoker<DemoService> invoker = createMockInvoker(invocation);
given(invocation.getMethodName()).willReturn("$echo");
given(invocation.getParameterTypes()).willReturn(new Class<?>[]{Enum.class});
given(invocation.getArguments()).willReturn(new Object[]{"hello"});
given(invocation.getObjectAttachments()).willReturn(null);
Invoker<DemoService> invoker = mock(Invoker.class);
given(invoker.isAvailable()).willReturn(true);
given(invoker.getInterface()).willReturn(DemoService.class);
AppResponse result = new AppResponse();
result.setValue("High");
given(invoker.invoke(invocation)).willReturn(result);
URL url = URL.valueOf("test://test:11/test?group=dubbo&version=1.1");
given(invoker.getUrl()).willReturn(url);
Result filterResult = echoFilter.invoke(invoker, invocation);
assertEquals("hello", filterResult.getValue());
@ -60,22 +49,31 @@ class EchoFilterTest {
@SuppressWarnings("unchecked")
@Test
void testNonEcho() {
Invocation invocation = mock(Invocation.class);
Invocation invocation = createMockRpcInvocation();
Invoker<DemoService> invoker = createMockInvoker(invocation);
given(invocation.getMethodName()).willReturn("echo");
Result filterResult = echoFilter.invoke(invoker, invocation);
assertEquals("High", filterResult.getValue());
}
Invocation createMockRpcInvocation() {
Invocation invocation = mock(RpcInvocation.class);
given(invocation.getParameterTypes()).willReturn(new Class<?>[]{Enum.class});
given(invocation.getArguments()).willReturn(new Object[]{"hello"});
given(invocation.getObjectAttachments()).willReturn(null);
return invocation;
}
Invoker<DemoService> createMockInvoker(Invocation invocation){
Invoker<DemoService> invoker = mock(Invoker.class);
given(invoker.isAvailable()).willReturn(true);
given(invoker.getInterface()).willReturn(DemoService.class);
AppResponse result = new AppResponse();
result.setValue("High");
given(invoker.invoke(invocation)).willReturn(result);
URL url = URL.valueOf("test://test:11/test?group=dubbo&version=1.1");
given(invoker.getUrl()).willReturn(url);
Result filterResult = echoFilter.invoke(invoker, invocation);
assertEquals("High", filterResult.getValue());
return invoker;
}
}

33
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/support/RpcUtilsTest.java
View File

@ -50,6 +50,14 @@ class RpcUtilsTest {
* regular scenario: async invocation in URL
* verify: 1. whether invocationId is set correctly, 2. idempotent or not
*/
Invoker createMockInvoker(URL url){
Invoker invoker = createMockInvoker();
given(invoker.getUrl()).willReturn(url);
return invoker;
}
Invoker createMockInvoker(){
return mock(Invoker.class);
}
@Test
void testAttachInvocationIdIfAsync_normal() {
URL url = URL.valueOf("dubbo://localhost/?test.async=true");
@ -117,8 +125,7 @@ class RpcUtilsTest {
void testGetReturnType() {
Class<?> demoServiceClass = DemoService.class;
String serviceName = demoServiceClass.getName();
Invoker invoker = mock(Invoker.class);
given(invoker.getUrl()).willReturn(URL.valueOf(
Invoker invoker = createMockInvoker(URL.valueOf(
"test://127.0.0.1:1/org.apache.dubbo.rpc.support.DemoService?interface=org.apache.dubbo.rpc.support.DemoService"));
// void sayHello(String name);
@ -157,8 +164,7 @@ class RpcUtilsTest {
void testGetReturnTypesUseCache() throws Exception {
Class<?> demoServiceClass = DemoService.class;
String serviceName = demoServiceClass.getName();
Invoker invoker = mock(Invoker.class);
given(invoker.getUrl()).willReturn(URL.valueOf(
Invoker invoker = createMockInvoker(URL.valueOf(
"test://127.0.0.1:1/org.apache.dubbo.rpc.support.DemoService?interface=org.apache.dubbo.rpc.support.DemoService"));
RpcInvocation inv = new RpcInvocation("testReturnType", serviceName, "", new Class<?>[] {String.class}, null, null, invoker, null);
@ -221,8 +227,7 @@ class RpcUtilsTest {
void testGetReturnTypesWithoutCache() throws Exception {
Class<?> demoServiceClass = DemoService.class;
String serviceName = demoServiceClass.getName();
Invoker invoker = mock(Invoker.class);
given(invoker.getUrl()).willReturn(URL.valueOf(
Invoker invoker = createMockInvoker(URL.valueOf(
"test://127.0.0.1:1/org.apache.dubbo.rpc.support.DemoService?interface=org.apache.dubbo.rpc.support.DemoService"));
RpcInvocation inv = new RpcInvocation("testReturnType", serviceName, "", new Class<?>[] {String.class}, null, null, invoker, null);
@ -286,8 +291,7 @@ class RpcUtilsTest {
void testGetReturnTypesWhenGeneric() throws Exception {
Class<?> demoServiceClass = DemoService.class;
String serviceName = demoServiceClass.getName();
Invoker invoker = mock(Invoker.class);
given(invoker.getUrl()).willReturn(URL.valueOf(
Invoker invoker = createMockInvoker(URL.valueOf(
"test://127.0.0.1:1/org.apache.dubbo.rpc.support.DemoService?interface=org.apache.dubbo.rpc.support.DemoService"));
RpcInvocation inv = new RpcInvocation("testReturnType", serviceName, "", new Class<?>[] {String.class}, null, null, invoker, null);
@ -329,7 +333,7 @@ class RpcUtilsTest {
void testGetParameterTypes() {
Class<?> demoServiceClass = DemoService.class;
String serviceName = demoServiceClass.getName();
Invoker invoker = mock(Invoker.class);
Invoker invoker =createMockInvoker();
// void sayHello(String name);
RpcInvocation inv1 = new RpcInvocation("sayHello", serviceName, "",
@ -383,7 +387,7 @@ class RpcUtilsTest {
public void testGetMethodName(String methodName) {
Class<?> demoServiceClass = DemoService.class;
String serviceName = demoServiceClass.getName();
Invoker invoker = mock(Invoker.class);
Invoker invoker =createMockInvoker();
RpcInvocation inv1 = new RpcInvocation(methodName, serviceName, "",
new Class<?>[] {String.class}, null, null, invoker, null);
@ -401,7 +405,7 @@ class RpcUtilsTest {
public void testGet_$invoke_MethodName(String method) {
Class<?> demoServiceClass = DemoService.class;
String serviceName = demoServiceClass.getName();
Invoker invoker = mock(Invoker.class);
Invoker invoker =createMockInvoker();
RpcInvocation inv = new RpcInvocation("$invoke", serviceName, "",
new Class<?>[] {String.class, String[].class},
@ -418,7 +422,7 @@ class RpcUtilsTest {
Object[] args = new Object[] {"hello", "dubbo", 520};
Class<?> demoServiceClass = DemoService.class;
String serviceName = demoServiceClass.getName();
Invoker invoker = mock(Invoker.class);
Invoker invoker =createMockInvoker();
RpcInvocation inv = new RpcInvocation("$invoke", serviceName, "",
new Class<?>[] {String.class, String[].class, Object[].class},
@ -438,7 +442,7 @@ class RpcUtilsTest {
Object[] args = new Object[] {"hello", "dubbo", 520};
Class<?> demoServiceClass = DemoService.class;
String serviceName = demoServiceClass.getName();
Invoker invoker = mock(Invoker.class);
Invoker invoker =createMockInvoker();
URL url = URL.valueOf(
"test://127.0.0.1:1/org.apache.dubbo.rpc.support.DemoService?interface=org.apache.dubbo.rpc.support.DemoService");
@ -470,8 +474,7 @@ class RpcUtilsTest {
void testIsReturnTypeFuture() {
Class<?> demoServiceClass = DemoService.class;
String serviceName = demoServiceClass.getName();
Invoker invoker = mock(Invoker.class);
given(invoker.getUrl()).willReturn(URL.valueOf(
Invoker invoker = createMockInvoker(URL.valueOf(
"test://127.0.0.1:1/org.apache.dubbo.rpc.support.DemoService?interface=org.apache.dubbo.rpc.support.DemoService"));
RpcInvocation inv = new RpcInvocation("testReturnType", serviceName, "", new Class<?>[] {String.class}, null, null, invoker, null);

2
dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/filter/ServiceInvokeRestFilter.java
View File

@ -46,6 +46,7 @@ import org.apache.dubbo.rpc.protocol.rest.request.RequestFacade;
import org.apache.dubbo.rpc.protocol.rest.util.MediaTypeUtil;
import java.util.List;
import java.util.Objects;
@Activate(value = "invoke", order = Integer.MAX_VALUE)
public class ServiceInvokeRestFilter implements RestRequestFilter {
@ -179,6 +180,7 @@ public class ServiceInvokeRestFilter implements RestRequestFilter {
*/
public static MediaType getAcceptMediaType(RequestFacade request, Class<?> returnType) {
String accept = request.getHeader(RestHeaderEnum.ACCEPT.getHeader());
accept = Objects.isNull(accept) ? MediaType.ALL_VALUE.value : accept;
MediaType mediaType = MediaTypeUtil.convertMediaType(returnType, accept);
return mediaType;
}

14
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/DeadlineFuture.java
View File

@ -47,6 +47,9 @@ public class DeadlineFuture extends CompletableFuture<AppResponse> {
private final List<Runnable> timeoutListeners = new ArrayList<>();
private final Timeout timeoutTask;
private ExecutorService executor;
private static final GlobalResourceInitializer<Timer> TIME_OUT_TIMER = new GlobalResourceInitializer<>(
() -> new HashedWheelTimer(new NamedThreadFactory("dubbo-future-timeout", true), 30,
TimeUnit.MILLISECONDS), DeadlineFuture::destroy);
private DeadlineFuture(String serviceName, String methodName, String address, int timeout) {
this.serviceName = serviceName;
@ -76,20 +79,15 @@ public class DeadlineFuture extends CompletableFuture<AppResponse> {
}
public void received(TriRpcStatus status, AppResponse appResponse) {
if (status.code != TriRpcStatus.Code.DEADLINE_EXCEEDED) {
// decrease Time
if (!timeoutTask.isCancelled()) {
timeoutTask.cancel();
}
if (status.code != TriRpcStatus.Code.DEADLINE_EXCEEDED && !timeoutTask.isCancelled()) {
timeoutTask.cancel();
}
if (getExecutor() != null) {
getExecutor().execute(() -> doReceived(status, appResponse));
} else {
doReceived(status, appResponse);
}
} private static final GlobalResourceInitializer<Timer> TIME_OUT_TIMER = new GlobalResourceInitializer<>(
() -> new HashedWheelTimer(new NamedThreadFactory("dubbo-future-timeout", true), 30,
TimeUnit.MILLISECONDS), DeadlineFuture::destroy);
}
public void addTimeoutListener(Runnable runnable) {
timeoutListeners.add(runnable);

19
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
View File

@ -22,16 +22,16 @@ import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.serialize.MultipleSerialization;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.config.Constants;
import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.remoting.transport.CodecSupport;
import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.apache.dubbo.rpc.model.Pack;
import org.apache.dubbo.rpc.model.PackableMethod;
import com.google.protobuf.Message;
import org.apache.dubbo.rpc.model.UnPack;
import org.apache.dubbo.rpc.model.WrapperUnPack;
import com.google.protobuf.Message;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
@ -42,7 +42,6 @@ import java.util.stream.Stream;
import static org.apache.dubbo.common.constants.CommonConstants.$ECHO;
import static org.apache.dubbo.common.constants.CommonConstants.PROTOBUF_MESSAGE_CLASS_NAME;
import static org.apache.dubbo.rpc.protocol.tri.TripleProtocol.METHOD_ATTR_PACK;
public class ReflectionPackableMethod implements PackableMethod {
@ -118,16 +117,10 @@ public class ReflectionPackableMethod implements PackableMethod {
}
public static ReflectionPackableMethod init(MethodDescriptor methodDescriptor, URL url) {
Object stored = methodDescriptor.getAttribute(METHOD_ATTR_PACK);
if (stored != null) {
return (ReflectionPackableMethod) stored;
}
final String serializeName = UrlUtils.serializationOrDefault(url);
final Collection<String> allSerialize = UrlUtils.allSerializations(url);
ReflectionPackableMethod reflectionPackableMethod = new ReflectionPackableMethod(
String serializeName = UrlUtils.serializationOrDefault(url);
Collection<String> allSerialize = UrlUtils.allSerializations(url);
return new ReflectionPackableMethod(
methodDescriptor, url, serializeName, allSerialize);
methodDescriptor.addAttribute(METHOD_ATTR_PACK, reflectionPackableMethod);
return reflectionPackableMethod;
}
static boolean isStreamType(Class<?> type) {

29
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
View File

@ -17,15 +17,6 @@
package org.apache.dubbo.rpc.protocol.tri;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http2.Http2FrameCodec;
import io.netty.handler.codec.http2.Http2FrameLogger;
import io.netty.handler.codec.http2.Http2MultiplexHandler;
import io.netty.handler.codec.http2.Http2Settings;
import io.netty.handler.codec.http2.Http2StreamChannel;
import io.netty.handler.flush.FlushConsolidationHandler;
import io.netty.handler.logging.LogLevel;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.Configuration;
import org.apache.dubbo.common.config.ConfigurationUtils;
@ -37,17 +28,29 @@ import org.apache.dubbo.remoting.api.AbstractWireProtocol;
import org.apache.dubbo.remoting.api.pu.ChannelHandlerPretender;
import org.apache.dubbo.remoting.api.pu.ChannelOperator;
import org.apache.dubbo.remoting.api.ssl.ContextOperator;
import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.HeaderFilter;
import org.apache.dubbo.rpc.executor.ExecutorSupport;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.ScopeModelAware;
import org.apache.dubbo.rpc.protocol.tri.transport.TripleClientHandler;
import org.apache.dubbo.rpc.protocol.tri.transport.TripleCommandOutBoundHandler;
import org.apache.dubbo.rpc.protocol.tri.transport.TripleGoAwayHandler;
import org.apache.dubbo.rpc.protocol.tri.transport.TripleHttp2FrameServerHandler;
import org.apache.dubbo.rpc.protocol.tri.transport.TripleServerConnectionHandler;
import org.apache.dubbo.rpc.protocol.tri.transport.TripleTailHandler;
import org.apache.dubbo.rpc.protocol.tri.transport.TripleWriteQueue;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http2.Http2FrameCodec;
import io.netty.handler.codec.http2.Http2FrameLogger;
import io.netty.handler.codec.http2.Http2MultiplexHandler;
import io.netty.handler.codec.http2.Http2Settings;
import io.netty.handler.codec.http2.Http2StreamChannel;
import io.netty.handler.flush.FlushConsolidationHandler;
import io.netty.handler.logging.LogLevel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@ -160,11 +163,11 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo
.frameLogger(CLIENT_LOGGER)
.build();
codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter());
final Http2MultiplexHandler handler = new Http2MultiplexHandler(
new TripleClientHandler(frameworkModel));
List<ChannelHandler> handlers = new ArrayList<>();
handlers.add(new ChannelHandlerPretender(codec));
handlers.add(new ChannelHandlerPretender(handler));
handlers.add(new ChannelHandlerPretender(new Http2MultiplexHandler(new ChannelDuplexHandler())));
handlers.add(new ChannelHandlerPretender(new TripleGoAwayHandler()));
handlers.add(new ChannelHandlerPretender(new TriplePingPongHandler(UrlUtils.getCloseTimeout(url))));
handlers.add(new ChannelHandlerPretender(new TripleTailHandler()));
operator.configChannelHandler(handlers);
}

25
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
View File

@ -62,9 +62,11 @@ import org.apache.dubbo.rpc.support.RpcUtils;
import io.netty.util.AsciiString;
import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.ReentrantLock;
@ -94,18 +96,22 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
private final TripleWriteQueue writeQueue = new TripleWriteQueue(256);
private static final boolean setFutureWhenSync = Boolean.parseBoolean(System.getProperty(CommonConstants.SET_FUTURE_IN_SYNC_MODE, "true"));
private final PackableMethodFactory packableMethodFactory;
private final Map<MethodDescriptor, PackableMethod> packableMethodCache = new ConcurrentHashMap<>();
public TripleInvoker(Class<T> serviceType,
URL url,
String acceptEncodings,
AbstractConnectionClient connectionClient,
Set<Invoker<?>> invokers,
ExecutorService streamExecutor) {
URL url,
String acceptEncodings,
AbstractConnectionClient connectionClient,
Set<Invoker<?>> invokers,
ExecutorService streamExecutor) {
super(serviceType, url, new String[]{INTERFACE_KEY, GROUP_KEY, TOKEN_KEY});
this.invokers = invokers;
this.connectionClient = connectionClient;
this.acceptEncodings = acceptEncodings;
this.streamExecutor = streamExecutor;
this.packableMethodFactory = url.getOrDefaultFrameworkModel().getExtensionLoader(PackableMethodFactory.class)
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel()).getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY));
}
private static AsciiString getSchemeFromUrl(URL url) {
@ -175,7 +181,7 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
}
}
private static boolean isSync(MethodDescriptor methodDescriptor, Invocation invocation){
private static boolean isSync(MethodDescriptor methodDescriptor, Invocation invocation) {
if (!(invocation instanceof RpcInvocation)) {
return false;
}
@ -230,7 +236,7 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
if (timeout <= 0) {
return AsyncRpcResult.newDefaultAsyncResult(new RpcException(RpcException.TIMEOUT_TERMINATE,
"No time left for making the following call: " + invocation.getServiceName() + "."
+ RpcUtils.getMethodName(invocation)+ ", terminate directly."), invocation);
+ RpcUtils.getMethodName(invocation) + ", terminate directly."), invocation);
}
invocation.setAttachment(TIMEOUT_KEY, String.valueOf(timeout));
@ -280,9 +286,8 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
if (methodDescriptor instanceof PackableMethod) {
meta.packableMethod = (PackableMethod) methodDescriptor;
} else {
meta.packableMethod = url.getOrDefaultFrameworkModel().getExtensionLoader(PackableMethodFactory.class)
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel()).getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY))
.create(methodDescriptor, url, TripleConstant.CONTENT_PROTO);
meta.packableMethod = packableMethodCache.computeIfAbsent(methodDescriptor,
(md) -> packableMethodFactory.create(md, url, TripleConstant.CONTENT_PROTO));
}
meta.convertNoLowerHeader = TripleProtocol.CONVERT_NO_LOWER_HEADER;
meta.ignoreDefaultVersion = TripleProtocol.IGNORE_1_0_0_VERSION;

76
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TriplePingPongHandler.java Normal file
View File

@ -0,0 +1,76 @@
/*
* 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.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http2.DefaultHttp2PingFrame;
import io.netty.handler.codec.http2.Http2PingFrame;
import io.netty.handler.timeout.IdleStateEvent;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
public class TriplePingPongHandler extends ChannelDuplexHandler {
private final long pingAckTimeout;
private ScheduledFuture<?> pingAckTimeoutFuture;
public TriplePingPongHandler(long pingAckTimeout) {
this.pingAckTimeout = pingAckTimeout;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (!(msg instanceof Http2PingFrame) || pingAckTimeoutFuture == null) {
super.channelRead(ctx, msg);
return;
}
//cancel task when read anything, include http2 ping ack
pingAckTimeoutFuture.cancel(true);
pingAckTimeoutFuture = null;
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (!(evt instanceof IdleStateEvent)) {
ctx.fireUserEventTriggered(evt);
return;
}
ctx.writeAndFlush(new DefaultHttp2PingFrame(0));
if (pingAckTimeoutFuture == null) {
pingAckTimeoutFuture = ctx.executor().schedule(new CloseChannelTask(ctx), pingAckTimeout, TimeUnit.MILLISECONDS);
}
//not null means last ping ack not received
}
private static class CloseChannelTask implements Runnable {
private final ChannelHandlerContext ctx;
public CloseChannelTask(ChannelHandlerContext ctx) {
this.ctx = ctx;
}
@Override
public void run() {
ctx.close();
}
}
}

16
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java
View File

@ -44,6 +44,10 @@ import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutorService;
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.common.constants.CommonConstants.THREAD_NAME_KEY;
import static org.apache.dubbo.config.Constants.CLIENT_THREAD_POOL_NAME;
import static org.apache.dubbo.config.Constants.SERVER_THREAD_POOL_NAME;
import static org.apache.dubbo.rpc.Constants.H2_IGNORE_1_0_0_KEY;
import static org.apache.dubbo.rpc.Constants.H2_RESOLVE_FALLBACK_TO_DEFAULT_KEY;
@ -51,8 +55,6 @@ import static org.apache.dubbo.rpc.Constants.H2_SUPPORT_NO_LOWER_HEADER_KEY;
public class TripleProtocol extends AbstractProtocol {
public static final String METHOD_ATTR_PACK = "pack";
private static final Logger logger = LoggerFactory.getLogger(TripleProtocol.class);
private final PathResolver pathResolver;
private final TriBuiltinService triBuiltinService;
@ -154,8 +156,7 @@ public class TripleProtocol extends AbstractProtocol {
@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
optimizeSerialization(url);
ExecutorService streamExecutor = getOrCreateStreamExecutor(
url.getOrDefaultApplicationModel(), url);
ExecutorService streamExecutor = getOrCreateStreamExecutor(url.getOrDefaultApplicationModel(), url);
AbstractConnectionClient connectionClient = PortUnificationExchanger.connect(url, new DefaultPuHandler());
TripleInvoker<T> invoker = new TripleInvoker<>(type, url, acceptEncodings,
connectionClient, invokers, streamExecutor);
@ -164,9 +165,10 @@ public class TripleProtocol extends AbstractProtocol {
}
private ExecutorService getOrCreateStreamExecutor(ApplicationModel applicationModel, URL url) {
ExecutorService executor = ExecutorRepository.getInstance(applicationModel).createExecutorIfAbsent(ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME));
Objects.requireNonNull(executor,
String.format("No available executor found in %s", url));
url = url.addParameter(THREAD_NAME_KEY, CLIENT_THREAD_POOL_NAME)
.addParameterIfAbsent(THREADPOOL_KEY, DEFAULT_CLIENT_THREADPOOL);
ExecutorService executor = ExecutorRepository.getInstance(applicationModel).createExecutorIfAbsent(url);
Objects.requireNonNull(executor, String.format("No available executor found in %s", url));
return executor;
}

29
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ReflectionAbstractServerCall.java
View File

@ -17,7 +17,6 @@
package org.apache.dubbo.rpc.protocol.tri.call;
import io.netty.handler.codec.http.HttpHeaderNames;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.constants.CommonConstants;
@ -29,6 +28,7 @@ import org.apache.dubbo.rpc.TriRpcStatus;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.apache.dubbo.rpc.model.MethodDescriptor.RpcType;
import org.apache.dubbo.rpc.model.PackableMethod;
import org.apache.dubbo.rpc.model.PackableMethodFactory;
import org.apache.dubbo.rpc.model.ProviderModel;
import org.apache.dubbo.rpc.model.ServiceDescriptor;
@ -37,8 +37,12 @@ import org.apache.dubbo.rpc.protocol.tri.TripleCustomerProtocolWapper;
import org.apache.dubbo.rpc.protocol.tri.stream.ServerStream;
import org.apache.dubbo.rpc.service.ServiceDescriptorInternalCache;
import io.netty.handler.codec.http.HttpHeaderNames;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
@ -46,6 +50,7 @@ import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_PACKABLE_M
public class ReflectionAbstractServerCall extends AbstractServerCall {
private static final String PACKABLE_METHOD_CACHE = "PACKABLE_METHOD_CACHE";
private final List<HeaderFilter> headerFilters;
private List<MethodDescriptor> methodDescriptors;
@ -120,10 +125,7 @@ public class ReflectionAbstractServerCall extends AbstractServerCall {
}
}
if (methodDescriptor != null) {
final URL url = invoker.getUrl();
packableMethod = frameworkModel.getExtensionLoader(PackableMethodFactory.class)
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel()).getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY))
.create(methodDescriptor, url, (String) requestMetadata.get(HttpHeaderNames.CONTENT_TYPE.toString()));
loadPackableMethod(invoker.getUrl());
}
trySetListener();
if (listener == null) {
@ -191,10 +193,19 @@ public class ReflectionAbstractServerCall extends AbstractServerCall {
+ serviceDescriptor.getInterfaceName()), null);
return;
}
final URL url = invoker.getUrl();
packableMethod = frameworkModel.getExtensionLoader(PackableMethodFactory.class)
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel()).getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY))
.create(methodDescriptor, url, (String) requestMetadata.get(HttpHeaderNames.CONTENT_TYPE.toString()));
loadPackableMethod(invoker.getUrl());
}
@SuppressWarnings("unchecked")
private void loadPackableMethod(URL url) {
Map<MethodDescriptor, PackableMethod> cacheMap = (Map<MethodDescriptor, PackableMethod>) url.getServiceModel()
.getServiceMetadata()
.getAttributeMap()
.computeIfAbsent(PACKABLE_METHOD_CACHE, (k) -> new ConcurrentHashMap<>());
packableMethod = cacheMap.computeIfAbsent(methodDescriptor,
(md) -> frameworkModel.getExtensionLoader(PackableMethodFactory.class)
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel()).getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY))
.create(methodDescriptor, url, (String) requestMetadata.get(HttpHeaderNames.CONTENT_TYPE.toString())));
}
}

18
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleClientHandler.java → dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleGoAwayHandler.java
View File

@ -17,28 +17,30 @@
package org.apache.dubbo.rpc.protocol.tri.transport;
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.ConnectionHandler;
import org.apache.dubbo.rpc.model.FrameworkModel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http2.Http2GoAwayFrame;
import io.netty.util.ReferenceCountUtil;
public class TripleClientHandler extends ChannelDuplexHandler {
public class TripleGoAwayHandler extends ChannelDuplexHandler {
private final FrameworkModel frameworkModel;
private static final Logger logger = LoggerFactory.getLogger(TripleGoAwayHandler.class);
private static final String CONNECTION_HANDLER_NAME = "connectionHandler";
public TripleClientHandler(FrameworkModel frameworkModel) {
this.frameworkModel = frameworkModel;
public TripleGoAwayHandler() {
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof Http2GoAwayFrame) {
final ConnectionHandler connectionHandler = (ConnectionHandler) ctx.pipeline().get(CONNECTION_HANDLER_NAME);
final ConnectionHandler connectionHandler = (ConnectionHandler) ctx.pipeline().get(Constants.CONNECTION_HANDLER_NAME);
if (logger.isInfoEnabled()) {
logger.info("Receive go away frame of " + ctx.channel().localAddress() + " -> " + ctx.channel().remoteAddress() + " and will reconnect later.");
}
connectionHandler.onGoAway(ctx.channel());
}
ReferenceCountUtil.release(msg);

4
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/call/ReflectionServerCallTest.java
View File

@ -24,6 +24,7 @@ import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.apache.dubbo.rpc.model.ProviderModel;
import org.apache.dubbo.rpc.model.ReflectionMethodDescriptor;
import org.apache.dubbo.rpc.model.ServiceDescriptor;
import org.apache.dubbo.rpc.model.ServiceMetadata;
import org.apache.dubbo.rpc.protocol.tri.DescriptorService;
import org.apache.dubbo.rpc.protocol.tri.HelloReply;
import org.apache.dubbo.rpc.protocol.tri.stream.TripleServerStream;
@ -46,6 +47,7 @@ class ReflectionServerCallTest {
Invoker<?> invoker = Mockito.mock(Invoker.class);
TripleServerStream serverStream = Mockito.mock(TripleServerStream.class);
ProviderModel providerModel = Mockito.mock(ProviderModel.class);
ServiceMetadata serviceMetadata = new ServiceMetadata();
Method method = DescriptorService.class.getMethod("sayHello", HelloReply.class);
MethodDescriptor methodDescriptor = new ReflectionMethodDescriptor(method);
URL url = Mockito.mock(URL.class);
@ -53,6 +55,8 @@ class ReflectionServerCallTest {
.thenReturn(url);
when(url.getServiceModel())
.thenReturn(providerModel);
when(providerModel.getServiceMetadata())
.thenReturn(serviceMetadata);
String service = "testService";
String methodName = "method";

2
dubbo-spring-boot/dubbo-spring-boot-starters/dubbo-zookeeper-curator5-spring-boot-starter/pom.xml
View File

@ -33,7 +33,7 @@
<properties>
<curator5_version>5.1.0</curator5_version>
<zookeeper_version>3.8.1</zookeeper_version>
<zookeeper_version>3.8.3</zookeeper_version>
</properties>
<dependencyManagement>

4
dubbo-spring-boot/dubbo-spring-boot-starters/observability/pom.xml
View File

@ -38,8 +38,8 @@
</modules>
<properties>
<micrometer.version>1.11.4</micrometer.version>
<micrometer-tracing.version>1.1.5</micrometer-tracing.version>
<micrometer.version>1.11.5</micrometer.version>
<micrometer-tracing.version>1.1.6</micrometer-tracing.version>
<opentelemetry.version>1.31.0</opentelemetry.version>
<zipkin-reporter.version>2.16.4</zipkin-reporter.version>
<prometheus-client.version>0.16.0</prometheus-client.version>

2
dubbo-spring-boot/pom.xml
View File

@ -46,7 +46,7 @@
<!-- Fix the bug of log4j refer:https://github.com/apache/logging-log4j2/pull/608 -->
<log4j2_version>2.20.0</log4j2_version>
<!-- Spring boot buddy is lower than the delivery dependency package version and can only show the defined dependency version -->
<byte-buddy.version>1.14.8</byte-buddy.version>
<byte-buddy.version>1.14.9</byte-buddy.version>
</properties>
<dependencyManagement>

2
dubbo-test/dubbo-test-check/pom.xml
View File

@ -33,7 +33,7 @@
<maven.compiler.target>8</maven.compiler.target>
<skip_maven_deploy>true</skip_maven_deploy>
<curator.test.version>4.2.0</curator.test.version>
<zookeeper.version>3.4.14</zookeeper.version>
<zookeeper.version>3.7.2</zookeeper.version>
<curator5.version>4.2.0</curator5.version>
<commons.compress.version>1.24.0</commons.compress.version>
<junit.platform.launcher.version>1.9.3</junit.platform.launcher.version>

2
pom.xml
View File

@ -119,7 +119,7 @@
<maven_compiler_version>3.11.0</maven_compiler_version>
<maven_source_version>3.2.1</maven_source_version>
<maven_javadoc_version>3.5.0</maven_javadoc_version>
<maven_jetty_version>9.4.52.v20230823</maven_jetty_version>
<maven_jetty_version>9.4.53.v20231009</maven_jetty_version>
<maven_checkstyle_version>3.2.1</maven_checkstyle_version>
<maven_jacoco_version>0.8.10</maven_jacoco_version>
<maven_flatten_version>1.5.0</maven_flatten_version>