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

Merge branch 'apache-3.0' into 3.0.11-release

This commit is contained in:
Albumen Kevin 2022-08-17 21:44:58 +08:00
parent 5081fe15b3 4da708f3df
commit e4abc7a4ee
5 changed files with 198 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvoker.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.cluster.support.wrapper;
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.common.utils.CollectionUtils;
@ -24,6 +25,7 @@ import org.apache.dubbo.common.utils.ConfigUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.AsyncRpcResult;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.InvokeMode;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
@ -31,7 +33,9 @@ import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.protocol.dubbo.FutureAdapter;
import org.apache.dubbo.rpc.support.MockInvoker;
import org.apache.dubbo.rpc.support.RpcUtils;
import java.util.List;
@ -42,6 +46,7 @@ import static org.apache.dubbo.rpc.cluster.Constants.INVOCATION_NEED_MOCK;
public class MockClusterInvoker<T> implements ClusterInvoker<T> {
private static final Logger logger = LoggerFactory.getLogger(MockClusterInvoker.class);
private static final boolean setFutureWhenSync = Boolean.parseBoolean(System.getProperty(CommonConstants.SET_FUTURE_IN_SYNC_MODE, "true"));
private final Directory<T> directory;
@ -135,6 +140,9 @@ public class MockClusterInvoker<T> implements ClusterInvoker<T> {
Result result;
Invoker<T> mockInvoker;
RpcInvocation rpcInvocation = (RpcInvocation)invocation;
rpcInvocation.setInvokeMode(RpcUtils.getInvokeMode(getUrl(),invocation));
List<Invoker<T>> mockInvokers = selectMockInvoker(invocation);
if (CollectionUtils.isEmpty(mockInvokers)) {
mockInvoker = (Invoker<T>) new MockInvoker(getUrl(), directory.getInterface());
@ -152,6 +160,10 @@ public class MockClusterInvoker<T> implements ClusterInvoker<T> {
} catch (Throwable me) {
throw new RpcException(getMockExceptionMessage(e, me), me.getCause());
}
if (setFutureWhenSync || rpcInvocation.getInvokeMode() != InvokeMode.SYNC) {
// set server context
RpcContext.getServiceContext().setFuture(new FutureAdapter<>(((AsyncRpcResult)result).getResponseFuture()));
}
return result;
}

44
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvokerTest.java
View File

@ -37,6 +37,8 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.REFER_KEY;
@ -157,6 +159,9 @@ public class MockClusterInvokerTest {
invocation.setMethodName("sayHello");
ret = cluster.invoke(invocation);
Assertions.assertNull(ret.getValue());
}
@Test
@ -589,6 +594,29 @@ public class MockClusterInvokerTest {
Result ret = cluster.invoke(invocation);
Assertions.assertEquals(0, ((List<User>) ret.getValue()).size());
}
@Test
public void testMockInvokerFromOverride_Invoke_check_ListPojoAsync() throws ExecutionException, InterruptedException {
URL url = URL.valueOf("remote://1.2.3.4/" + IHelloService.class.getName())
.addParameter(REFER_KEY,
URL.encode(PATH_KEY + "=" + IHelloService.class.getName()
+ "&" + "getUsersAsync.mock=force"))
.addParameter("invoke_return_error", "true");
Invoker<IHelloService> cluster = getClusterInvoker(url);
//Configured with mock
RpcInvocation invocation = new RpcInvocation();
invocation.setMethodName("getUsersAsync");
invocation.setReturnType(CompletableFuture.class);
Result ret = cluster.invoke(invocation);
CompletableFuture<List<User>> cf = null;
try {
cf = (CompletableFuture<List<User>>) ret.recreate();
} catch (Throwable e) {
e.printStackTrace();
}
Assertions.assertEquals(2, cf.get().size());
Assertions.assertEquals("Tommock", cf.get().get(0).getName());
}
@SuppressWarnings("unchecked")
@Test
@ -753,6 +781,8 @@ public class MockClusterInvokerTest {
List<User> getUsers();
CompletableFuture<List<User>> getUsersAsync();
void sayHello();
}
@ -793,6 +823,13 @@ public class MockClusterInvokerTest {
return Arrays.asList(new User[]{new User(1, "Tom"), new User(2, "Jerry")});
}
@Override
public CompletableFuture<List<User>> getUsersAsync() {
CompletableFuture<List<User>> cf=new CompletableFuture<>();
cf.complete(Arrays.asList(new User[]{new User(1, "Tom"), new User(2, "Jerry")}));
return cf;
}
public void sayHello() {
System.out.println("hello prety");
}
@ -827,6 +864,13 @@ public class MockClusterInvokerTest {
return Arrays.asList(new User[]{new User(1, "Tommock"), new User(2, "Jerrymock")});
}
@Override
public CompletableFuture<List<User>> getUsersAsync() {
CompletableFuture<List<User>> cf=new CompletableFuture<>();
cf.complete(Arrays.asList(new User[]{new User(1, "Tommock"), new User(2, "Jerrymock")}));
return cf;
}
public int getInt1() {
return 1;
}

89
dubbo-common/src/main/java/org/apache/dubbo/rpc/service/GenericException.java
View File

@ -20,7 +20,7 @@ import org.apache.dubbo.common.utils.JsonUtils;
import org.apache.dubbo.common.utils.StringUtils;
import java.beans.Transient;
import java.io.Serializable;
/**
* GenericException
@ -37,8 +37,10 @@ public class GenericException extends RuntimeException {
private String exceptionMessage;
private final GenericExceptionInfo genericExceptionInfo;
public GenericException() {
this.useCause = false;
this(null, null);
}
public GenericException(String exceptionClass, String exceptionMessage) {
@ -46,13 +48,7 @@ public class GenericException extends RuntimeException {
this.useCause = false;
this.exceptionClass = exceptionClass;
this.exceptionMessage = exceptionMessage;
}
public GenericException(String exceptionClass, String exceptionMessage, String message) {
super(message);
this.useCause = false;
this.exceptionClass = exceptionClass;
this.exceptionMessage = exceptionMessage;
this.genericExceptionInfo = new GenericExceptionInfo(exceptionClass, exceptionMessage, exceptionMessage, getStackTrace());
}
public GenericException(Throwable cause) {
@ -60,43 +56,83 @@ public class GenericException extends RuntimeException {
this.useCause = false;
this.exceptionClass = cause.getClass().getName();
this.exceptionMessage = cause.getMessage();
this.genericExceptionInfo = new GenericExceptionInfo(this.exceptionClass, this.exceptionMessage, super.getMessage(), getStackTrace());
}
protected GenericException(GenericExceptionInfo info) {
super(info.getMsg(), null, true, false);
setStackTrace(info.getStackTrace());
this.useCause = false;
this.exceptionClass = info.getExClass();
this.exceptionMessage = info.getExMsg();
this.genericExceptionInfo = info;
}
@Transient
public String getExceptionClass() {
if(this.useCause) {
return ((GenericException)getCause()).getExceptionClass();
}
return exceptionClass;
}
public void setExceptionClass(String exceptionClass) {
if(this.useCause) {
((GenericException)getCause()).setExceptionClass(exceptionClass);
return;
}
this.exceptionClass = exceptionClass;
}
@Transient
public String getExceptionMessage() {
if(this.useCause) {
return ((GenericException)getCause()).getExceptionMessage();
}
return exceptionMessage;
}
public void setExceptionMessage(String exceptionMessage) {
if(this.useCause) {
((GenericException)getCause()).setExceptionMessage(exceptionMessage);
return;
}
this.exceptionMessage = exceptionMessage;
}
@Override
@Transient
public StackTraceElement[] getStackTrace() {
if(this.useCause) {
return ((GenericException)getCause()).getStackTrace();
}
return super.getStackTrace();
}
@Override
@Transient
public String getMessage() {
if(this.useCause) {
Throwable cause = getCause();
if(cause != null) {
return cause.getMessage();
}
return getCause().getMessage();
}
return JsonUtils.getJson().toJson(GenericExceptionInfo.createNoStackTrace(genericExceptionInfo));
}
public String getGenericException() {
if(this.useCause) {
return ((GenericException)getCause()).getGenericException();
}
GenericExceptionInfo genericExceptionInfo = new GenericExceptionInfo(exceptionClass, exceptionMessage, getMessage());
return JsonUtils.getJson().toJson(genericExceptionInfo);
}
public void setMessage(String json) {
public void setGenericException(String json) {
GenericExceptionInfo info = JsonUtils.getJson().toJavaObject(json, GenericExceptionInfo.class);
if(info == null) {
return;
}
this.useCause = true;
initCause(new GenericException(info.getExClass(), info.getExMsg(), info.getMsg()));
initCause(new GenericException(info));
}
@Override
@ -108,15 +144,24 @@ public class GenericException extends RuntimeException {
/**
* create generic exception info
*/
public static class GenericExceptionInfo {
public static class GenericExceptionInfo implements Serializable {
private String exClass;
private String exMsg;
private String msg;
private StackTraceElement[] stackTrace;
public GenericExceptionInfo(String exceptionClass, String exceptionMessage, String message) {
public GenericExceptionInfo() {
}
public GenericExceptionInfo(String exceptionClass, String exceptionMessage, String message, StackTraceElement[] stackTrace) {
this.exClass = exceptionClass;
this.exMsg = exceptionMessage;
this.msg = message;
this.stackTrace = stackTrace;
}
public static GenericExceptionInfo createNoStackTrace(GenericExceptionInfo info) {
return new GenericExceptionInfo(info.getExClass(), info.getExMsg(), info.getMsg(), null);
}
public String getMsg() {
@ -142,5 +187,13 @@ public class GenericException extends RuntimeException {
public void setMsg(String msg) {
this.msg = msg;
}
public StackTraceElement[] getStackTrace() {
return stackTrace;
}
public void setStackTrace(StackTraceElement[] stackTrace) {
this.stackTrace = stackTrace;
}
}
}

59
dubbo-common/src/test/java/org/apache/dubbo/rpc/service/GenericExceptionTest.java Normal file
View File

@ -0,0 +1,59 @@
/*
* 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.service;
import org.apache.dubbo.common.utils.JsonUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.io.IOException;
public class GenericExceptionTest {
@Test
void jsonSupport() throws IOException {
{
GenericException src = new GenericException();
String s = JsonUtils.getJson().toJson(src);
GenericException dst = JsonUtils.getJson().toJavaObject(s, GenericException.class);
Assertions.assertEquals(src.getExceptionClass(), dst.getExceptionClass());
Assertions.assertEquals(src.getExceptionMessage(), dst.getExceptionMessage());
Assertions.assertEquals(src.getMessage(), dst.getMessage());
Assertions.assertEquals(src.getExceptionMessage(), dst.getExceptionMessage());
}
{
GenericException src = new GenericException(this.getClass().getSimpleName(), "test");
String s = JsonUtils.getJson().toJson(src);
GenericException dst = JsonUtils.getJson().toJavaObject(s, GenericException.class);
Assertions.assertEquals(src.getExceptionClass(), dst.getExceptionClass());
Assertions.assertEquals(src.getExceptionMessage(), dst.getExceptionMessage());
Assertions.assertEquals(src.getMessage(), dst.getMessage());
Assertions.assertEquals(src.getExceptionMessage(), dst.getExceptionMessage());
}
{
Throwable throwable = new Throwable("throwable");
GenericException src = new GenericException(throwable);
String s = JsonUtils.getJson().toJson(src);
GenericException dst = JsonUtils.getJson().toJavaObject(s, GenericException.class);
Assertions.assertEquals(src.getExceptionClass(), dst.getExceptionClass());
Assertions.assertEquals(src.getExceptionMessage(), dst.getExceptionMessage());
Assertions.assertEquals(src.getMessage(), dst.getMessage());
Assertions.assertEquals(src.getExceptionMessage(), dst.getExceptionMessage());
}
}
}

16
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/AbstractServer.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
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.ConcurrentHashSet;
import org.apache.dubbo.common.utils.ExecutorUtil;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.remoting.Channel;
@ -30,6 +31,7 @@ import org.apache.dubbo.remoting.RemotingServer;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_KEY;
@ -44,7 +46,7 @@ public abstract class AbstractServer extends AbstractEndpoint implements Remotin
protected static final String SERVER_THREAD_POOL_NAME = "DubboServerHandler";
private static final Logger logger = LoggerFactory.getLogger(AbstractServer.class);
private ExecutorService executor;
private Set<ExecutorService> executors = new ConcurrentHashSet<>();
private InetSocketAddress localAddress;
private InetSocketAddress bindAddress;
private int accepts;
@ -72,7 +74,7 @@ public abstract class AbstractServer extends AbstractEndpoint implements Remotin
throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName()
+ " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
}
executor = executorRepository.createExecutorIfAbsent(url);
executors.add(executorRepository.createExecutorIfAbsent(url));
}
protected abstract void doOpen() throws Throwable;
@ -96,6 +98,8 @@ public abstract class AbstractServer extends AbstractEndpoint implements Remotin
logger.error(t.getMessage(), t);
}
ExecutorService executor = executorRepository.createExecutorIfAbsent(url);
executors.add(executor);
executorRepository.updateThreadpool(url, executor);
super.setUrl(getUrl().addParameters(url.getParameters()));
}
@ -116,7 +120,9 @@ public abstract class AbstractServer extends AbstractEndpoint implements Remotin
logger.info("Close " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
}
ExecutorUtil.shutdownNow(executor, 100);
for (ExecutorService executor : executors) {
ExecutorUtil.shutdownNow(executor, 100);
}
try {
super.close();
@ -133,7 +139,9 @@ public abstract class AbstractServer extends AbstractEndpoint implements Remotin
@Override
public void close(int timeout) {
ExecutorUtil.gracefulShutdown(executor, timeout);
for (ExecutorService executor : executors) {
ExecutorUtil.gracefulShutdown(executor, timeout);
}
close();
}