Add experimental gRPC streaming over the event bus

Signed-off-by: Daniel Fiala <danfiala23@gmail.com>

Author: zZHorizonZz

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

@@ -0,0 +1,48 @@
+package io.vertx.grpc.common;
+
+/**
+ * The cardinality of a gRPC service method.
+ */
+public enum MethodType {
+
+  /**
+   * A single request and a single response.
+   */
+  UNARY,
+
+  /**
+   * A stream of requests and a single response.
+   */
+  CLIENT_STREAMING,
+
+  /**
+   * A single request and a stream of responses.
+   */
+  SERVER_STREAMING,
+
+  /**
+   * A stream of requests and a stream of responses.
+   */
+  BIDI;
+
+  /**
+   * @return whether the client sends a stream of requests
+   */
+  public boolean clientStreaming() {
+    return this == CLIENT_STREAMING || this == BIDI;
+  }
+
+  /**
+   * @return whether the server sends a stream of responses
+   */
+  public boolean serverStreaming() {
+    return this == SERVER_STREAMING || this == BIDI;
+  }
+
+  /**
+   * @return whether either side streams, i.e. anything other than {@link #UNARY}
+   */
+  public boolean streaming() {
+    return this != UNARY;
+  }
+}

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

@@ -19,6 +19,10 @@
 public interface ServiceMethod<I, O> {
 
   static <Req, Resp> ServiceMethod<Resp, Req> client(ServiceName serviceName, String methodName, GrpcMessageEncoder<Req> encoder, GrpcMessageDecoder<Resp> decoder) {
+    return client(serviceName, methodName, MethodType.UNARY, encoder, decoder);
+  }
+
+  static <Req, Resp> ServiceMethod<Resp, Req> client(ServiceName serviceName, String methodName, MethodType type, GrpcMessageEncoder<Req> encoder, GrpcMessageDecoder<Resp> decoder) {
     return new ServiceMethod<>() {
       @Override
       public ServiceName serviceName() {
@@ -29,6 +33,10 @@ public String methodName() {
         return methodName;
       }
       @Override
+      public MethodType type() {
+        return type;
+      }
+      @Override
       public GrpcMessageDecoder<Resp> decoder() {
         return decoder;
       }
@@ -40,6 +48,10 @@ public GrpcMessageEncoder<Req> encoder() {
   }
 
   static <Req, Resp> ServiceMethod<Req, Resp> server(ServiceName serviceName, String methodName, GrpcMessageEncoder<Resp> encoder, GrpcMessageDecoder<Req> decoder) {
+    return server(serviceName, methodName, MethodType.UNARY, encoder, decoder);
+  }
+
+  static <Req, Resp> ServiceMethod<Req, Resp> server(ServiceName serviceName, String methodName, MethodType type, GrpcMessageEncoder<Resp> encoder, GrpcMessageDecoder<Req> decoder) {
     return new ServiceMethod<>() {
       @Override
       public ServiceName serviceName() {
@@ -50,6 +62,10 @@ public String methodName() {
         return methodName;
       }
       @Override
+      public MethodType type() {
+        return type;
+      }
+      @Override
       public GrpcMessageDecoder<Req> decoder() {
         return decoder;
       }
@@ -70,6 +86,24 @@ public GrpcMessageEncoder<Resp> encoder() {
    */
   String methodName();
 
+  /**
+   * Retrieves the cardinality of the gRPC service method.
+   * By default, it will return {@code MethodType.UNARY}, indicating
+   * a single request and a single response.
+   *
+   * @return the {@code MethodType} of the service method
+   */
+  default MethodType type() {
+    return MethodType.UNARY;
+  }
+
+  /**
+   * Computes the fully qualified method name for a gRPC service method.
+   * The name is constructed by combining the fully qualified service name
+   * and the method name, separated by a slash ('/').
+   *
+   * @return the fully qualified method name in the format "fullyQualifiedServiceName/methodName".
+   */
   default String fullMethodName() {
     return serviceName().fullyQualifiedName() + "/" + methodName();
   }

vertx-grpc-docs/src/main/java/examples/GrpcEventBusExamples.java

@@ -3,6 +3,7 @@
 import examples.grpc.*;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
+import io.vertx.core.streams.ReadStream;
 import io.vertx.docgen.Source;
 import io.vertx.grpc.common.WireFormat;
 import io.vertx.grpc.eventbus.EventBusGrpcClient;
@@ -52,4 +53,30 @@ public void jsonWireFormat(Vertx vertx) {
     greeter.sayHello(HelloRequest.newBuilder().setName("World").build())
       .onSuccess(reply -> System.out.println("Received: " + reply.getMessage()));
   }
+
+  public void streamingServer(Vertx vertx) {
+    EventBusGrpcServer server = EventBusGrpcServer.server(vertx);
+
+    Service service = StreamingGrpcService.of(new StreamingService() {
+      @Override
+      public Future<ReadStream<Item>> pipe(ReadStream<Item> request) {
+        return Future.succeededFuture(request);
+      }
+    });
+
+    server.addService(service);
+  }
+
+  public void streamingClient(Vertx vertx) {
+    EventBusGrpcClient client = EventBusGrpcClient.client(vertx);
+
+    StreamingClient streaming = StreamingGrpcClient.create(client);
+
+    streaming.pipe((stream, err) -> {
+      stream.write(Item.newBuilder().setValue("a").build());
+      stream.write(Item.newBuilder().setValue("b").build());
+      stream.end();
+    }).onSuccess(response -> response
+      .handler(item -> System.out.println("Received: " + item.getValue())));
+  }
 }

vertx-grpc-docs/src/main/java/examples/grpc/GreeterGrpcClient.java

@@ -25,6 +25,7 @@ public interface GreeterGrpcClient extends GreeterClient {
   ServiceMethod<examples.grpc.HelloReply, examples.grpc.HelloRequest> SayHello = ServiceMethod.client(
     ServiceName.create("examples.grpc", "Greeter"),
     "SayHello",
+    io.vertx.grpc.common.MethodType.UNARY,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.HelloReply.newBuilder()));
 

vertx-grpc-docs/src/main/java/examples/grpc/StreamingGrpcClient.java

@@ -25,6 +25,7 @@ public interface StreamingGrpcClient extends StreamingClient {
   ServiceMethod<examples.grpc.Item, examples.grpc.Empty> Source = ServiceMethod.client(
     ServiceName.create("examples.grpc", "Streaming"),
     "Source",
+    io.vertx.grpc.common.MethodType.SERVER_STREAMING,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));
 
@@ -34,6 +35,7 @@ public interface StreamingGrpcClient extends StreamingClient {
   ServiceMethod<examples.grpc.Empty, examples.grpc.Item> Sink = ServiceMethod.client(
     ServiceName.create("examples.grpc", "Streaming"),
     "Sink",
+    io.vertx.grpc.common.MethodType.CLIENT_STREAMING,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Empty.newBuilder()));
 
@@ -43,6 +45,7 @@ public interface StreamingGrpcClient extends StreamingClient {
   ServiceMethod<examples.grpc.Item, examples.grpc.Item> Pipe = ServiceMethod.client(
     ServiceName.create("examples.grpc", "Streaming"),
     "Pipe",
+    io.vertx.grpc.common.MethodType.BIDI,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));
 

vertx-grpc-docs/src/main/java/examples/grpc/StreamingGrpcService.java

@@ -71,6 +71,7 @@ public static Service of(StreamingService service) {
   public static final ServiceMethod<examples.grpc.Empty, examples.grpc.Item> Source = ServiceMethod.server(
     SERVICE_NAME,
     "Source",
+    io.vertx.grpc.common.MethodType.SERVER_STREAMING,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Empty.newBuilder()));
 
@@ -80,6 +81,7 @@ public static Service of(StreamingService service) {
   public static final ServiceMethod<examples.grpc.Item, examples.grpc.Empty> Sink = ServiceMethod.server(
     SERVICE_NAME,
     "Sink",
+    io.vertx.grpc.common.MethodType.CLIENT_STREAMING,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));
 
@@ -89,6 +91,7 @@ public static Service of(StreamingService service) {
   public static final ServiceMethod<examples.grpc.Item, examples.grpc.Item> Pipe = ServiceMethod.server(
     SERVICE_NAME,
     "Pipe",
+    io.vertx.grpc.common.MethodType.BIDI,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));