Feat/server end async

motan-core/src/main/java/com/weibo/api/motan/rpc/AbstractProvider.java

@@ -37,21 +37,18 @@ public abstract class AbstractProvider<T> implements Provider<T> {
     protected boolean alive = false;
     protected boolean close = false;
 
-    protected Map<String, Method> methodMap = new HashMap<String, Method>();
+    protected Map<String, Method> methodMap = new HashMap<>();
 
     public AbstractProvider(URL url, Class<T> clz) {
         this.url = url;
         this.clz = clz;
-
-        initMethodMap(clz);
     }
 
     @Override
     public Response call(Request request) {
         MotanFrameworkUtil.logEvent(request, MotanConstants.TRACE_BEFORE_BIZ);
         Response response = invoke(request);
         MotanFrameworkUtil.logEvent(response, MotanConstants.TRACE_AFTER_BIZ);
-
         return response;
     }
 
@@ -78,7 +75,6 @@ public String desc() {
         if (url != null) {
             return url.toString();
         }
-
         return null;
     }
 
@@ -94,7 +90,7 @@ public Class<T> getInterface() {
 
     @Override
     public Method lookupMethod(String methodName, String methodDesc) {
-        Method method = null;
+        Method method;
         String fullMethodName = ReflectUtil.getMethodDesc(methodName, methodDesc);
         method = methodMap.get(fullMethodName);
         if (method == null && StringUtils.isBlank(methodDesc)) {
@@ -103,14 +99,13 @@ public Method lookupMethod(String methodName, String methodDesc) {
                 method = methodMap.get(methodName.substring(0, 1).toLowerCase() + methodName.substring(1));
             }
         }
-
         return method;
     }
 
-    private void initMethodMap(Class<T> clz) {
+    protected void initMethodMap(Class<?> clz) {
         Method[] methods = clz.getMethods();
 
-        List<String> dupList = new ArrayList<String>();
+        List<String> dupList = new ArrayList<>();
         for (Method method : methods) {
             String methodDesc = ReflectUtil.getMethodDesc(method);
             methodMap.put(methodDesc, method);
@@ -126,5 +121,4 @@ private void initMethodMap(Class<T> clz) {
             }
         }
     }
-
 }

motan-core/src/main/java/com/weibo/api/motan/rpc/Callbackable.java

@@ -10,4 +10,8 @@ public interface Callbackable {
     void addFinishCallback(Runnable runnable, Executor executor);
 
     void onFinish();
+
+    default Callbackable getCallbackHolder(){
+        return null;
+    }
 }

motan-core/src/main/java/com/weibo/api/motan/rpc/DefaultCallbackHolder.java

@@ -0,0 +1,73 @@
+/*
+ *
+ *   Copyright 2009-2023 Weibo, Inc.
+ *
+ *     Licensed 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 com.weibo.api.motan.rpc;
+
+import com.weibo.api.motan.util.AsyncUtil;
+import com.weibo.api.motan.util.LoggerUtil;
+import org.apache.commons.lang3.tuple.Pair;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+
+/**
+ * @author zhanglei28
+ * @date 2023/10/13.
+ */
+public class DefaultCallbackHolder implements Callbackable {
+    private final List<Pair<Runnable, Executor>> taskList = new ArrayList<>();
+    private volatile boolean isFinished = false;
+
+    public void addFinishCallback(Runnable runnable, Executor executor) {
+        if (!isFinished) {
+            synchronized (this) {
+                if (!isFinished) {
+                    taskList.add(Pair.of(runnable, executor));
+                    return;
+                }
+            }
+        }
+        process(runnable, executor);
+    }
+
+    @Override
+    public void onFinish() {
+        if (!isFinished) {
+            synchronized (this) {
+                if (!isFinished) {
+                    for (Pair<Runnable, Executor> pair : taskList) {
+                        process(pair.getKey(), pair.getValue());
+                    }
+                    isFinished = true;
+                }
+            }
+        }
+    }
+
+    private void process(Runnable runnable, Executor executor) {
+        if (executor == null) {
+            executor = AsyncUtil.getDefaultCallbackExecutor();
+        }
+        try {
+            executor.execute(runnable);
+        } catch (Exception e) {
+            LoggerUtil.error("Callbackable response exec callback task error, e: ", e);
+        }
+    }
+}

motan-core/src/main/java/com/weibo/api/motan/rpc/DefaultProvider.java

@@ -33,10 +33,25 @@
 @SpiMeta(name = "motan")
 public class DefaultProvider<T> extends AbstractProvider<T> {
     protected T proxyImpl;
+    protected boolean isAsync = false;
 
     public DefaultProvider(T proxyImpl, URL url, Class<T> clz) {
         super(url, clz);
         this.proxyImpl = proxyImpl;
+        Class<?> asyncInterface = null;
+        try {
+            asyncInterface = Class.forName(clz.getName() + "Async");
+            if (asyncInterface.isInterface() && asyncInterface.isAssignableFrom(proxyImpl.getClass())) {
+                isAsync = true;
+            }
+        } catch (Exception ignore) {
+        }
+
+        if (isAsync) {
+            initMethodMap(asyncInterface);
+        } else {
+            initMethodMap(clz);
+        }
     }
 
     @Override
@@ -48,7 +63,11 @@ public T getImpl() {
     public Response invoke(Request request) {
         DefaultResponse response = new DefaultResponse();
 
-        Method method = lookupMethod(request.getMethodName(), request.getParamtersDesc());
+        String methodName = request.getMethodName();
+        if (isAsync) { // change to async method
+            methodName += "Async";
+        }
+        Method method = lookupMethod(methodName, request.getParamtersDesc());
 
         if (method == null) {
             LoggerUtil.error("can not found rpc method:" + request.getMethodName() + ", paramDesc:" + request.getParamtersDesc() + ", service:" + request.getInterfaceName());
@@ -63,6 +82,9 @@ public Response invoke(Request request) {
         boolean defaultThrowExceptionStack = URLParamType.transExceptionStack.getBooleanValue();
         try {
             Object value = method.invoke(proxyImpl, request.getArguments());
+            if (value instanceof ResponseFuture) { // async method
+                return (Response) value;
+            }
             response.setValue(value);
         } catch (Exception e) {
             if (e.getCause() != null) {
@@ -81,9 +103,9 @@ public Response invoke(Request request) {
                 }
             }
             if (logException) {
-                LoggerUtil.error("Exception caught when during method invocation. request:" + request.toString(), e);
+                LoggerUtil.error("Exception caught when during method invocation. request:" + request, e);
             } else {
-                LoggerUtil.info("Exception caught when during method invocation. request:" + request.toString() + ", exception:" + response.getException().getCause().toString());
+                LoggerUtil.info("Exception caught when during method invocation. request:" + request + ", exception:" + response.getException().getCause().toString());
             }
         } catch (Throwable t) {
             // 如果服务发生Error，将Error转化为Exception，防止拖垮调用方
@@ -93,7 +115,7 @@ public Response invoke(Request request) {
                 response.setException(new MotanServiceException("provider has encountered a fatal error!", t));
             }
             //对于Throwable,也记录日志
-            LoggerUtil.error("Exception caught when during method invocation. request:" + request.toString(), t);
+            LoggerUtil.error("Exception caught when during method invocation. request:" + request, t);
         }
 
         if (response.getException() != null) {

motan-core/src/main/java/com/weibo/api/motan/rpc/DefaultResponse.java

@@ -16,21 +16,15 @@
 
 package com.weibo.api.motan.rpc;
 
-import com.weibo.api.motan.core.DefaultThreadFactory;
-import com.weibo.api.motan.core.StandardThreadExecutor;
+import com.weibo.api.motan.exception.MotanBizException;
 import com.weibo.api.motan.exception.MotanServiceException;
 import com.weibo.api.motan.protocol.rpc.RpcProtocolVersion;
-import com.weibo.api.motan.util.LoggerUtil;
-import org.apache.commons.lang3.tuple.Pair;
 
 import java.io.Serializable;
-import java.util.*;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.Executor;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static com.weibo.api.motan.core.StandardThreadExecutor.DEFAULT_MAX_IDLE_TIME;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 /**
  * Response received via rpc.
@@ -40,10 +34,6 @@
  */
 public class DefaultResponse implements Response, Traceable, Callbackable, Serializable {
     private static final long serialVersionUID = 4281186647291615871L;
-    protected static ThreadPoolExecutor defaultCallbackExecutor = new StandardThreadExecutor(20, 200,
-            DEFAULT_MAX_IDLE_TIME, MILLISECONDS, 5000,
-            new DefaultThreadFactory("defaultResponseCallbackPool-", true), new ThreadPoolExecutor.DiscardPolicy());
-
     private Object value;
     private Exception exception;
     private long requestId;
@@ -52,9 +42,8 @@ public class DefaultResponse implements Response, Traceable, Callbackable, Seria
     private Map<String, String> attachments;// rpc协议版本兼容时可以回传一些额外的信息
     private byte rpcProtocolVersion = RpcProtocolVersion.VERSION_1.getVersion();
     private int serializeNumber = 0;// default serialization is hessian2
-    private List<Pair<Runnable, Executor>> taskList = new ArrayList<>();
-    private AtomicBoolean isFinished = new AtomicBoolean();
     private TraceableContext traceableContext = new TraceableContext();
+    private Callbackable callbackHolder = new DefaultCallbackHolder();
 
     public DefaultResponse() {
     }
@@ -63,6 +52,7 @@ public DefaultResponse(long requestId) {
         this.requestId = requestId;
     }
 
+    // for client end. Blocking to get value or throw exception
     public DefaultResponse(Response response) {
         this.value = response.getValue();
         this.exception = response.getException();
@@ -72,10 +62,7 @@ public DefaultResponse(Response response) {
         this.rpcProtocolVersion = response.getRpcProtocolVersion();
         this.serializeNumber = response.getSerializeNumber();
         this.attachments = response.getAttachments();
-        if (response instanceof Traceable) {
-            traceableContext.setReceiveTime(((Traceable) response).getTraceableContext().getReceiveTime());
-            traceableContext.traceInfoMap = ((Traceable) response).getTraceableContext().getTraceInfoMap();
-        }
+        updateTraceableContextFromResponse(response);
     }
 
     public DefaultResponse(Object value) {
@@ -87,6 +74,18 @@ public DefaultResponse(Object value, long requestId) {
         this.requestId = requestId;
     }
 
+    public static DefaultResponse fromServerEndResponseFuture(ResponseFuture responseFuture) {
+        DefaultResponse response = new DefaultResponse();
+        if (responseFuture.getException() != null) { // change to biz exception
+            response.setException(new MotanBizException("provider call process error", responseFuture.getException()));
+        } else {
+            response.setValue(responseFuture.getValue());
+        }
+        response.updateTraceableContextFromResponse(responseFuture);
+        response.updateCallbackHolderFromResponse(responseFuture);
+        return response;
+    }
+
     @Override
     public Object getValue() {
         if (exception != null) {
@@ -175,36 +174,46 @@ public int getSerializeNumber() {
     /**
      * 未指定线程池时，统一使用默认线程池执行。默认线程池满时采用丢弃策略，不保证任务一定会被执行。
      * 如果默认线程池不满足需求时，可以自行携带executor。
+     *
      * @param runnable 准备在response on finish时执行的任务
      * @param executor 指定执行任务的线程池
      */
     public void addFinishCallback(Runnable runnable, Executor executor) {
-        if (!isFinished.get()) {
-            taskList.add(Pair.of(runnable, executor));
-        }
+        callbackHolder.addFinishCallback(runnable, executor);
     }
 
     @Override
     public void onFinish() {
-        if (!isFinished.compareAndSet(false, true)) {
-            return;
-        }
-        for (Pair<Runnable, Executor> pair : taskList) {
-            Runnable runnable = pair.getKey();
-            Executor executor = pair.getValue();
-            if (executor == null) {
-                executor = defaultCallbackExecutor;
-            }
-            try {
-                executor.execute(runnable);
-            } catch (Exception e) {
-                LoggerUtil.error("Callbackable response exec callback task error, e: ", e);
-            }
-        }
+        callbackHolder.onFinish();
     }
 
     @Override
     public TraceableContext getTraceableContext() {
         return traceableContext;
     }
+
+    @Override
+    public Callbackable getCallbackHolder() {
+        return callbackHolder;
+    }
+
+    // only for constructor
+    private void updateTraceableContextFromResponse(Response response) {
+        if (response instanceof Traceable) {
+            TraceableContext tempTraceableContext = ((Traceable) response).getTraceableContext();
+            if (tempTraceableContext != null) {
+                traceableContext = tempTraceableContext;
+            }
+        }
+    }
+
+    // only for constructor
+    private void updateCallbackHolderFromResponse(Response response) {
+        if (response instanceof Callbackable) {
+            Callbackable holder = ((Callbackable) response).getCallbackHolder();
+            if (holder != null) {
+                callbackHolder = holder;
+            }
+        }
+    }
 }