Merge pull request #106 from eclipse-vertx/bunch-of-improvements

Improvements

vertx-grpc-client/src/main/java/io/vertx/grpc/client/impl/GrpcClientImpl.java

@@ -27,7 +27,7 @@
 import io.vertx.grpc.common.ServiceMethod;
 import io.vertx.grpc.common.GrpcMessageDecoder;
 import io.vertx.grpc.common.GrpcMessageEncoder;
-import io.vertx.grpc.common.impl.GrpcRequestLocal;
+import io.vertx.grpc.common.GrpcLocal;
 
 import java.util.concurrent.TimeUnit;
 
@@ -85,11 +85,11 @@ public Future<GrpcClientRequest<Buffer, Buffer>> request(Address server) {
 
   private void configureTimeout(GrpcClientRequest<?, ?> request) {
     ContextInternal current = (ContextInternal) vertx.getOrCreateContext();
-    GrpcRequestLocal local = current.getLocal(GrpcRequestLocal.CONTEXT_LOCAL_KEY);
+    GrpcLocal local = current.getLocal(GrpcLocal.CONTEXT_LOCAL_KEY);
     long timeout = this.timeout;
     TimeUnit timeoutUnit = this.timeoutUnit;
     if (local != null) {
-      timeout = local.deadline - System.currentTimeMillis();
+      timeout = local.deadline().toEpochMilli() - System.currentTimeMillis();
       timeoutUnit = TimeUnit.MILLISECONDS;
       if (timeout < 0L) {
         throw new UnsupportedOperationException("Handle this case");

vertx-grpc-client/src/main/java/io/vertx/grpc/client/impl/GrpcClientRequestImpl.java

@@ -23,16 +23,13 @@
 import java.util.concurrent.TimeUnit;
 
 import io.vertx.core.http.HttpConnection;
+import io.vertx.core.http.StreamResetException;
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.PromiseInternal;
 import io.vertx.grpc.client.GrpcClientRequest;
 import io.vertx.grpc.client.GrpcClientResponse;
-import io.vertx.grpc.common.CodecException;
-import io.vertx.grpc.common.GrpcError;
-import io.vertx.grpc.common.GrpcMessage;
-import io.vertx.grpc.common.GrpcMessageDecoder;
-import io.vertx.grpc.common.GrpcMessageEncoder;
-import io.vertx.grpc.common.ServiceName;
+import io.vertx.grpc.common.GrpcErrorException;
+import io.vertx.grpc.common.*;
 import io.vertx.grpc.common.impl.GrpcMessageImpl;
 
 /**
@@ -69,10 +66,15 @@ public GrpcClientRequestImpl(HttpClientRequest httpRequest,
     this.timeoutHeader = null;
     this.response = httpRequest
       .response()
-      .map(httpResponse -> {
+      .compose(httpResponse -> {
         GrpcClientResponseImpl<Req, Resp> grpcResponse = new GrpcClientResponseImpl<>(context, this, httpResponse, messageDecoder);
         grpcResponse.init();
-        return grpcResponse;
+        return Future.succeededFuture(grpcResponse);
+      }, err -> {
+        if (err instanceof StreamResetException) {
+          err = GrpcErrorException.create((StreamResetException) err);
+        }
+        return Future.failedFuture(err);
       });
   }
 

vertx-grpc-client/src/test/java/io/vertx/tests/client/ClientRequestTest.java

@@ -28,8 +28,9 @@
 import io.vertx.ext.unit.TestContext;
 import io.vertx.grpc.client.GrpcClient;
 import io.vertx.grpc.client.GrpcClientOptions;
+import io.vertx.grpc.common.GrpcErrorException;
 import io.vertx.grpc.common.GrpcStatus;
-import io.vertx.grpc.common.impl.GrpcRequestLocal;
+import io.vertx.grpc.common.GrpcLocal;
 import org.junit.Test;
 
 import java.io.File;
@@ -287,9 +288,9 @@ public void testClientStreamingCompletedBeforeHalfClose(TestContext should) thro
     client.request(SocketAddress.inetSocketAddress(port, "localhost"), STREAMING_SINK)
       .onComplete(should.asyncAssertSuccess(callRequest -> {
         callRequest.response().onComplete(should.asyncAssertFailure(failure -> {
-          should.assertEquals(StreamResetException.class, failure.getClass());
-          StreamResetException reset = (StreamResetException) failure;
-          should.assertEquals(8L, reset.getCode());
+          should.assertEquals(GrpcErrorException.class, failure.getClass());
+          GrpcErrorException f = (GrpcErrorException) failure;
+          should.assertEquals(GrpcStatus.CANCELLED, f.status());
           done.complete();
         }));
         callRequest.write(Item.newBuilder().setValue("the-value").build());
@@ -546,9 +547,9 @@ public void testTimeoutOnClient(TestContext should) throws Exception {
           .timeout(1, TimeUnit.SECONDS);
         callRequest.write(Item.getDefaultInstance());
         callRequest.response().onComplete(should.asyncAssertFailure(err -> {
-          should.assertTrue(err instanceof StreamResetException);
-          StreamResetException sre = (StreamResetException) err;
-          should.assertEquals(8L, sre.getCode());
+          should.assertTrue(err instanceof GrpcErrorException);
+          GrpcErrorException failure = (GrpcErrorException) err;
+          should.assertEquals(GrpcStatus.CANCELLED, failure.status());
         }));
       }));
   }
@@ -559,14 +560,14 @@ public void testTimeoutOnClientPropagation(TestContext should) throws Exception
     client = GrpcClient.client(vertx, new GrpcClientOptions().setScheduleDeadlineAutomatically(true));
     ContextInternal context = (ContextInternal) vertx.getOrCreateContext();
     context.runOnContext(v -> {
-      context.putLocal(GrpcRequestLocal.CONTEXT_LOCAL_KEY, AccessMode.CONCURRENT, new GrpcRequestLocal(System.currentTimeMillis() + 1000));
+      context.putLocal(GrpcLocal.CONTEXT_LOCAL_KEY, AccessMode.CONCURRENT, new GrpcLocal(System.currentTimeMillis() + 1000));
       client.request(SocketAddress.inetSocketAddress(port, "localhost"), STREAMING_SINK)
         .onComplete(should.asyncAssertSuccess(callRequest -> {
           callRequest.write(Item.getDefaultInstance());
           callRequest.response().onComplete(should.asyncAssertFailure(err -> {
-            should.assertTrue(err instanceof StreamResetException);
-            StreamResetException sre = (StreamResetException) err;
-            should.assertEquals(8L, sre.getCode());
+            should.assertTrue(err instanceof GrpcErrorException);
+            GrpcErrorException failure = (GrpcErrorException) err;
+            should.assertEquals(GrpcStatus.CANCELLED, failure.status());
           }));
         }));
     });

vertx-grpc-client/src/test/java/io/vertx/tests/client/ClientTest.java

@@ -219,7 +219,7 @@ public void testClientStreamingCompletedBeforeHalfClose(TestContext should) thro
     startServer(new StreamingGrpc.StreamingImplBase() {
       @Override
       public StreamObserver<Item> sink(StreamObserver<Empty> responseObserver) {
-        return new StreamObserver<Item>() {
+        return new StreamObserver<>() {
           @Override
           public void onNext(Item item) {
             responseObserver.onCompleted();

vertx-grpc-common/src/main/java/io/vertx/grpc/common/GrpcErrorException.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright (c) 2011-2022 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+package io.vertx.grpc.common;
+
+import io.vertx.core.VertxException;
+import io.vertx.core.http.StreamResetException;
+
+/**
+ * Thrown when a failure happens before the response, and it could be interpreted to a gRPC failure, e.g.
+ * in practice it means an HTTP/2 stream reset mapped to a gRPC code according to the
+ * <a href="https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#http2-transport-mapping">spec</a>.
+ */
+public class GrpcErrorException extends VertxException {
+
+  public static GrpcErrorException create(StreamResetException sre) {
+    GrpcError error = GrpcError.mapHttp2ErrorCode(sre.getCode());
+    GrpcStatus status = GrpcStatus.UNKNOWN;
+    if (error != null) {
+      status = error.status;
+    }
+    return new GrpcErrorException(status);
+  }
+
+  private final GrpcStatus status;
+
+  public GrpcErrorException(GrpcStatus status) {
+    super("gRPC error status: " + status.name());
+
+    this.status = status;
+  }
+
+  public GrpcStatus status() {
+    return status;
+  }
+}

vertx-grpc-common/src/main/java/io/vertx/grpc/common/GrpcLocal.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2011-2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+package io.vertx.grpc.common;
+
+import io.vertx.core.Context;
+import io.vertx.core.Vertx;
+import io.vertx.core.internal.ContextInternal;
+import io.vertx.core.spi.context.storage.ContextLocal;
+import io.vertx.grpc.common.impl.GrpcRequestLocalRegistration;
+
+import java.time.Instant;
+
+/**
+ * gRPC request local propagation.
+ */
+public class GrpcLocal {
+
+  /**
+   * Context local key.
+   */
+  public static final ContextLocal<GrpcLocal> CONTEXT_LOCAL_KEY = GrpcRequestLocalRegistration.CONTEXT_LOCAL;
+
+  private final long deadlineMillis;
+  private Instant deadline;
+
+  public GrpcLocal(long deadlineMillis) {
+    this.deadlineMillis = deadlineMillis;
+  }
+
+  /**
+   * @return the local associated with the given {@code context}
+   */
+  public static GrpcLocal of(Context context) {
+    return ((ContextInternal)context).getLocal(CONTEXT_LOCAL_KEY);
+  }
+
+  /**
+   * @return the current request local or {@code null}
+   */
+  public static GrpcLocal current() {
+    Context ctx = Vertx.currentContext();
+    return ctx != null ? of(ctx) : null;
+  }
+
+  /**
+   * @return the deadline
+   */
+  public Instant deadline() {
+    if (deadline == null) {
+      deadline = Instant.ofEpochMilli(deadlineMillis);
+    }
+    return deadline;
+  }
+}

vertx-grpc-common/src/main/java/io/vertx/grpc/common/impl/GrpcReadStreamBase.java

@@ -175,7 +175,7 @@ public S endHandler(Handler<Void> endHandler) {
     return (S) this;
   }
 
-  protected void handleReset(long code) {
+  public void handleReset(long code) {
     Handler<GrpcError> handler = errorHandler;
     if (handler != null) {
       GrpcError error = mapHttp2ErrorCode(code);

vertx-grpc-common/src/main/java/io/vertx/grpc/common/impl/GrpcRequestLocalRegistration.java

@@ -13,13 +13,14 @@
 import io.vertx.core.internal.VertxBootstrap;
 import io.vertx.core.spi.VertxServiceProvider;
 import io.vertx.core.spi.context.storage.ContextLocal;
+import io.vertx.grpc.common.GrpcLocal;
 
 /**
- * Registration of context local for {@link GrpcRequestLocal}.
+ * Registration of context local for {@link GrpcLocal}.
  */
 public class GrpcRequestLocalRegistration implements VertxServiceProvider {
 
-  static final ContextLocal<GrpcRequestLocal> CONTEXT_LOCAL = ContextLocal.registerLocal(GrpcRequestLocal.class);
+  public static final ContextLocal<GrpcLocal> CONTEXT_LOCAL = ContextLocal.registerLocal(GrpcLocal.class);
 
   @Override
   public void init(VertxBootstrap builder) {

vertx-grpc-it/pom.xml

@@ -36,12 +36,12 @@
   <dependencies>
     <dependency>
       <groupId>io.vertx</groupId>
-      <artifactId>vertx-grpcio-client</artifactId>
+      <artifactId>vertx-grpc-client</artifactId>
       <version>${project.version}</version>
     </dependency>
     <dependency>
       <groupId>io.vertx</groupId>
-      <artifactId>vertx-grpcio-server</artifactId>
+      <artifactId>vertx-grpc-server</artifactId>
       <version>${project.version}</version>
     </dependency>
 
@@ -68,16 +68,6 @@
       <artifactId>bcpkix-jdk15on</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>io.grpc</groupId>
-      <artifactId>grpc-stub</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>io.grpc</groupId>
-      <artifactId>grpc-netty-shaded</artifactId>
-      <scope>test</scope>
-    </dependency>
   </dependencies>
 
   <build>
@@ -113,7 +103,6 @@
             <id>test-compile</id>
             <goals>
               <goal>test-compile</goal>
-              <goal>test-compile-custom</goal>
             </goals>
           </execution>
         </executions>