Supports Provider-side asynchronous

example/src/test/java/com/alipay/sofa/rpc/serviceasync/HelloService.java

@@ -0,0 +1,21 @@
+/*
+ * 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 com.alipay.sofa.rpc.serviceasync;
+
+public interface HelloService {
+    String sayHello(String string);
+}

example/src/test/java/com/alipay/sofa/rpc/serviceasync/HelloServiceImpl.java

@@ -0,0 +1,35 @@
+/*
+ * 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 com.alipay.sofa.rpc.serviceasync;
+
+import com.alipay.sofa.rpc.message.bolt.BoltAsyncContext;
+
+public class HelloServiceImpl implements HelloService {
+    @Override
+    public String sayHello(String name) {
+        BoltAsyncContext boltAsyncContext = new BoltAsyncContext();
+        new Thread(() -> {
+            // 如果需要在新线程中使用调用上下文，需要调用signalContextSwitch方法
+            boltAsyncContext.signalContextSwitch();
+
+            boltAsyncContext.write("Hello " + name);
+
+            boltAsyncContext.resetContext();
+        }).start();
+        return null;
+    }
+}

example/src/test/java/com/alipay/sofa/rpc/serviceasync/start/Client.java

@@ -0,0 +1,52 @@
+/*
+ * 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 com.alipay.sofa.rpc.serviceasync.start;
+
+import com.alipay.sofa.rpc.config.ConsumerConfig;
+import com.alipay.sofa.rpc.log.Logger;
+import com.alipay.sofa.rpc.log.LoggerFactory;
+import com.alipay.sofa.rpc.serviceasync.HelloService;
+
+public class Client {
+    private final static Logger LOGGER = LoggerFactory.getLogger(Client.class);
+
+    public static void main(String[] args) {
+
+        ConsumerConfig<HelloService> consumerConfig = new ConsumerConfig<HelloService>()
+            .setInterfaceId(HelloService.class.getName()) // 指定接口
+            .setProtocol("bolt") // 指定协议
+            .setDirectUrl("bolt://127.0.0.1:12200") // 指定直连地址
+            .setConnectTimeout(10 * 1000);
+
+        HelloService helloService = consumerConfig.refer();
+
+        while (true) {
+            try {
+                LOGGER.info(helloService.sayHello("world"));
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+
+            try {
+                Thread.sleep(2000);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+
+        }
+    }
+}

example/src/test/java/com/alipay/sofa/rpc/serviceasync/start/Service.java

@@ -0,0 +1,38 @@
+/*
+ * 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 com.alipay.sofa.rpc.serviceasync.start;
+
+import com.alipay.sofa.rpc.config.ProviderConfig;
+import com.alipay.sofa.rpc.config.ServerConfig;
+import com.alipay.sofa.rpc.serviceasync.HelloService;
+import com.alipay.sofa.rpc.serviceasync.HelloServiceImpl;
+
+public class Service {
+    public static void main(String[] args) {
+        ServerConfig serverConfig = new ServerConfig()
+            .setProtocol("bolt") // 设置一个协议，默认bolt
+            .setPort(12200) // 设置一个端口，默认12200
+            .setDaemon(false); // 非守护线程
+
+        ProviderConfig<HelloService> providerConfig = new ProviderConfig<HelloService>()
+            .setInterfaceId(HelloService.class.getName()) // 指定接口
+            .setRef(new HelloServiceImpl()) // 指定实现
+            .setServer(serverConfig); // 指定服务端
+
+        providerConfig.export(); // 发布服务
+    }
+}

remoting/remoting-bolt/src/main/java/com/alipay/sofa/rpc/message/bolt/BoltAsyncContext.java

@@ -0,0 +1,116 @@
+/*
+ * 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 com.alipay.sofa.rpc.message.bolt;
+
+import com.alipay.remoting.AsyncContext;
+import com.alipay.sofa.rpc.common.RemotingConstants;
+import com.alipay.sofa.rpc.common.RpcConstants;
+import com.alipay.sofa.rpc.context.RpcInternalContext;
+import com.alipay.sofa.rpc.context.RpcInvokeContext;
+import com.alipay.sofa.rpc.core.exception.SofaRpcException;
+import com.alipay.sofa.rpc.core.request.SofaRequest;
+import com.alipay.sofa.rpc.core.response.SofaResponse;
+import com.alipay.sofa.rpc.event.EventBus;
+import com.alipay.sofa.rpc.event.ServerEndHandleEvent;
+import com.alipay.sofa.rpc.event.ServerSendEvent;
+
+public class BoltAsyncContext {
+    private volatile boolean         sent = false;
+
+    private final AsyncContext       asyncContext;
+    private final SofaRequest        sofaRequest;
+
+    private final RpcInternalContext internalContext;
+    private final RpcInvokeContext   invokeCtx;
+    private final ClassLoader        restoreClassLoader;
+    private volatile ClassLoader     stagedClassLoader;
+
+    public BoltAsyncContext() {
+        internalContext = RpcInternalContext.getContext();
+        asyncContext = (AsyncContext) internalContext.getAttachment(RpcConstants.HIDDEN_KEY_ASYNC_CONTEXT);
+        sofaRequest = (SofaRequest) internalContext.getAttachment(RpcConstants.HIDDEN_KEY_ASYNC_REQUEST);
+        invokeCtx = RpcInvokeContext.getContext();
+        restoreClassLoader = Thread.currentThread().getContextClassLoader();
+        invokeCtx.put(RemotingConstants.INVOKE_CTX_IS_ASYNC_CHAIN, true);
+    }
+
+    public synchronized void signalContextSwitch() {
+        RpcInvokeContext.setContext(invokeCtx);
+        RpcInternalContext.setContext(internalContext);
+        if (restoreClassLoader != null) {
+            Thread.currentThread().setContextClassLoader(restoreClassLoader);
+            stagedClassLoader = Thread.currentThread().getContextClassLoader();
+        }
+    }
+
+    public synchronized void resetContext() {
+        RpcInvokeContext.removeContext();
+        RpcInternalContext.removeContext();
+        // 修复 ClassLoader 恢复逻辑
+        if (stagedClassLoader != null) {
+            Thread.currentThread().setContextClassLoader(stagedClassLoader);
+        }
+    }
+
+    private synchronized void checkState() {
+        if (sent) {
+            throw new IllegalStateException("Current async context has already sent response");
+        }
+        sent = true;
+    }
+
+    public void write(Object result) {
+        checkState();
+        try {
+            SofaResponse response = new SofaResponse();
+            response.setAppResponse(result);
+            sendResponse(response, null);
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to send async response", e);
+        }
+    }
+
+    public void writeException(Throwable throwable) {
+        checkState();
+        try {
+            SofaResponse response = new SofaResponse();
+            if (throwable instanceof SofaRpcException) {
+                SofaRpcException sofaRpcException = (SofaRpcException) throwable;
+                response.setErrorMsg(sofaRpcException.getMessage());
+                sendResponse(response, sofaRpcException);
+            } else {
+                response.setAppResponse(throwable);
+                sendResponse(response, null);
+            }
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to send async exception response", e);
+        }
+    }
+
+    private void sendResponse(SofaResponse response, SofaRpcException sofaRpcException) {
+        try {
+            asyncContext.sendResponse(response);
+        } finally {
+            if (EventBus.isEnable(ServerSendEvent.class)) {
+                EventBus.post(new ServerSendEvent(sofaRequest, response, sofaRpcException));
+            }
+            if (EventBus.isEnable(ServerEndHandleEvent.class)) {
+                EventBus.post(new ServerEndHandleEvent());
+            }
+        }
+    }
+}

remoting/remoting-bolt/src/main/java/com/alipay/sofa/rpc/server/bolt/BoltServerProcessor.java

@@ -110,11 +110,13 @@ public void handleRequest(BizContext bizCtx, AsyncContext asyncCtx, SofaRequest
 
         // 是否链路异步化中
         boolean isAsyncChain = false;
+
         try { // 这个 try-finally 为了保证Context一定被清理
             processingCount.incrementAndGet(); // 统计值加1
 
             context.setRemoteAddress(bizCtx.getRemoteHost(), bizCtx.getRemotePort()); // 远程地址
             context.setAttachment(RpcConstants.HIDDEN_KEY_ASYNC_CONTEXT, asyncCtx); // 远程返回的通道
+            context.setAttachment(RpcConstants.HIDDEN_KEY_ASYNC_REQUEST, request);
 
             InvokeContext boltInvokeCtx = bizCtx.getInvokeContext();
             if (RpcInternalContext.isAttachmentEnable()) {
@@ -192,7 +194,7 @@ public void handleRequest(BizContext bizCtx, AsyncContext asyncCtx, SofaRequest
                 RpcInvokeContext invokeContext = RpcInvokeContext.peekContext();
                 isAsyncChain = CommonUtils.isTrue(invokeContext != null ?
                     (Boolean) invokeContext.remove(RemotingConstants.INVOKE_CTX_IS_ASYNC_CHAIN) : null);
-                // 如果是服务端异步代理模式，特殊处理，因为该模式是在业务代码自主异步返回的
+                // 如果是服务端异步代理模式或服务端模式，特殊处理，因为该模式是在业务代码自主异步返回的
                 if (!isAsyncChain) {
                     // 其它正常请求
                     try { // 这个try-catch 保证一定要记录tracer