Compatible with CompleteFuture asynchronous programming

core/api/src/main/java/com/alipay/sofa/rpc/message/AbstractResponseFuture.java

@@ -18,16 +18,13 @@
 
 import com.alipay.sofa.rpc.context.RpcRuntimeContext;
 
-import java.util.concurrent.CancellationException;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
 
 /**
  * @author <a href="mailto:[email protected]">GengZhang</a>
  * @since 5.4.0
  */
-public abstract class AbstractResponseFuture<V> implements ResponseFuture<V> {
+public abstract class AbstractResponseFuture<V> extends CompletableFuture<V> implements ResponseFuture<V> {
 
     protected static final CancellationException CANCELLATION_CAUSE = new CancellationException();
 

core/api/src/main/java/com/alipay/sofa/rpc/message/ResponseFuture.java

@@ -19,14 +19,15 @@
 import com.alipay.sofa.rpc.core.invoke.SofaResponseCallback;
 
 import java.util.List;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Future;
 
 /**
  * 响应Future，可以调用get方法进行获取响应，也可以注入监听器，有结果或者都会通知
  *
  * @author <a href=mailto:[email protected]>GengZhang</a>
  */
-public interface ResponseFuture<V> extends Future<V> {
+public interface ResponseFuture<V> extends Future<V>, CompletionStage<V> {
 
     /**
      * 增加多个响应监听器

core/api/src/test/java/com/alipay/sofa/rpc/context/RpcInternalContextTest.java

@@ -18,6 +18,7 @@
 
 import com.alipay.sofa.rpc.client.ProviderHelper;
 import com.alipay.sofa.rpc.core.invoke.SofaResponseCallback;
+import com.alipay.sofa.rpc.message.AbstractResponseFuture;
 import com.alipay.sofa.rpc.message.ResponseFuture;
 import org.junit.After;
 import org.junit.Assert;
@@ -136,48 +137,69 @@ public void testAttachment() {
         Assert.assertNull(context.removeAttachment("11"));
     }
 
-    @Test
-    public void testClear() {
-        RpcInternalContext context = RpcInternalContext.getContext();
-        context.setRemoteAddress("127.0.0.1", 1234);
-        context.setLocalAddress("127.0.0.1", 2345);
-        context.setFuture(new ResponseFuture<String>() {
-            @Override
-            public boolean cancel(boolean mayInterruptIfRunning) {
-                return false;
-            }
+    private static class MyResponseFuture extends AbstractResponseFuture<String> {
 
-            @Override
-            public boolean isCancelled() {
-                return false;
-            }
+        /**
+         * 构造函数
+         *
+         * @param timeout
+         */
+        public MyResponseFuture(int timeout) {
+            super(timeout);
+        }
 
-            @Override
-            public boolean isDone() {
-                return false;
-            }
+        @Override
+        public ResponseFuture addListeners(List<SofaResponseCallback> sofaResponseCallbacks) {
+            return null;
+        }
 
-            @Override
-            public String get() throws InterruptedException, ExecutionException {
-                return null;
-            }
+        @Override
+        public ResponseFuture addListener(SofaResponseCallback sofaResponseCallback) {
+            return null;
+        }
 
-            @Override
-            public String get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException,
-                TimeoutException {
-                return null;
-            }
+        @Override
+        public boolean cancel(boolean mayInterruptIfRunning) {
+            return false;
+        }
 
-            @Override
-            public ResponseFuture addListeners(List<SofaResponseCallback> sofaResponseCallbacks) {
-                return null;
-            }
+        @Override
+        public boolean isCancelled() {
+            return false;
+        }
 
-            @Override
-            public ResponseFuture addListener(SofaResponseCallback sofaResponseCallback) {
-                return null;
-            }
-        });
+        @Override
+        public boolean isDone() {
+            return false;
+        }
+
+        @Override
+        public void notifyListeners() {
+
+        }
+
+        @Override
+        public String get() throws InterruptedException, ExecutionException {
+            return null;
+        }
+
+        @Override
+        protected String getNow() throws ExecutionException {
+            return null;
+        }
+
+        @Override
+        protected void releaseIfNeed(Object result) {
+
+        }
+    }
+
+    @Test
+    public void testClear() {
+        RpcInternalContext context = RpcInternalContext.getContext();
+        context.setRemoteAddress("127.0.0.1", 1234);
+        context.setLocalAddress("127.0.0.1", 2345);
+        context.setFuture(new MyResponseFuture(1));
 
         context.setProviderInfo(ProviderHelper.toProviderInfo("127.0.0.1:80"));
         context.setAttachment("_xxxx", "yyyy");

fault/fault-hystrix/src/main/java/com/alipay/sofa/rpc/hystrix/HystrixResponseFuture.java

@@ -21,17 +21,14 @@
 import com.netflix.hystrix.HystrixCommand;
 
 import java.util.List;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
 
 /**
  * the {@link Future}(from {@link HystrixCommand#queue()}) wrapper that can be used as a {@link ResponseFuture}
  *
  * @author <a href=mailto:[email protected]>ScienJus</a>
  */
-public class HystrixResponseFuture implements ResponseFuture {
+public class HystrixResponseFuture extends CompletableFuture implements ResponseFuture {
 
     private Future delegate;
 

fault/fault-tolerance/src/test/java/com/alipay/sofa/rpc/client/aft/InvocationStatDimensionStatTest.java

@@ -26,6 +26,7 @@
 import com.alipay.sofa.rpc.core.exception.SofaTimeOutException;
 import com.alipay.sofa.rpc.core.invoke.SofaResponseCallback;
 import com.alipay.sofa.rpc.core.request.RequestBase;
+import com.alipay.sofa.rpc.message.AbstractResponseFuture;
 import com.alipay.sofa.rpc.message.ResponseFuture;
 import org.junit.Assert;
 import org.junit.Before;
@@ -231,49 +232,81 @@ public void testRegulationEffective() throws InterruptedException {
 
     }
 
+    private static class MyResponseFuture extends AbstractResponseFuture<String> {
+
+        /**
+         * 构造函数
+         *
+         * @param timeout
+         */
+        public MyResponseFuture(int timeout) {
+            super(timeout);
+        }
+
+        @Override
+        public ResponseFuture addListeners(List<SofaResponseCallback> sofaResponseCallbacks) {
+            return null;
+        }
+
+        @Override
+        public ResponseFuture addListener(SofaResponseCallback sofaResponseCallback) {
+            return null;
+        }
+
+        @Override
+        public boolean cancel(boolean mayInterruptIfRunning) {
+            return false;
+        }
+
+        @Override
+        public boolean isCancelled() {
+            return false;
+        }
+
+        @Override
+        public boolean isDone() {
+            return false;
+        }
+
+        @Override
+        public void notifyListeners() {
+
+        }
+
+        @Override
+        public String get() throws InterruptedException, ExecutionException {
+            return null;
+        }
+
+        @Override
+        public String get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException,
+            TimeoutException {
+            return null;
+        }
+
+        @Override
+        protected String getNow() throws ExecutionException {
+            return null;
+        }
+
+        @Override
+        protected void releaseIfNeed(Object result) {
+
+        }
+    }
+
     private void prepareInvokeContext() {
         final RpcInvokeContext context = new RpcInvokeContext();
         context.setResponseCallback(new SofaResponseCallback() {
             @Override
             public void onAppResponse(final Object appResponse, String methodName, RequestBase request) {
                 //放到 future 中方便测试.
                 LOGGER.info("回调成功" + appResponse);
-                context.setFuture(new ResponseFuture<String>() {
-                    @Override
-                    public ResponseFuture addListeners(List<SofaResponseCallback> sofaResponseCallbacks) {
-                        return null;
-                    }
-
-                    @Override
-                    public ResponseFuture addListener(SofaResponseCallback sofaResponseCallback) {
-                        return null;
-                    }
-
-                    @Override
-                    public boolean cancel(boolean mayInterruptIfRunning) {
-                        return false;
-                    }
-
-                    @Override
-                    public boolean isCancelled() {
-                        return false;
-                    }
-
-                    @Override
-                    public boolean isDone() {
-                        return false;
-                    }
-
+                context.setFuture(new MyResponseFuture(1) {
                     @Override
                     public String get() throws InterruptedException, ExecutionException {
                         return (String) appResponse;
                     }
-
-                    @Override
-                    public String get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException,
-                        TimeoutException {
-                        return null;
-                    }
                 });
             }
 

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/async/AsyncChainTest.java

@@ -91,6 +91,7 @@ public void testAll() {
 
         // 链路异步化调用--正常
         RpcInvokeContext.getContext().setResponseCallback(buildCallback(ret, latch));
+
         String ret0 = asyncHelloService.sayHello("xxx", 22);
         Assert.assertNull(ret0); // 第一次返回null
         try {

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/invoke/FutureTest.java

@@ -25,9 +25,12 @@
 import com.alipay.sofa.rpc.context.RpcInvokeContext;
 import com.alipay.sofa.rpc.core.exception.SofaRpcException;
 import com.alipay.sofa.rpc.filter.Filter;
+import com.alipay.sofa.rpc.log.Logger;
+import com.alipay.sofa.rpc.log.LoggerFactory;
 import com.alipay.sofa.rpc.test.ActivelyDestroyTest;
 import com.alipay.sofa.rpc.test.HelloService;
 import com.alipay.sofa.rpc.test.HelloServiceImpl;
+import com.alipay.sofa.rpc.triple.TripleHessianInvokeTest;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -44,6 +47,8 @@
  */
 public class FutureTest extends ActivelyDestroyTest {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(FutureTest.class);
+
     @Test
     public void testAll() {
 
@@ -152,6 +157,10 @@ public void testAll() {
             String ret = helloService.sayHello("xxx", 22);
             Assert.assertNull(ret); // 第一次返回null
 
+            RpcInvokeContext.getContext().getFuture().thenAccept(req->{
+                LOGGER.info("FutureTest RpcInvokeContext CompletableFuture result: {}", req);
+            });
+
             Thread.sleep(1500); // 1s 过去，被rpc设置超时了
             Future future = SofaResponseFuture.getFuture();
             ret = (String) future.get();

test/test-integration/src/test/java/com/alipay/sofa/rpc/transport/http/Http2ClearTextHessianTest.java

@@ -25,6 +25,8 @@
 import com.alipay.sofa.rpc.core.exception.SofaRpcException;
 import com.alipay.sofa.rpc.core.invoke.SofaResponseCallback;
 import com.alipay.sofa.rpc.core.request.RequestBase;
+import com.alipay.sofa.rpc.log.Logger;
+import com.alipay.sofa.rpc.log.LoggerFactory;
 import com.alipay.sofa.rpc.message.ResponseFuture;
 import com.alipay.sofa.rpc.server.bolt.pb.EchoRequest;
 import com.alipay.sofa.rpc.server.bolt.pb.Group;
@@ -43,6 +45,8 @@
  */
 public class Http2ClearTextHessianTest extends ActivelyDestroyTest {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(Http2ClearTextHessianTest.class);
+
     @Test
     public void testHessian() {
         // 只有1个线程 执行
@@ -116,6 +120,9 @@ public void testHessian() {
             ResponseFuture<ExampleObj> future = RpcInvokeContext.getContext().getFuture();
             try {
                 response = future.get();
+                RpcInvokeContext.getContext().getFuture().thenAccept(req->{
+                    LOGGER.info("Http2ClearTextHessianTest CompletableFuture result: {}", req);
+                });
                 Assert.assertEquals(200, response.getId());
                 Assert.assertEquals("yyyxx", response.getName());
             } catch (Exception e) {