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

FutureAdapter depends on AsyncRpcResult (#4471) 

#4471
This commit is contained in:
ken.lj 2019-07-05 17:30:53 +08:00 committed by cvictory
parent f456745423
commit e3e5460e4e
9 changed files with 126 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dubbo-compatible/src/main/java/com/alibaba/dubbo/rpc/RpcContext.java
View File

@ -57,7 +57,7 @@ public class RpcContext extends org.apache.dubbo.rpc.RpcContext {
}
public <T> Future<T> getFuture() {
CompletableFuture completableFuture = FutureContext.getCompletableFuture();
CompletableFuture completableFuture = FutureContext.getContext().getCompatibleCompletableFuture();
if (completableFuture == null) {
return null;
}

45
dubbo-compatible/src/main/java/com/alibaba/dubbo/rpc/protocol/dubbo/FutureAdapter.java
View File

@ -18,10 +18,12 @@
package com.alibaba.dubbo.rpc.protocol.dubbo;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Result;
import com.alibaba.dubbo.remoting.RemotingException;
import com.alibaba.dubbo.remoting.exchange.ResponseCallback;
import com.alibaba.dubbo.remoting.exchange.ResponseFuture;
import com.alibaba.dubbo.rpc.RpcException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@ -36,11 +38,11 @@ import java.util.function.BiConsumer;
*/
@Deprecated
public class FutureAdapter<V> implements Future<V> {
private CompletableFuture<V> future;
public FutureAdapter(CompletableFuture<V> future) {
private CompletableFuture<Object> future;
public FutureAdapter(CompletableFuture<Object> future) {
this.future = future;
}
public ResponseFuture getFuture() {
@ -48,7 +50,7 @@ public class FutureAdapter<V> implements Future<V> {
@Override
public Object get() throws RemotingException {
try {
return FutureAdapter.this.get();
return future.get();
} catch (InterruptedException e) {
throw new RemotingException(e);
} catch (ExecutionException e) {
@ -59,7 +61,7 @@ public class FutureAdapter<V> implements Future<V> {
@Override
public Object get(int timeoutInMillis) throws RemotingException {
try {
return FutureAdapter.this.get(timeoutInMillis, TimeUnit.MILLISECONDS);
return future.get(timeoutInMillis, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
throw new RemotingException(e);
} catch (ExecutionException e) {
@ -76,26 +78,23 @@ public class FutureAdapter<V> implements Future<V> {
@Override
public boolean isDone() {
return FutureAdapter.this.isDone();
return future.isDone();
}
};
}
void setCallback(ResponseCallback callback) {
if (!(future instanceof org.apache.dubbo.rpc.protocol.dubbo.FutureAdapter)) {
return;
}
org.apache.dubbo.rpc.protocol.dubbo.FutureAdapter futureAdapter = (org.apache.dubbo.rpc.protocol.dubbo.FutureAdapter) future;
BiConsumer<AppResponse, ? super Throwable> biConsumer = new BiConsumer<AppResponse, Throwable>() {
BiConsumer<Object, ? super Throwable> biConsumer = new BiConsumer<Object, Throwable>() {
@Override
public void accept(AppResponse appResponse, Throwable t) {
public void accept(Object obj, Throwable t) {
if (t != null) {
if (t instanceof CompletionException) {
t = t.getCause();
}
callback.caught(t);
} else {
AppResponse appResponse = (AppResponse)obj;
if (appResponse.hasException()) {
callback.caught(appResponse.getException());
} else {
@ -104,15 +103,15 @@ public class FutureAdapter<V> implements Future<V> {
}
}
};
futureAdapter.getAppResponseFuture().whenComplete(biConsumer);
future.whenComplete(biConsumer);
}
public boolean cancel(boolean mayInterruptIfRunning) {
return future.cancel(mayInterruptIfRunning);
return false;
}
public boolean isCancelled() {
return future.isCancelled();
return false;
}
public boolean isDone() {
@ -121,11 +120,23 @@ public class FutureAdapter<V> implements Future<V> {
@SuppressWarnings("unchecked")
public V get() throws InterruptedException, ExecutionException {
return future.get();
try {
return (V) (((Result) future.get()).recreate());
} catch (InterruptedException | ExecutionException e) {
throw e;
} catch (Throwable e) {
throw new RpcException(e);
}
}
@SuppressWarnings("unchecked")
public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
return future.get(timeout, unit);
try {
return (V) (((Result) future.get(timeout, unit)).recreate());
} catch (InterruptedException | ExecutionException | TimeoutException e) {
throw e;
} catch (Throwable e) {
throw new RpcException(e);
}
}
}

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

@ -18,10 +18,10 @@ package org.apache.dubbo.rpc;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.protocol.dubbo.FutureAdapter;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.function.BiConsumer;
/**
@ -139,29 +139,13 @@ public class AsyncRpcResult extends AbstractResult {
@Override
public Object recreate() throws Throwable {
RpcInvocation rpcInvocation = (RpcInvocation) invocation;
FutureAdapter future = new FutureAdapter(this);
RpcContext.getContext().setFuture(future);
if (InvokeMode.FUTURE == rpcInvocation.getInvokeMode()) {
AppResponse appResponse = new AppResponse();
CompletableFuture<Object> future = new CompletableFuture<>();
appResponse.setValue(future);
this.whenComplete((result, t) -> {
if (t != null) {
if (t instanceof CompletionException) {
t = t.getCause();
}
future.completeExceptionally(t);
} else {
if (result.hasException()) {
future.completeExceptionally(result.getException());
} else {
future.complete(result.getValue());
}
}
});
return appResponse.recreate();
} else if (this.isDone()) {
return this.get().recreate();
return future;
}
return (new AppResponse()).recreate();
return getAppResponse().recreate();
}
@Override

33
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/FutureContext.java
View File

@ -29,7 +29,19 @@ import java.util.concurrent.CompletableFuture;
*/
public class FutureContext {
public static InternalThreadLocal<CompletableFuture<?>> futureTL = new InternalThreadLocal<>();
private static InternalThreadLocal<FutureContext> futureTL = new InternalThreadLocal<FutureContext>() {
@Override
protected FutureContext initialValue() {
return new FutureContext();
}
};
public static FutureContext getContext() {
return futureTL.get();
}
private CompletableFuture<?> future;
private CompletableFuture<?> compatibleFuture;
/**
* get future.
@ -38,8 +50,8 @@ public class FutureContext {
* @return future
*/
@SuppressWarnings("unchecked")
public static <T> CompletableFuture<T> getCompletableFuture() {
return (CompletableFuture<T>) futureTL.get();
public <T> CompletableFuture<T> getCompletableFuture() {
return (CompletableFuture<T>) future;
}
/**
@ -47,8 +59,19 @@ public class FutureContext {
*
* @param future
*/
public static void setFuture(CompletableFuture<?> future) {
futureTL.set(future);
public void setFuture(CompletableFuture<?> future) {
this.future = future;
}
@Deprecated
@SuppressWarnings("unchecked")
public <T> CompletableFuture<T> getCompatibleCompletableFuture() {
return (CompletableFuture<T>) compatibleFuture;
}
@Deprecated
public void setCompatibleFuture(CompletableFuture<?> compatibleFuture) {
this.compatibleFuture = compatibleFuture;
}
}

7
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcContext.java
View File

@ -105,6 +105,7 @@ public class RpcContext {
private Object response;
private AsyncContext asyncContext;
protected RpcContext() {
}
@ -224,7 +225,7 @@ public class RpcContext {
*/
@SuppressWarnings("unchecked")
public <T> CompletableFuture<T> getCompletableFuture() {
return FutureContext.getCompletableFuture();
return FutureContext.getContext().getCompletableFuture();
}
/**
@ -235,7 +236,7 @@ public class RpcContext {
*/
@SuppressWarnings("unchecked")
public <T> Future<T> getFuture() {
return FutureContext.getCompletableFuture();
return FutureContext.getContext().getCompletableFuture();
}
/**
@ -244,7 +245,7 @@ public class RpcContext {
* @param future
*/
public void setFuture(CompletableFuture<?> future) {
FutureContext.setFuture(future);
FutureContext.getContext().setFuture(future);
}
public List<URL> getUrls() {

0
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/FutureAdapter.java → dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/FutureAdapter.java
View File

53
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/FutureContextTest.java Normal file
View File

@ -0,0 +1,53 @@
/*
* 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;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.concurrent.CompletableFuture;
/**
*
*/
public class FutureContextTest {
@Test
public void testFutureContext() throws Exception {
Thread thread1 = new Thread(() -> {
FutureContext.getContext().setFuture(CompletableFuture.completedFuture("future from thread1"));
try {
Thread.sleep(500);
Assertions.assertEquals("future from thread1", FutureContext.getContext().getCompletableFuture().get());
} catch (Exception e) {
e.printStackTrace();
}
});
thread1.start();
Thread.sleep(100);
Thread thread2 = new Thread(() -> {
CompletableFuture future = FutureContext.getContext().getCompletableFuture();
Assertions.assertNull(future);
FutureContext.getContext().setFuture(CompletableFuture.completedFuture("future from thread2"));
});
thread2.start();
Thread.sleep(1000);
}
}

6
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java
View File

@ -24,10 +24,10 @@ import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.TimeoutException;
import org.apache.dubbo.remoting.exchange.ExchangeClient;
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.protocol.AbstractInvoker;
@ -91,13 +91,13 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
if (isOneway) {
boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
currentClient.send(inv, isSent);
RpcContext.getContext().setFuture(null);
return AsyncRpcResult.newDefaultAsyncResult(invocation);
} else {
AsyncRpcResult asyncRpcResult = new AsyncRpcResult(inv);
CompletableFuture<Object> responseFuture = currentClient.request(inv, timeout);
asyncRpcResult.subscribeTo(responseFuture);
RpcContext.getContext().setFuture(new FutureAdapter(asyncRpcResult));
// save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter
FutureContext.getContext().setCompatibleFuture(responseFuture);
return asyncRpcResult;
}
} catch (TimeoutException e) {

6
dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftInvoker.java
View File

@ -23,14 +23,13 @@ import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.TimeoutException;
import org.apache.dubbo.remoting.exchange.ExchangeClient;
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.protocol.AbstractInvoker;
import org.apache.dubbo.rpc.protocol.dubbo.FutureAdapter;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@ -97,7 +96,8 @@ public class ThriftInvoker<T> extends AbstractInvoker<T> {
AsyncRpcResult asyncRpcResult = new AsyncRpcResult(invocation);
CompletableFuture<Object> responseFuture = currentClient.request(inv, timeout);
asyncRpcResult.subscribeTo(responseFuture);
RpcContext.getContext().setFuture(new FutureAdapter(asyncRpcResult));
// save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter
FutureContext.getContext().setCompatibleFuture(responseFuture);
return asyncRpcResult;
} catch (TimeoutException e) {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, e.getMessage(), e);