Add gRPC streaming over the event bus

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, null, null, encoder, decoder);
+  }
+
+  static <Req, Resp> ServiceMethod<Resp, Req> client(ServiceName serviceName, String methodName, Boolean clientStreaming, Boolean serverStreaming, GrpcMessageEncoder<Req> encoder, GrpcMessageDecoder<Resp> decoder) {
     return new ServiceMethod<>() {
       @Override
       public ServiceName serviceName() {
@@ -29,6 +33,14 @@ public String methodName() {
         return methodName;
       }
       @Override
+      public Boolean clientStreaming() {
+        return clientStreaming;
+      }
+      @Override
+      public Boolean serverStreaming() {
+        return serverStreaming;
+      }
+      @Override
       public GrpcMessageDecoder<Resp> decoder() {
         return decoder;
       }
@@ -40,6 +52,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, null, null, encoder, decoder);
+  }
+
+  static <Req, Resp> ServiceMethod<Req, Resp> server(ServiceName serviceName, String methodName, Boolean clientStreaming, Boolean serverStreaming, GrpcMessageEncoder<Resp> encoder, GrpcMessageDecoder<Req> decoder) {
     return new ServiceMethod<>() {
       @Override
       public ServiceName serviceName() {
@@ -50,6 +66,14 @@ public String methodName() {
         return methodName;
       }
       @Override
+      public Boolean clientStreaming() {
+        return clientStreaming;
+      }
+      @Override
+      public Boolean serverStreaming() {
+        return serverStreaming;
+      }
+      @Override
       public GrpcMessageDecoder<Req> decoder() {
         return decoder;
       }
@@ -70,6 +94,27 @@ public GrpcMessageEncoder<Resp> encoder() {
    */
   String methodName();
 
+  /**
+   * @return whether the client side sends a stream of requests, {@code null} when this is not known
+   */
+  default Boolean clientStreaming() {
+    return null;
+  }
+
+  /**
+   * @return whether the server side sends a stream of responses, {@code null} when this is not known
+   */
+  default Boolean serverStreaming() {
+    return null;
+  }
+
+  /**
+   * 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;
@@ -13,16 +14,14 @@
 public class GrpcEventBusExamples {
 
   public void createServer(Vertx vertx) {
-    EventBusGrpcServer server = EventBusGrpcServer.server(vertx);
+    Future<EventBusGrpcServer> server = EventBusGrpcServer.server(vertx);
   }
 
   public void createClient(Vertx vertx) {
-    EventBusGrpcClient client = EventBusGrpcClient.client(vertx);
+    Future<EventBusGrpcClient> client = EventBusGrpcClient.client(vertx);
   }
 
   public void serverWithService(Vertx vertx) {
-    EventBusGrpcServer server = EventBusGrpcServer.server(vertx);
-
     Service service = GreeterGrpcService.of(new GreeterService() {
       @Override
       public Future<HelloReply> sayHello(HelloRequest request) {
@@ -32,24 +31,48 @@ public Future<HelloReply> sayHello(HelloRequest request) {
       }
     });
 
-    server.addService(service);
+    EventBusGrpcServer.server(vertx).onSuccess(server -> server.addService(service));
   }
 
   public void clientWithService(Vertx vertx) {
-    EventBusGrpcClient client = EventBusGrpcClient.client(vertx);
-
-    GreeterClient greeter = GreeterGrpcClient.create(client);
+    EventBusGrpcClient.client(vertx).onSuccess(client -> {
+      GreeterClient greeter = GreeterGrpcClient.create(client);
 
-    greeter.sayHello(HelloRequest.newBuilder().setName("World").build())
-      .onSuccess(reply -> System.out.println("Received: " + reply.getMessage()));
+      greeter.sayHello(HelloRequest.newBuilder().setName("World").build())
+        .onSuccess(reply -> System.out.println("Received: " + reply.getMessage()));
+    });
   }
 
   public void jsonWireFormat(Vertx vertx) {
-    EventBusGrpcClient client = EventBusGrpcClient.client(vertx);
+    EventBusGrpcClient.client(vertx).onSuccess(client -> {
+      GreeterClient greeter = GreeterGrpcClient.create(client, WireFormat.JSON);
 
-    GreeterClient greeter = GreeterGrpcClient.create(client, WireFormat.JSON);
+      greeter.sayHello(HelloRequest.newBuilder().setName("World").build())
+        .onSuccess(reply -> System.out.println("Received: " + reply.getMessage()));
+    });
+  }
 
-    greeter.sayHello(HelloRequest.newBuilder().setName("World").build())
-      .onSuccess(reply -> System.out.println("Received: " + reply.getMessage()));
+  public void streamingServer(Vertx vertx) {
+    Service service = StreamingGrpcService.of(new StreamingService() {
+      @Override
+      public Future<ReadStream<Item>> pipe(ReadStream<Item> request) {
+        return Future.succeededFuture(request);
+      }
+    });
+
+    EventBusGrpcServer.server(vertx).onSuccess(server -> server.addService(service));
+  }
+
+  public void streamingClient(Vertx vertx) {
+    EventBusGrpcClient.client(vertx).onSuccess(client -> {
+      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/GrpcIoClientExamples.java

@@ -49,12 +49,11 @@ public void onError(Throwable t) {
   }
 
   public void eventBusExample(Vertx vertx) {
+    EventBusGrpcClient.client(vertx).onSuccess(client -> {
+      GrpcIoClientChannel channel = new GrpcIoClientChannel(client);
 
-    EventBusGrpcClient client = EventBusGrpcClient.client(vertx);
-
-    GrpcIoClientChannel channel = new GrpcIoClientChannel(client);
-
-    GreeterGrpc.GreeterStub greeter = GreeterGrpc.newStub(channel);
+      GreeterGrpc.GreeterStub greeter = GreeterGrpc.newStub(channel);
+    });
   }
 
   public void stubWithDeadline(GrpcIoClientChannel channel, StreamObserver<HelloReply> observer) {

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

@@ -50,12 +50,9 @@ public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseOb
   }
 
   public void eventBusExample(Vertx vertx, GreeterGrpc.GreeterImplBase service) {
-
-    EventBusGrpcServer server = EventBusGrpcServer.server(vertx);
-
     GrpcIoServiceBridge bridge = GrpcIoServiceBridge.bridge(service);
 
-    server.addService(bridge);
+    EventBusGrpcServer.server(vertx).onSuccess(server -> server.addService(bridge));
   }
 
   public void reflectionExample(Vertx vertx, HttpServerConfig config, ServerSSLOptions sslOptions) {

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

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

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

@@ -87,6 +87,8 @@ public static Service of(GreeterService service) {
   public static final io.vertx.grpc.transcoding.TranscodingServiceMethod<examples.grpc.HelloRequest, examples.grpc.HelloReply> SayHello = io.vertx.grpc.transcoding.TranscodingServiceMethod.server(
     SERVICE_NAME,
     "SayHello",
+    false,
+    false,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.HelloRequest.newBuilder()),
     SayHello_OPTIONS

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

@@ -25,6 +25,8 @@ public interface StreamingGrpcClient extends StreamingClient {
   ServiceMethod<examples.grpc.Item, examples.grpc.Empty> Source = ServiceMethod.client(
     ServiceName.create("examples.grpc", "Streaming"),
     "Source",
+    false,
+    true,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));
 
@@ -34,6 +36,8 @@ public interface StreamingGrpcClient extends StreamingClient {
   ServiceMethod<examples.grpc.Empty, examples.grpc.Item> Sink = ServiceMethod.client(
     ServiceName.create("examples.grpc", "Streaming"),
     "Sink",
+    true,
+    false,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Empty.newBuilder()));
 
@@ -43,6 +47,8 @@ public interface StreamingGrpcClient extends StreamingClient {
   ServiceMethod<examples.grpc.Item, examples.grpc.Item> Pipe = ServiceMethod.client(
     ServiceName.create("examples.grpc", "Streaming"),
     "Pipe",
+    true,
+    true,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));
 

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

@@ -71,6 +71,8 @@ public static Service of(StreamingService service) {
   public static final ServiceMethod<examples.grpc.Empty, examples.grpc.Item> Source = ServiceMethod.server(
     SERVICE_NAME,
     "Source",
+    false,
+    true,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Empty.newBuilder()));
 
@@ -80,6 +82,8 @@ public static Service of(StreamingService service) {
   public static final ServiceMethod<examples.grpc.Item, examples.grpc.Empty> Sink = ServiceMethod.server(
     SERVICE_NAME,
     "Sink",
+    true,
+    false,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));
 
@@ -89,6 +93,8 @@ public static Service of(StreamingService service) {
   public static final ServiceMethod<examples.grpc.Item, examples.grpc.Item> Pipe = ServiceMethod.server(
     SERVICE_NAME,
     "Pipe",
+    true,
+    true,
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.Item.newBuilder()));