Merge branch 'main' into compression

# Conflicts:
#	vertx-grpc-server/src/main/java/io/vertx/grpc/server/impl/GrpcServerImpl.java
#	vertx-grpc-server/src/main/java/io/vertx/grpc/server/impl/GrpcServerResponseImpl.java

Author: zZHorizonZz

README.md

@@ -8,4 +8,4 @@ to write gRPC middlewares like reverse proxies.
 
 # License
 
-Eclipse Public License - Version 1.0 / Apache License - Version 2.0
+Eclipse Public License - Version 2.0 / Apache License - Version 2.0

vertx-grpc-client/src/main/java/io/vertx/grpc/client/GrpcClientResponse.java

@@ -31,6 +31,11 @@
 @VertxGen
 public interface GrpcClientResponse<Req, Resp> extends GrpcReadStream<Resp> {
 
+  /**
+   * @return the associated client request
+   */
+  GrpcClientRequest<Req, Resp> request();
+
   /**
    * @return the gRPC status or {@code null} when the status has not yet been received
    */

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

@@ -46,7 +46,6 @@ public class GrpcClientRequestImpl<Req, Resp> extends GrpcWriteStreamBase<GrpcCl
   private TimeUnit timeoutUnit;
   private String timeoutHeader;
   private Timer deadline;
-  private boolean cancelled;
 
   public GrpcClientRequestImpl(HttpClientRequest httpRequest,
                                long maxMessageSize,
@@ -228,31 +227,11 @@ void cancelTimeout() {
   }
 
   @Override
-  public void cancel() {
-    if (cancelled) {
-      return;
-    }
-    cancelled = true;
-    context.execute(() -> {
-      boolean responseEnded;
-      if (response.failed()) {
-        return;
-      } else if (response.succeeded()) {
-        GrpcClientResponse<Req, Resp> resp = response.result();
-        if (resp.end().failed()) {
-          return;
-        } else {
-          responseEnded = resp.end().succeeded();
-        }
-      } else {
-        responseEnded = false;
-      }
-      if (!isTrailersSent() || !responseEnded) {
-        httpRequest
-          .reset(GrpcError.CANCELLED.http2ResetCode)
-          .onSuccess(v -> handleError(GrpcError.CANCELLED));
-      }
-    });
+  protected boolean sendCancel() {
+    httpRequest
+      .reset(GrpcError.CANCELLED.http2ResetCode)
+      .onSuccess(v -> handleError(GrpcError.CANCELLED));
+    return true;
   }
 
   @Override

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

@@ -18,6 +18,7 @@
 import io.vertx.core.http.HttpClientResponse;
 
 import io.vertx.core.internal.ContextInternal;
+import io.vertx.grpc.client.GrpcClientRequest;
 import io.vertx.grpc.client.GrpcClientResponse;
 import io.vertx.grpc.client.InvalidStatusException;
 import io.vertx.grpc.common.*;
@@ -58,6 +59,11 @@ public GrpcClientResponseImpl(ContextInternal context,
     this.status = status;
   }
 
+  @Override
+  public GrpcClientRequest<Req, Resp> request() {
+    return request;
+  }
+
   @Override
   public MultiMap headers() {
     return httpResponse.headers();
@@ -79,6 +85,7 @@ protected void handleEnd() {
       }
     }
     super.handleEnd();
+    request.handleStatus(status);
     if (!request.isTrailersSent()) {
       request.cancel();
     }

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

@@ -293,15 +293,19 @@ public void testClientStreamingCompletedBeforeHalfClose(TestContext should) thro
 
     super.testClientStreamingCompletedBeforeHalfClose(should);
 
-    Async done = should.async();
+    Async done = should.async(2);
     client = GrpcClient.client(vertx);
     client.request(SocketAddress.inetSocketAddress(port, "localhost"), SINK)
       .onComplete(should.asyncAssertSuccess(callRequest -> {
+        callRequest.exceptionHandler(err -> {
+          should.assertTrue(callRequest.isCancelled());
+          done.countDown();
+        });
         callRequest.response().onComplete(should.asyncAssertFailure(failure -> {
           should.assertEquals(GrpcErrorException.class, failure.getClass());
           GrpcErrorException f = (GrpcErrorException) failure;
           should.assertEquals(GrpcStatus.CANCELLED, f.status());
-          done.complete();
+          done.countDown();
         }));
         callRequest.write(Request.newBuilder().setName("the-value").build());
       }));
@@ -469,12 +473,17 @@ public void onCompleted() {
     client.request(SocketAddress.inetSocketAddress(port, "localhost"), SINK)
       .onComplete(should.asyncAssertSuccess(callRequest -> {
         callRequest.write(Request.getDefaultInstance());
-        cf.whenComplete((v, t) -> {
-          callRequest.cancel();
-          try {
-            callRequest.write(Request.getDefaultInstance());
-          } catch (IllegalStateException ignore) {
-          }
+        io.vertx.core.Context ctx = vertx.getOrCreateContext();
+        cf.whenComplete((v1, t) -> {
+          ctx.runOnContext(v2 -> {
+            should.assertFalse(callRequest.isCancelled());
+            callRequest.cancel();
+            should.assertTrue(callRequest.isCancelled());
+            try {
+              callRequest.write(Request.getDefaultInstance());
+            } catch (IllegalStateException ignore) {
+            }
+          });
         });
       }));
   }

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

@@ -68,4 +68,9 @@ public interface GrpcWriteStream<T> extends WriteStream<T> {
    */
   void cancel();
 
+  /**
+   * @return whether the stream is cancelled
+   */
+  boolean isCancelled();
+
 }

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

@@ -30,6 +30,7 @@ public abstract class GrpcWriteStreamBase<S extends GrpcWriteStreamBase<S, T>, T
   private boolean headersSent;
   private boolean trailersSent;
   private GrpcError error;
+  private boolean cancelled;
 
   private MultiMap headers;
   private MultiMap trailers;
@@ -52,9 +53,8 @@ public void init() {
         StreamResetException reset = (StreamResetException) err;
         GrpcError error = mapHttp2ErrorCode(reset.getCode());
         handleError(error);
-      } else {
-        handleException(err);
       }
+      handleException(err);
     });
   }
 
@@ -65,6 +65,7 @@ public S errorHandler(Handler<GrpcError> handler) {
 
   public void handleError(GrpcError error) {
     if (this.error == null) {
+      cancelled |= error == GrpcError.CANCELLED;
       this.error = error;
       Handler<GrpcError> handler = errorHandler;
       if (handler != null) {
@@ -80,6 +81,22 @@ private void handleException(Throwable err) {
     }
   }
 
+  public void handleStatus(GrpcStatus status) {
+    cancelled |= status == GrpcStatus.CANCELLED;
+  }
+
+  @Override
+  public boolean isCancelled() {
+    return cancelled;
+  }
+
+  @Override
+  public void cancel() {
+    if (!cancelled) {
+      cancelled = sendCancel();
+    }
+  }
+
   @Override
   public final S encoding(String encoding) {
     if (headersSent) {
@@ -110,10 +127,6 @@ public boolean isTrailersSent() {
     return trailersSent;
   }
 
-  public boolean isCancelled() {
-    return error == GrpcError.CANCELLED;
-  }
-
   @Override
   public final MultiMap headers() {
     if (headersSent) {
@@ -192,6 +205,8 @@ public final Future<Void> end() {
   protected abstract Future<Void> sendMessage(Buffer message, boolean compressed);
   protected abstract Future<Void> sendEnd();
   protected abstract Future<Void> sendHead();
+  protected abstract boolean sendCancel();
+
 
   protected String contentType(WireFormat wireFormat) {
     if (wireFormat != null) {

vertx-grpc-protoc-plugin2/pom.xml

@@ -30,7 +30,6 @@
   <name>Vert.x gRPC Protoc Plugin</name>
 
   <properties>
-    <canteen.version>1.1.0</canteen.version>
     <protoc.version>3.21.12</protoc.version>
   </properties>
 
@@ -88,18 +87,6 @@
           </archive>
         </configuration>
       </plugin>
-      <plugin>
-        <groupId>com.salesforce.servicelibs</groupId>
-        <artifactId>canteen-maven-plugin</artifactId>
-        <version>${canteen.version}</version>
-        <executions>
-          <execution>
-            <goals>
-              <goal>bootstrap</goal>
-            </goals>
-          </execution>
-        </executions>
-      </plugin>
     </plugins>
   </build>
 </project>

vertx-grpc-server/src/main/generated/io/vertx/grpc/server/GrpcServerOptionsConverter.java

@@ -2,8 +2,6 @@
 
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.json.JsonArray;
-import java.time.Instant;
-import java.time.format.DateTimeFormatter;
 
 /**
  * Converter and mapper for {@link io.vertx.grpc.server.GrpcServerOptions}.

vertx-grpc-server/src/main/java/io/vertx/grpc/server/Service.java

@@ -2,6 +2,7 @@
 
 import com.google.protobuf.Descriptors;
 import io.vertx.codegen.annotations.GenIgnore;
+import io.vertx.core.Future;
 import io.vertx.grpc.common.ServiceName;
 import io.vertx.grpc.server.impl.ServiceBuilderImpl;
 
@@ -50,6 +51,13 @@ static ServiceBuilder service(ServiceName serviceName, Descriptors.ServiceDescri
    */
   void bind(GrpcServer server);
 
+  /**
+   * Close the service.
+   */
+  default Future<Void> close() {
+    return Future.succeededFuture();
+  }
+
   /**
    * Get a list of all method descriptors for this service.
    *