Set callback decode (#7429)

apache · Mar 23, 2021 · 631aedc · 631aedc
1 parent 5af4004
commit 631aedc
Show file tree

Hide file tree

Showing 4 changed files with 45 additions and 41 deletions.
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java
@@ -170,7 +170,7 @@ public Result invoke(Invocation inv) throws RpcException {
         } catch (Throwable e) {
             asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
         }
-        RpcContext.getContext().setFuture(new FutureAdapter(asyncResult.getResponseFuture()));
+        RpcContext.getContext().setFuture(new FutureAdapter<>(asyncResult.getResponseFuture()));
 
         waitForResultIfSync(asyncResult, invocation);
         return asyncResult;

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
@@ -51,9 +51,9 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Map;
+import java.util.concurrent.Executor;
 
 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 GrpcStatus MISSING_RESP = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
@@ -62,9 +62,11 @@ public class ClientStream extends AbstractStream implements Stream {
     private final String authority;
     private final Request request;
     private final RpcInvocation invocation;
+    private final Executor callback;
 
-    public ClientStream(URL url, ChannelHandlerContext ctx, boolean needWrap, Request request) {
+    public ClientStream(URL url, ChannelHandlerContext ctx, boolean needWrap, Request request, Executor callback) {
         super(url, ctx, needWrap);
+        this.callback = callback;
         if (needWrap) {
             setSerializeType((String) ((RpcInvocation) (request.getData())).getObjectAttachment(Constants.SERIALIZATION_KEY));
         }
@@ -106,7 +108,7 @@ public void write(Object obj, ChannelPromise promise) throws IOException {
                 .method(HttpMethod.POST.asciiName())
                 .path("/" + invocation.getObjectAttachment(CommonConstants.PATH_KEY) + "/" + invocation.getMethodName())
                 .set(HttpHeaderNames.CONTENT_TYPE, TripleConstant.CONTENT_PROTO)
-                .set(TripleConstant.TIMEOUT, invocation.get(CommonConstants.TIMEOUT_KEY) +"m")
+                .set(TripleConstant.TIMEOUT, invocation.get(CommonConstants.TIMEOUT_KEY) + "m")
                 .set(HttpHeaderNames.TE, HttpHeaderValues.TRAILERS);
 
         final String version = invocation.getInvoker().getUrl().getVersion();
@@ -165,10 +167,10 @@ public void write(Object obj, ChannelPromise promise) throws IOException {
             ClassLoadUtil.switchContextLoader(tccl);
         }
         final DefaultHttp2DataFrame data = new DefaultHttp2DataFrame(out, true);
-        streamChannel.write(data).addListener(f->{
-            if(f.isSuccess()){
+        streamChannel.write(data).addListener(f -> {
+            if (f.isSuccess()) {
                 promise.trySuccess();
-            }else{
+            } else {
                 promise.tryFailure(f.cause());
             }
         });
@@ -183,10 +185,7 @@ public void halfClose() {
             onError(status);
             return;
         }
-        Http2Headers te = getTe();
-        if (te == null) {
-            te = getHeaders();
-        }
+        final Http2Headers te = getTe() == null ? getHeaders() : getTe();
         final Integer code = te.getInt(TripleConstant.STATUS_KEY);
         if (!GrpcStatus.Code.isOk(code)) {
             final GrpcStatus status = GrpcStatus.fromCode(code)
@@ -196,38 +195,40 @@ public void halfClose() {
         }
         final InputStream data = getData();
         if (data == null) {
-            responseErr(getCtx(), MISSING_RESP);
+            onError(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());
+        callback.execute(() -> {
+            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);
             }
-            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-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleClientHandler.java b/...dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleClientHandler.java
@@ -29,6 +29,7 @@
 import io.netty.util.ReferenceCountUtil;
 
 import java.io.IOException;
+import java.util.concurrent.Executor;
 
 public class TripleClientHandler extends ChannelDuplexHandler {
 
@@ -58,7 +59,8 @@ private void writeRequest(ChannelHandlerContext ctx, final Request req, ChannelP
         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);
+        final Executor callback = (Executor) inv.getAttributes().remove("callback.executor");
+        ClientStream clientStream = new ClientStream(url, ctx, needWrapper, req,callback);
         clientStream.write(req, promise);
     }
 }
diff --git a/...o-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java b/...o-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
@@ -101,6 +101,7 @@ protected Result doInvoke(final Invocation invocation) throws Throwable {
             FutureContext.getContext().setCompatibleFuture(respFuture);
             AsyncRpcResult result = new AsyncRpcResult(respFuture, inv);
             result.setExecutor(executor);
+            inv.put("callback.executor",executor );
 
 
             if (!connection.isAvailable()) {