[SCB-1184] add IT for the case that provider's response type is CompletableFuture<ResponseEntity>

integration-tests/it-consumer/src/main/java/org/apache/servicecomb/it/ConsumerMain.java

@@ -25,6 +25,7 @@
 import org.apache.servicecomb.it.testcase.TestAcceptType;
 import org.apache.servicecomb.it.testcase.TestAnnotatedAttribute;
 import org.apache.servicecomb.it.testcase.TestApiParam;
+import org.apache.servicecomb.it.testcase.TestAsyncInvoke;
 import org.apache.servicecomb.it.testcase.TestChangeTransport;
 import org.apache.servicecomb.it.testcase.TestDataTypePrimitive;
 import org.apache.servicecomb.it.testcase.TestDefaultJsonValueJaxrsSchema;
@@ -125,6 +126,8 @@ private static void runShareTestCases() throws Throwable {
     ITJUnitUtils.run(TestDefaultJsonValueJaxrsSchema.class);
     ITJUnitUtils.run(TestRestController.class);
     ITJUnitUtils.runWithRest(TestRestController.class);
+
+    ITJUnitUtils.runWithHighwayAndRest(TestAsyncInvoke.class);
   }
 
   interface ITTask {

integration-tests/it-consumer/src/main/java/org/apache/servicecomb/it/deploy/Deploys.java

@@ -192,7 +192,7 @@ private void initBaseProducer() {
     MicroserviceDeployDefinition definition = new MicroserviceDeployDefinition();
     definition.setDeployName("baseProducer");
     definition.setCmd("it-producer");
-    definition.setArgs(new String[] {});
+    definition.setArgs(new String[] {"-Xmx128m"});
     definition.setAppId("integration-test");
     definition.setMicroserviceName("it-producer");
     definition.setVersion(DEFAULT_MICROSERVICE_VERSION);
@@ -206,11 +206,12 @@ private void initBaseHttp2Producer() {
     MicroserviceDeployDefinition definition = new MicroserviceDeployDefinition();
     definition.setDeployName("baseHttp2Producer");
     definition.setCmd("it-producer");
-    definition.setArgs(new String[] {});
+    definition.setArgs(new String[] {"-Xmx128m"});
     URL urlServer = Thread.currentThread().getContextClassLoader().getResource("certificates/server.p12");
     URL urlTrust = Thread.currentThread().getContextClassLoader().getResource("certificates/trust.jks");
     if (urlServer != null && urlTrust != null) {
       definition.setArgs(new String[] {"-Dservicecomb.rest.address=0.0.0.0:0?sslEnabled=true&protocol=http2",
+          "-Xmx128m",
           "-Dservicecomb.highway.address=0.0.0.0:0?sslEnabled=true",
           "-Dserver.p12=" + urlServer.getPath(),
           "-Dtrust.jks=" + urlTrust.getPath()
@@ -229,7 +230,7 @@ private void initBaseHttp2CProducer() {
     MicroserviceDeployDefinition definition = new MicroserviceDeployDefinition();
     definition.setDeployName("baseHttp2CProducer");
     definition.setCmd("it-producer");
-    definition.setArgs(new String[] {"-Dservicecomb.rest.address=0.0.0.0:0?protocol=http2"});
+    definition.setArgs(new String[] {"-Dservicecomb.rest.address=0.0.0.0:0?protocol=http2", "-Xmx128m"});
     definition.setAppId("integration-test");
     definition.setMicroserviceName("it-producer-h2c");
     definition.setVersion(DEFAULT_MICROSERVICE_VERSION);
@@ -243,7 +244,7 @@ private void initSpringBoot2ServletProducer() {
     MicroserviceDeployDefinition definition = new MicroserviceDeployDefinition();
     definition.setDeployName("springBoot2ServletProducer");
     definition.setCmd("it-producer-deploy-springboot2-servlet");
-    definition.setArgs(new String[] {});
+    definition.setArgs(new String[] {"-Xmx128m"});
     definition.setAppId("integration-test");
     definition.setMicroserviceName("it-producer-deploy-springboot2-servlet");
     definition.setVersion(DEFAULT_MICROSERVICE_VERSION);
@@ -257,7 +258,7 @@ private void initSpringBoot2StandaloneProducer() {
     MicroserviceDeployDefinition definition = new MicroserviceDeployDefinition();
     definition.setDeployName("springBoot2StandaloneProducer");
     definition.setCmd("it-producer-deploy-springboot2-standalone");
-    definition.setArgs(new String[] {});
+    definition.setArgs(new String[] {"-Xmx128m"});
     definition.setAppId("integration-test");
     definition.setMicroserviceName("it-producer-deploy-springboot2-standalone");
     definition.setVersion(DEFAULT_MICROSERVICE_VERSION);
@@ -271,7 +272,7 @@ private void initEdge() {
     MicroserviceDeployDefinition definition = new MicroserviceDeployDefinition();
     definition.setDeployName("edge");
     definition.setCmd("it-edge");
-    definition.setArgs(new String[] {});
+    definition.setArgs(new String[] {"-Xmx128m"});
     definition.setAppId("integration-test");
     definition.setMicroserviceName("it-edge");
     definition.setVersion(DEFAULT_MICROSERVICE_VERSION);

integration-tests/it-consumer/src/main/java/org/apache/servicecomb/it/schema/DefaultJsonValueResponse.java

@@ -0,0 +1,39 @@
+/*
+ * 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 org.apache.servicecomb.it.schema;
+
+public class DefaultJsonValueResponse {
+  private int type;
+
+  private String message;
+
+  public int getType() {
+    return type;
+  }
+
+  public void setType(int type) {
+    this.type = type;
+  }
+
+  public String getMessage() {
+    return message;
+  }
+
+  public void setMessage(String message) {
+    this.message = message;
+  }
+}

integration-tests/it-consumer/src/main/java/org/apache/servicecomb/it/testcase/TestAsyncInvoke.java

@@ -0,0 +1,149 @@
+/*
+ * 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 org.apache.servicecomb.it.testcase;
+
+import static org.junit.Assert.fail;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+
+import javax.xml.ws.Holder;
+
+import org.apache.servicecomb.it.Consumers;
+import org.apache.servicecomb.it.junit.ITJUnitUtils;
+import org.apache.servicecomb.it.schema.DefaultJsonValueResponse;
+import org.apache.servicecomb.provider.springmvc.reference.async.CseAsyncRestTemplate;
+import org.hamcrest.Matchers;
+import org.junit.Assert;
+import org.junit.Test;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.concurrent.ListenableFuture;
+
+public class TestAsyncInvoke {
+
+  private CseAsyncRestTemplate cseAsyncRestTemplate = new CseAsyncRestTemplate();
+
+  interface DataTypeAsyncIntf {
+    CompletableFuture<ResponseEntity<String>> responseEntityString();
+
+    CompletableFuture<ResponseEntity<DefaultJsonValueResponse>> responseEntityDataObject();
+  }
+
+  private static Consumers<DataTypeAsyncIntf> dataTypeAsyncJaxrs =
+      new Consumers<>("dataTypeAsyncJaxrs", DataTypeAsyncIntf.class);
+
+  @Test
+  public void responseEntity_string_intf() {
+    BiConsumer<ResponseEntity<String>, Throwable> checkLogic = (responseEntity, ex) -> {
+      Assert.assertEquals(203, responseEntity.getStatusCodeValue());
+      Assert.assertThat(responseEntity.getHeaders().get("testH"), Matchers.containsInAnyOrder("testV1", "testV2"));
+      Assert.assertEquals("TestOK", responseEntity.getBody());
+      Assert.assertNull(ex);
+    };
+
+    // Async RPC
+    CompletableFuture<ResponseEntity<String>> responseEntityCompletableFuture =
+        dataTypeAsyncJaxrs.getIntf().responseEntityString();
+    check(responseEntityCompletableFuture, checkLogic);
+
+    // RestTemplate
+    ResponseEntity<String> result = dataTypeAsyncJaxrs.getSCBRestTemplate()
+        .getForEntity("/responseEntityString", String.class);
+    checkLogic.accept(result, null);
+
+    // AsyncRestTemplate
+    ListenableFuture<ResponseEntity<String>> responseEntityListenableFuture = cseAsyncRestTemplate
+        .getForEntity("cse://" + ITJUnitUtils.getProducerName() + "/v1/dataTypeAsyncJaxrs/responseEntityString",
+            String.class);
+    checkAsyncRt(responseEntityListenableFuture, checkLogic);
+
+    ResponseEntity<String> edgeResponseEntity = dataTypeAsyncJaxrs.getEdgeRestTemplate()
+        .getForEntity("/responseEntityString", String.class);
+    checkLogic.accept(edgeResponseEntity, null);
+  }
+
+  @Test
+  public void responseEntity_dataObject_intf() {
+    BiConsumer<ResponseEntity<DefaultJsonValueResponse>, Throwable> checkLogic = (responseEntity, ex) -> {
+      Assert.assertEquals(203, responseEntity.getStatusCodeValue());
+      Assert.assertThat(responseEntity.getHeaders().get("testH"), Matchers.containsInAnyOrder("testV1", "testV2"));
+      Assert.assertEquals(DefaultJsonValueResponse.class, responseEntity.getBody().getClass());
+      Assert.assertEquals("TestOK", responseEntity.getBody().getMessage());
+      Assert.assertEquals(2, responseEntity.getBody().getType());
+      Assert.assertNull(ex);
+    };
+
+    // Async RPC
+    CompletableFuture<ResponseEntity<DefaultJsonValueResponse>> responseEntityCompletableFuture =
+        dataTypeAsyncJaxrs.getIntf().responseEntityDataObject();
+    check(responseEntityCompletableFuture, checkLogic);
+
+    // RestTemplate
+    ResponseEntity<DefaultJsonValueResponse> result =
+        dataTypeAsyncJaxrs.getSCBRestTemplate()
+            .getForEntity("/responseEntityDataObject", DefaultJsonValueResponse.class);
+    checkLogic.accept(result, null);
+
+    // AsyncRestTemplate
+    ListenableFuture<ResponseEntity<DefaultJsonValueResponse>> responseEntityListenableFuture = cseAsyncRestTemplate
+        .getForEntity("cse://" + ITJUnitUtils.getProducerName() + "/v1/dataTypeAsyncJaxrs/responseEntityDataObject",
+            DefaultJsonValueResponse.class);
+    checkAsyncRt(responseEntityListenableFuture, checkLogic);
+
+    ResponseEntity<DefaultJsonValueResponse> edgeResponseEntity = dataTypeAsyncJaxrs.getEdgeRestTemplate()
+        .getForEntity("/responseEntityDataObject", DefaultJsonValueResponse.class);
+    checkLogic.accept(edgeResponseEntity, null);
+  }
+
+  /**
+   * Wait for response and check
+   */
+  private <T> void check(CompletableFuture<ResponseEntity<T>> responseEntityCompletableFuture,
+      BiConsumer<ResponseEntity<T>, Throwable> checkLogic) {
+    CountDownLatch countDownLatch = new CountDownLatch(1);
+    Holder<Boolean> responseChecked = new Holder<>();
+
+    responseEntityCompletableFuture.whenComplete((responseEntity, ex) -> {
+      checkLogic.accept(responseEntity, ex);
+      responseChecked.value = true;
+    });
+
+    try {
+      countDownLatch.await(3000, TimeUnit.MILLISECONDS);
+    } catch (InterruptedException e) {
+      fail("error occurs while waiting for response, " + e.getMessage());
+    }
+
+    Assert.assertTrue("response check unfinished!", responseChecked.value);
+  }
+
+  /**
+   * Transfer {@link ListenableFuture} to {@link CompletableFuture}, wait for response and check
+   */
+  private <T> void checkAsyncRt(ListenableFuture<ResponseEntity<T>> responseEntityListenableFuture,
+      BiConsumer<ResponseEntity<T>, Throwable> checkLogic) {
+    CompletableFuture<ResponseEntity<T>> entityCompletableFuture = new CompletableFuture<>();
+    responseEntityListenableFuture.addCallback(
+        entityCompletableFuture::complete,
+        entityCompletableFuture::completeExceptionally
+    );
+    check(entityCompletableFuture, checkLogic);
+  }
+}

integration-tests/it-producer/src/main/java/org/apache/servicecomb/it/schema/DataTypeAsyncJaxrsSchema.java

@@ -0,0 +1,53 @@
+/*
+ * 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 org.apache.servicecomb.it.schema;
+
+import java.util.concurrent.CompletableFuture;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+
+import org.apache.servicecomb.provider.rest.common.RestSchema;
+import org.springframework.http.ResponseEntity;
+
+@RestSchema(schemaId = "dataTypeAsyncJaxrs")
+@Path("/v1/dataTypeAsyncJaxrs")
+public class DataTypeAsyncJaxrsSchema {
+  @Path("/responseEntityString")
+  @GET
+  public CompletableFuture<ResponseEntity<String>> responseEntityString() {
+    CompletableFuture<ResponseEntity<String>> responseFuture = new CompletableFuture<>();
+    ResponseEntity<String> responseEntity =
+        ResponseEntity.status(203).header("testH", "testV1", "testV2").body("TestOK");
+    responseFuture.complete(responseEntity);
+    return responseFuture;
+  }
+
+  @Path("/responseEntityDataObject")
+  @GET
+  public CompletableFuture<ResponseEntity<DefaultJsonValueResponse>> responseEntityDataObject() {
+    CompletableFuture<ResponseEntity<DefaultJsonValueResponse>> responseFuture = new CompletableFuture<>();
+    DefaultJsonValueResponse body = new DefaultJsonValueResponse();
+    body.setMessage("TestOK");
+    body.setType(2);
+    ResponseEntity<DefaultJsonValueResponse> responseEntity =
+        ResponseEntity.status(203).header("testH", "testV1", "testV2").body(body);
+    responseFuture.complete(responseEntity);
+    return responseFuture;
+  }
+}