feat: Update to non-eager FutureListener

gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonClientCalls.java

@@ -64,7 +64,7 @@ public static <RequestT, ResponseT> HttpJsonClientCall<RequestT, ResponseT> newC
     return httpJsonContext.getChannel().newCall(methodDescriptor, httpJsonContext.getCallOptions());
   }
 
-  static <RequestT, ResponseT> ApiFuture<ResponseT> eagerFutureUnaryCall(
+  static <RequestT, ResponseT> ApiFuture<ResponseT> futureUnaryCall(
       HttpJsonClientCall<RequestT, ResponseT> clientCall, RequestT request) {
     // Start the call
     HttpJsonFuture<ResponseT> future = new HttpJsonFuture<>(clientCall);
@@ -115,21 +115,26 @@ public boolean setException(Throwable throwable) {
 
   private static class FutureListener<T> extends HttpJsonClientCall.Listener<T> {
     private final HttpJsonFuture<T> future;
+    private T message;
+    private boolean isMessageReceived;
 
     private FutureListener(HttpJsonFuture<T> future) {
       this.future = future;
+      this.isMessageReceived = false;
     }
 
     @Override
     public void onMessage(T message) {
-      if (!future.set(message)) {
+      if (isMessageReceived) {
         throw new IllegalStateException("More than one value received for unary call");
       }
+      isMessageReceived = true;
+      this.message = message;
     }
 
     @Override
     public void onClose(int statusCode, HttpJsonMetadata trailers) {
-      if (!future.isDone()) {
+      if (!isMessageReceived) {
         if (trailers == null || trailers.getException() == null) {
           future.setException(
               new HttpJsonStatusRuntimeException(
@@ -140,9 +145,8 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
         } else {
           future.setException(trailers.getException());
         }
-      } else if (statusCode < 200 || statusCode >= 400) {
-        LOGGER.log(
-            Level.WARNING, "Received error for unary call after receiving a successful response");
+      } else {
+        future.set(message);
       }
     }
   }

gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonDirectCallable.java

@@ -65,7 +65,7 @@ public ApiFuture<ResponseT> futureCall(RequestT request, ApiCallContext inputCon
 
     HttpJsonClientCall<RequestT, ResponseT> clientCall =
         HttpJsonClientCalls.newCall(descriptor, context);
-    return HttpJsonClientCalls.eagerFutureUnaryCall(clientCall, request);
+    return HttpJsonClientCalls.futureUnaryCall(clientCall, request);
   }
 
   @Override

gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonClientInterceptorTest.java

@@ -56,9 +56,6 @@
 @RunWith(JUnit4.class)
 public class HttpJsonClientInterceptorTest {
 
-  private static final int NUM_RETRIES = 10;
-  private static final int BUSY_WAIT_TIME_IN_MS = 1000;
-
   private static class CapturingClientInterceptor implements HttpJsonClientInterceptor {
     // Manually capturing arguments instead of using Mockito. This is intentional, as this
     // specific test interceptor class represents a typical interceptor implementation. Doing the
@@ -213,21 +210,8 @@ public void testCustomInterceptor() throws ExecutionException, InterruptedExcept
     // Test that internal interceptor worked (the one which inserts headers)
     assertThat(headerValue).isEqualTo("headerValue");
 
-    // Test that the custom interceptor was called
-    // These two tests cases are guaranteed to respond back by the time `get()` is unblocked
+    assertThat(interceptor.capturedStatusCode).isEqualTo(200);
     assertThat(interceptor.capturedResponseHeaders).isNotNull();
     assertThat(interceptor.capturedMessage).isEqualTo(request);
-
-    // Attempt to busy wait {NUM_RETRIES} times until the interceptor's `onClose()` is called
-    int attempts = 0;
-    while (interceptor.capturedStatusCode == 0) {
-      Thread.sleep(BUSY_WAIT_TIME_IN_MS);
-      attempts++;
-      if (attempts == NUM_RETRIES) {
-        break;
-      }
-    }
-
-    assertThat(interceptor.capturedStatusCode).isEqualTo(200);
   }
 }

gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonDirectCallableTest.java

@@ -139,7 +139,7 @@ public void testSuccessfulUnaryResponse() throws ExecutionException, Interrupted
 
     Field request;
     Field expectedResponse;
-    request = expectedResponse = createTestMessage();
+    request = expectedResponse = createTestMessage(2);
 
     MOCK_SERVICE.addResponse(expectedResponse);
 
@@ -151,6 +151,78 @@ public void testSuccessfulUnaryResponse() throws ExecutionException, Interrupted
     assertThat(headerValue).isEqualTo("headerValue");
   }
 
+  /**
+   * This test is for a Unary Call where server mistakenly sends multiple responses back The
+   * expectation for this MOCK_SERVICE is to return what was sent into the request i.e.
+   * callable.futureCall(x) -> x
+   *
+   * <p>For a Unary Call, gax will return only the first (and hopefully only) response back.
+   *
+   * @throws InterruptedException
+   * @throws ExecutionException
+   */
+  @Test
+  public void testSuccessfulMultipleResponsesForUnaryCall()
+      throws InterruptedException, ExecutionException {
+    HttpJsonDirectCallable<Field, Field> callable =
+        new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
+
+    HttpJsonCallContext callContext = HttpJsonCallContext.createDefault().withChannel(channel);
+
+    Field request = createTestMessage(2);
+    Field expectedResponse = createTestMessage(2);
+    Field otherResponse = createTestMessage(10);
+    MOCK_SERVICE.addResponse(expectedResponse);
+    MOCK_SERVICE.addResponse(otherResponse);
+    MOCK_SERVICE.addResponse(otherResponse);
+
+    Field actualResponse = callable.futureCall(request, callContext).get();
+    assertThat(actualResponse).isEqualTo(expectedResponse);
+    assertThat(MOCK_SERVICE.getRequestPaths().size()).isEqualTo(1);
+    String headerValue = MOCK_SERVICE.getRequestHeaders().get("header-key").iterator().next();
+    assertThat(headerValue).isEqualTo("headerValue");
+  }
+
+  /**
+   * This test is for a Unary Call where server mistakenly sends multiple responses back The
+   * expectation for this MOCK_SERVICE is to return what was sent into the request i.e.
+   * callable.futureCall(x) -> x
+   *
+   * <p>For a Unary Call, gax will return only the first (and hopefully only) response back.
+   *
+   * @throws InterruptedException
+   * @throws ExecutionException
+   */
+  @Test
+  public void testErrorMultipleResponsesForUnaryCall()
+      throws InterruptedException, ExecutionException {
+    HttpJsonDirectCallable<Field, Field> callable =
+        new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
+
+    HttpJsonCallContext callContext = HttpJsonCallContext.createDefault().withChannel(channel);
+
+    Field request = createTestMessage(2);
+    Field expectedResponse = createTestMessage(2);
+    Field randomResponse1 = createTestMessage(10);
+    Field randomResponse2 = createTestMessage(3);
+    MOCK_SERVICE.addResponse(randomResponse1);
+    MOCK_SERVICE.addResponse(expectedResponse);
+    MOCK_SERVICE.addResponse(randomResponse2);
+
+    Field actualResponse = callable.futureCall(request, callContext).get();
+    // Gax returns the first response for Unary Call
+    assertThat(actualResponse).isEqualTo(randomResponse1);
+    assertThat(actualResponse).isNotEqualTo(expectedResponse);
+    assertThat(MOCK_SERVICE.getRequestPaths().size()).isEqualTo(1);
+    String headerValue = MOCK_SERVICE.getRequestHeaders().get("header-key").iterator().next();
+    assertThat(headerValue).isEqualTo("headerValue");
+  }
+
+  /**
+   * The expectation for gax is that an exception from the server will return an exception response
+   *
+   * @throws InterruptedException
+   */
   @Test
   public void testErrorUnaryResponse() throws InterruptedException {
     HttpJsonDirectCallable<Field, Field> callable =
@@ -164,7 +236,7 @@ public void testErrorUnaryResponse() throws InterruptedException {
     MOCK_SERVICE.addException(exception);
 
     try {
-      callable.futureCall(createTestMessage(), callContext).get();
+      callable.futureCall(createTestMessage(2), callContext).get();
       Assert.fail("No exception raised");
     } catch (ExecutionException e) {
       HttpResponseException respExp = (HttpResponseException) e.getCause();
@@ -173,6 +245,13 @@ public void testErrorUnaryResponse() throws InterruptedException {
     }
   }
 
+  /**
+   * This test is for a Unary Call where server sends back a null value but successful status code
+   * Gax expects the response back to be parse-able into JSON but the null value is not valid. This
+   * will throw an Exception for a successful response but invalid content
+   *
+   * @throws InterruptedException
+   */
   @Test
   public void testErrorNullContentSuccessfulResponse() throws InterruptedException {
     HttpJsonDirectCallable<Field, Field> callable =
@@ -183,7 +262,7 @@ public void testErrorNullContentSuccessfulResponse() throws InterruptedException
     MOCK_SERVICE.addNullResponse();
 
     try {
-      callable.futureCall(createTestMessage(), callContext).get();
+      callable.futureCall(createTestMessage(2), callContext).get();
       Assert.fail("No exception raised");
     } catch (ExecutionException e) {
       HttpJsonStatusRuntimeException respExp = (HttpJsonStatusRuntimeException) e.getCause();
@@ -193,6 +272,12 @@ public void testErrorNullContentSuccessfulResponse() throws InterruptedException
     }
   }
 
+  /**
+   * The expectation for a non-2xx from the server is an exception response regardless of the
+   * content sent back
+   *
+   * @throws InterruptedException
+   */
   @Test
   public void testErrorNullContentFailedResponse() throws InterruptedException {
     HttpJsonDirectCallable<Field, Field> callable =
@@ -202,7 +287,7 @@ public void testErrorNullContentFailedResponse() throws InterruptedException {
     MOCK_SERVICE.addNullResponse(400);
 
     try {
-      callable.futureCall(createTestMessage(), callContext).get();
+      callable.futureCall(createTestMessage(2), callContext).get();
       Assert.fail("No exception raised");
     } catch (ExecutionException e) {
       HttpResponseException respExp = (HttpResponseException) e.getCause();
@@ -211,10 +296,37 @@ public void testErrorNullContentFailedResponse() throws InterruptedException {
     }
   }
 
-  private Field createTestMessage() {
+  /**
+   * Expectation is that an Exception is returned even on a non-2xx status code
+   *
+   * @throws InterruptedException
+   */
+  @Test
+  public void testErrorNon2xxOr4xxResponse() throws InterruptedException {
+    HttpJsonDirectCallable<Field, Field> callable =
+        new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
+
+    HttpJsonCallContext callContext = HttpJsonCallContext.createDefault().withChannel(channel);
+
+    ApiException exception =
+        ApiExceptionFactory.createException(
+            new Exception(), FakeStatusCode.of(Code.INTERNAL), false);
+    MOCK_SERVICE.addException(500, exception);
+
+    try {
+      callable.futureCall(createTestMessage(2), callContext).get();
+      Assert.fail("No exception raised");
+    } catch (ExecutionException e) {
+      HttpResponseException respExp = (HttpResponseException) e.getCause();
+      assertThat(respExp.getStatusCode()).isEqualTo(500);
+      assertThat(respExp.getContent()).isEqualTo(exception.toString());
+    }
+  }
+
+  private Field createTestMessage(int number) {
     return Field.newBuilder() // "echo" service
         .setName("john/imTheBestField")
-        .setNumber(2)
+        .setNumber(number)
         .setCardinality(Cardinality.CARDINALITY_OPTIONAL)
         .setDefaultValue("blah")
         .build();