feat: introduce HttpJsonClientCall, Listeners infrastructure and ServerStreaming support in REST transport

gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcDirectServerStreamingCallableTest.java

@@ -145,7 +145,7 @@ public void testServerStreaming() throws Exception {
 
     streamingCallable.call(DEFAULT_REQUEST, moneyObserver);
 
-    latch.await(20, TimeUnit.SECONDS);
+    Truth.assertThat(latch.await(20, TimeUnit.SECONDS)).isTrue();
     Truth.assertThat(moneyObserver.error).isNull();
     Truth.assertThat(moneyObserver.response).isEqualTo(DEFAULT_RESPONSE);
   }
@@ -157,13 +157,13 @@ public void testManualFlowControl() throws Exception {
 
     streamingCallable.call(DEFAULT_REQUEST, moneyObserver);
 
-    latch.await(500, TimeUnit.MILLISECONDS);
+    Truth.assertThat(latch.await(500, TimeUnit.MILLISECONDS)).isFalse();
     Truth.assertWithMessage("Received response before requesting it")
         .that(moneyObserver.response)
         .isNull();
 
     moneyObserver.controller.request(1);
-    latch.await(500, TimeUnit.MILLISECONDS);
+    Truth.assertThat(latch.await(500, TimeUnit.MILLISECONDS)).isTrue();
 
     Truth.assertThat(moneyObserver.response).isEqualTo(DEFAULT_RESPONSE);
     Truth.assertThat(moneyObserver.completed).isTrue();
@@ -178,7 +178,7 @@ public void testCancelClientCall() throws Exception {
 
     moneyObserver.controller.cancel();
     moneyObserver.controller.request(1);
-    latch.await(500, TimeUnit.MILLISECONDS);
+    Truth.assertThat(latch.await(500, TimeUnit.MILLISECONDS)).isTrue();
 
     Truth.assertThat(moneyObserver.error).isInstanceOf(CancellationException.class);
     Truth.assertThat(moneyObserver.error).hasMessageThat().isEqualTo("User cancelled stream");
@@ -190,7 +190,7 @@ public void testOnResponseError() throws Throwable {
     MoneyObserver moneyObserver = new MoneyObserver(true, latch);
 
     streamingCallable.call(ERROR_REQUEST, moneyObserver);
-    latch.await(500, TimeUnit.MILLISECONDS);
+    Truth.assertThat(latch.await(500, TimeUnit.MILLISECONDS)).isTrue();
 
     Truth.assertThat(moneyObserver.error).isInstanceOf(ApiException.class);
     Truth.assertThat(((ApiException) moneyObserver.error).getStatusCode().getCode())

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

@@ -40,6 +40,7 @@
 import com.google.protobuf.TypeRegistry;
 import java.io.InputStream;
 import java.io.InputStreamReader;
+import java.io.Reader;
 import java.lang.reflect.Type;
 import java.nio.charset.StandardCharsets;
 
@@ -91,25 +92,28 @@ ApiMessageHttpResponseParser.Builder<ResponseT> newBuilder() {
 
   @Override
   public ResponseT parse(InputStream httpResponseBody) {
+    return parse(httpResponseBody, null);
+  }
+
+  @Override
+  public ResponseT parse(InputStream httpResponseBody, TypeRegistry registry) {
+    return parse(new InputStreamReader(httpResponseBody, StandardCharsets.UTF_8), registry);
+  }
+
+  @Override
+  public ResponseT parse(Reader httpResponseBody, TypeRegistry registry) {
     if (getResponseInstance() == null) {
       return null;
     } else {
       Type responseType = getResponseInstance().getClass();
       try {
-        return getResponseMarshaller()
-            .fromJson(
-                new InputStreamReader(httpResponseBody, StandardCharsets.UTF_8), responseType);
+        return getResponseMarshaller().fromJson(httpResponseBody, responseType);
       } catch (JsonIOException | JsonSyntaxException e) {
         throw new RestSerializationException(e);
       }
     }
   }
 
-  @Override
-  public ResponseT parse(InputStream httpResponseBody, TypeRegistry registry) {
-    return parse(httpResponseBody);
-  }
-
   @Override
   public String serialize(ResponseT response) {
     return getResponseMarshaller().toJson(response);

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

@@ -37,12 +37,18 @@
 @AutoValue
 /* Method descriptor for messages to be transmitted over HTTP. */
 public abstract class ApiMethodDescriptor<RequestT, ResponseT> {
+  public enum MethodType {
+    UNARY,
+    CLIENT_STREAMING,
+    SERVER_STREAMING,
+    BIDI_STREAMING,
+    UNKNOWN;
+  }
 
   public abstract String getFullMethodName();
 
   public abstract HttpRequestFormatter<RequestT> getRequestFormatter();
 
-  @Nullable
   public abstract HttpResponseParser<ResponseT> getResponseParser();
 
   /** Return the HTTP method for this request message type. */
@@ -55,8 +61,11 @@ public abstract class ApiMethodDescriptor<RequestT, ResponseT> {
   @Nullable
   public abstract PollingRequestFactory<RequestT> getPollingRequestFactory();
 
+  public abstract MethodType getType();
+
   public static <RequestT, ResponseT> Builder<RequestT, ResponseT> newBuilder() {
-    return new AutoValue_ApiMethodDescriptor.Builder<RequestT, ResponseT>();
+    return new AutoValue_ApiMethodDescriptor.Builder<RequestT, ResponseT>()
+        .setType(MethodType.UNARY);
   }
 
   @AutoValue.Builder
@@ -78,6 +87,8 @@ public abstract Builder<RequestT, ResponseT> setOperationSnapshotFactory(
     public abstract Builder<RequestT, ResponseT> setPollingRequestFactory(
         PollingRequestFactory<RequestT> pollingRequestFactory);
 
+    public abstract Builder<RequestT, ResponseT> setType(MethodType type);
+
     public abstract ApiMethodDescriptor<RequestT, ResponseT> build();
   }
 }

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

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google LLC nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.google.api.gax.httpjson;
+
+import com.google.api.client.http.HttpResponseException;
+import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.ApiExceptionFactory;
+import com.google.api.gax.rpc.StatusCode;
+import com.google.api.gax.rpc.StatusCode.Code;
+import com.google.common.collect.ImmutableSet;
+import java.util.Set;
+import java.util.concurrent.CancellationException;
+
+class HttpJsonApiExceptionFactory {
+  private final Set<Code> retryableCodes;
+
+  HttpJsonApiExceptionFactory(Set<Code> retryableCodes) {
+    this.retryableCodes = ImmutableSet.copyOf(retryableCodes);
+  }
+
+  ApiException create(Throwable throwable) {
+    if (throwable instanceof HttpResponseException) {
+      HttpResponseException e = (HttpResponseException) throwable;
+      StatusCode statusCode = HttpJsonStatusCode.of(e.getStatusCode());
+      boolean canRetry = retryableCodes.contains(statusCode.getCode());
+      String message = e.getStatusMessage();
+      return createApiException(throwable, statusCode, message, canRetry);
+    } else if (throwable instanceof StatusRuntimeException) {
+      StatusRuntimeException e = (StatusRuntimeException) throwable;
+      StatusCode statusCode = HttpJsonStatusCode.of(e.getStatusCode());
+      return createApiException(
+          throwable,
+          HttpJsonStatusCode.of(e.getStatusCode()),
+          e.getMessage(),
+          retryableCodes.contains(statusCode.getCode()));
+    } else if (throwable instanceof CancellationException) {
+      return ApiExceptionFactory.createException(
+          throwable, HttpJsonStatusCode.of(Code.CANCELLED), false);
+    } else if (throwable instanceof ApiException) {
+      return (ApiException) throwable;
+    } else {
+      // Do not retry on unknown throwable, even when UNKNOWN is in retryableCodes
+      return ApiExceptionFactory.createException(
+          throwable, HttpJsonStatusCode.of(StatusCode.Code.UNKNOWN), false);
+    }
+  }
+
+  private ApiException createApiException(
+      Throwable throwable, StatusCode statusCode, String message, boolean canRetry) {
+    return message == null
+        ? ApiExceptionFactory.createException(throwable, statusCode, canRetry)
+        : ApiExceptionFactory.createException(message, throwable, statusCode, canRetry);
+  }
+}