Added handlers for One/Many to Many stream

vert-x3 · Feb 8, 2024 · 6098af5 · 6098af5
1 parent 2f4457d
commit 6098af5
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 1 deletion.
diff --git a/vertx-grpc-protoc-plugin/src/main/resources/VertxStub.mustache b/vertx-grpc-protoc-plugin/src/main/resources/VertxStub.mustache
@@ -57,6 +57,10 @@ public final class {{className}} {
             return io.vertx.grpc.stub.ClientCalls.{{vertxCallsMethodName}}(ctx, request, delegateStub::{{methodName}});
         }
 
+        public io.vertx.core.streams.ReadStream<{{outputType}}> {{methodName}}WithHandler({{inputType}} request, io.vertx.core.Handler<{{outputType}}> handler, io.vertx.core.Handler<java.lang.Void> endHandler, io.vertx.core.Handler<java.lang.Throwable> exceptionHandler) {
+            return io.vertx.grpc.stub.ClientCalls.{{vertxCallsMethodName}}(ctx, request, delegateStub::{{methodName}}, handler, endHandler, exceptionHandler);
+        }
+
         {{/unaryManyMethods}}
         {{#manyUnaryMethods}}
         {{{methodHeader}}}
@@ -74,6 +78,10 @@ public final class {{className}} {
         public io.vertx.core.streams.ReadStream<{{outputType}}> {{methodName}}WithExceptionHandler(io.vertx.core.Handler<io.vertx.core.streams.WriteStream<{{inputType}}>> hdlr, io.vertx.core.Handler<java.lang.Throwable> exceptionHandler) {
             return io.vertx.grpc.stub.ClientCalls.{{vertxCallsMethodName}}(ctx, hdlr, delegateStub::{{methodName}}, exceptionHandler);
         }
+
+        public io.vertx.core.streams.ReadStream<{{outputType}}> {{methodName}}WithHandler(io.vertx.core.Handler<io.vertx.core.streams.WriteStream<{{inputType}}>> hdlr, io.vertx.core.Handler<{{outputType}}> handler, io.vertx.core.Handler<java.lang.Void> endHandler, io.vertx.core.Handler<java.lang.Throwable> exceptionHandler) {
+            return io.vertx.grpc.stub.ClientCalls.{{vertxCallsMethodName}}(ctx, hdlr, delegateStub::{{methodName}}, handler, endHandler, exceptionHandler);
+        }
         {{/manyManyMethods}}
     }
 

diff --git a/vertx-grpc/src/main/java/io/vertx/grpc/stub/ClientCalls.java b/vertx-grpc/src/main/java/io/vertx/grpc/stub/ClientCalls.java
@@ -43,7 +43,12 @@ public static <I, O> Future<O> oneToOne(ContextInternal ctx, I request, BiConsum
   }
 
   public static <I, O> ReadStream<O> oneToMany(ContextInternal ctx, I request, BiConsumer<I, StreamObserver<O>> delegate) {
+    return oneToMany(ctx, request, delegate, null, null, null);
+  }
+
+  public static <I, O> ReadStream<O> oneToMany(ContextInternal ctx, I request, BiConsumer<I, StreamObserver<O>> delegate, Handler<O> handler, Handler<Void> endHandler, Handler<Throwable> exceptionHandler) {
     StreamObserverReadStream<O> response = new StreamObserverReadStream<>();
+    response.handler(handler).endHandler(endHandler).exceptionHandler(exceptionHandler);
     delegate.accept(request, response);
     return response;
   }
@@ -60,8 +65,12 @@ public static <I, O> ReadStream<O> manyToMany(ContextInternal ctx, Handler<Write
   }
 
   public static <I, O> ReadStream<O> manyToMany(ContextInternal ctx, Handler<WriteStream<I>> requestHandler, Function<StreamObserver<O>, StreamObserver<I>> delegate, Handler<Throwable> exceptionHandler) {
+    return manyToMany(ctx, requestHandler, delegate, null, null, null);
+  }
+
+  public static <I, O> ReadStream<O> manyToMany(ContextInternal ctx, Handler<WriteStream<I>> requestHandler, Function<StreamObserver<O>, StreamObserver<I>> delegate, Handler<O> handler, Handler<Void> endHandler, Handler<Throwable> exceptionHandler) {
     StreamObserverReadStream<O> response = new StreamObserverReadStream<>();
-    response.exceptionHandler(exceptionHandler);
+    response.handler(handler).endHandler(endHandler).exceptionHandler(exceptionHandler);
     StreamObserver<I> request = delegate.apply(response);
     requestHandler.handle(new GrpcWriteStream<>(request));
     return response;