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

Remove processing streamObserver in TripleInvoker#invokeUnary (#10398) 

* Remove processing streamObserver in TripleInvoker#invokeUnary

* remove unaryCall with streamObserver in StubInvocationUtil

* add Compatible with unaryCall

* add unit test
This commit is contained in:
Kunshuai Zhu 2022-08-03 14:08:40 +08:00 committed by GitHub
parent a36c21c277
commit 16fc7b55ef
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 15 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/stub/StubInvocationUtil.java
View File

@ -33,7 +33,13 @@ public class StubInvocationUtil {
public static <T, R> void unaryCall(Invoker<?> invoker, MethodDescriptor method, T request,
StreamObserver<R> responseObserver) {
call(invoker, method, new Object[]{request, responseObserver});
try {
Object res = unaryCall(invoker, method, request);
responseObserver.onNext((R) res);
} catch (Exception e) {
responseObserver.onError(e);
}
responseObserver.onCompleted();
}
public static <T, R> StreamObserver<T> biOrClientStreamCall(Invoker<?> invoker,

9
dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/stub/StubInvocationUtilTest.java
View File

@ -136,13 +136,8 @@ class StubInvocationUtilTest {
Result result = Mockito.mock(Result.class);
String response = "response";
when(invoker.invoke(any(Invocation.class)))
.then(invocationOnMock -> {
Invocation invocation = (Invocation) invocationOnMock.getArguments()[0];
StreamObserver<Object> observer = (StreamObserver<Object>) invocation.getArguments()[1];
observer.onNext(response);
observer.onCompleted();
return result;
});
.then(invocationOnMock -> result);
when(result.recreate()).thenReturn(response);
MethodDescriptor method = Mockito.mock(MethodDescriptor.class);
when(method.getParameterClasses())
.thenReturn(new Class[]{String.class});

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

@ -196,35 +196,15 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
RequestMetadata request = createRequest(methodDescriptor, invocation, timeout);
final Object pureArgument;
if (methodDescriptor.getParameterClasses().length == 2
&& methodDescriptor.getParameterClasses()[1].isAssignableFrom(
StreamObserver.class)) {
StreamObserver<Object> observer = (StreamObserver<Object>) invocation.getArguments()[1];
future.whenComplete((r, t) -> {
if (t != null) {
observer.onError(t);
return;
}
if (r.hasException()) {
observer.onError(r.getException());
return;
}
observer.onNext(r.getValue());
observer.onCompleted();
});
if (methodDescriptor instanceof StubMethodDescriptor) {
pureArgument = invocation.getArguments()[0];
result = new AsyncRpcResult(CompletableFuture.completedFuture(new AppResponse()),
invocation);
} else {
if (methodDescriptor instanceof StubMethodDescriptor) {
pureArgument = invocation.getArguments()[0];
} else {
pureArgument = invocation.getArguments();
}
result = new AsyncRpcResult(future, invocation);
result.setExecutor(callbackExecutor);
FutureContext.getContext().setCompatibleFuture(future);
pureArgument = invocation.getArguments();
}
result = new AsyncRpcResult(future, invocation);
FutureContext.getContext().setCompatibleFuture(future);
result.setExecutor(callbackExecutor);
ClientCall.Listener callListener = new UnaryClientCallListener(future);
final StreamObserver<Object> requestObserver = call.start(request, callListener);