Support server-streaming in transcoding

vertx-grpc-docs/src/main/asciidoc/transcoding.adoc

@@ -212,6 +212,43 @@ message HelloResponse {
 }
 ----
 
+=== Server-streaming responses
+
+Server-streaming RPCs (`rpc Foo (Req) returns (stream Resp)`) are transcoded by emitting the response over chunked transfer encoding. The wire format is selected according to the HTTP `accept` header, letting the client choose the most appropriate encoding:
+
+|===
+| `Accept` value | Response `Content-Type` | Framing
+
+| `application/json` (default) | `application/json` | Top-level JSON array: `[m1, m2, m3]`. Envoy-compatible. The body is not valid JSON until the final `]` arrives, so `JSON.parse(body)` only works after the stream completes.
+| `application/x-ndjson` | `application/x-ndjson` | One JSON object per line, `\n`-separated. Each line is independently parseable as it arrives.
+| `text/event-stream` | `text/event-stream` | Server-Sent Events: `data: <json>\n\n` per message. Consumable by the browser `EventSource` API.
+|===
+
+[source]
+----
+# JSON array (default)
+curl -X POST -H "Content-Type: application/json" -d '...' http://localhost:8080/stream
+# [{"payload":"first"},{"payload":"second"}]
+
+# NDJSON
+curl -X POST -H "Accept: application/x-ndjson" -H "Content-Type: application/json" -d '...' http://localhost:8080/stream
+# {"payload":"first"}
+# {"payload":"second"}
+
+# Server-Sent Events
+curl -X POST -H "Accept: text/event-stream" -H "Content-Type: application/json" -d '...' http://localhost:8080/stream
+# data: {"payload":"first"}
+#
+# data: {"payload":"second"}
+----
+
+A gRPC error is reported to the client depending on when it happens:
+
+- a gRPC trailers-only response sends the corresponding HTTP status error without content
+- a gRPC trailers response terminates the response with `]`, such error cannot be reported to the client
+
+Transcoding does not support client-streaming and bidirectional-streaming RPCs.
+
 === Transcoding error handling
 
 If an error occurs during transcoding, the server will return an HTTP error response with the appropriate status code.

vertx-grpc-it/src/test/java/io/vertx/grpc/it/tests/TranscodingTest.java

@@ -1,23 +1,39 @@
 package io.vertx.grpc.it.tests;
 
 import io.grpc.examples.helloworld.*;
+import io.grpc.examples.streamingtranscoding.StreamingHelloReply;
+import io.grpc.examples.streamingtranscoding.StreamingHelloRequest;
+import io.grpc.examples.streamingtranscoding.StreamingTranscodingGreeterClient;
+import io.grpc.examples.streamingtranscoding.StreamingTranscodingGreeterGrpcClient;
+import io.grpc.examples.streamingtranscoding.StreamingTranscodingGreeterGrpcService;
+import io.grpc.examples.streamingtranscoding.StreamingTranscodingGreeterService;
 import io.grpc.stub.StreamObserver;
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.*;
 import io.vertx.core.json.Json;
+import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.SocketAddress;
+import io.vertx.core.streams.WriteStream;
 import io.vertx.grpc.client.GrpcClient;
 import io.vertx.grpc.server.GrpcServer;
+import io.vertx.grpc.server.GrpcServerResponse;
 import io.vertx.grpcio.server.GrpcIoServer;
 import org.junit.Test;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 public class TranscodingTest extends ProxyTestBase {
 
@@ -344,6 +360,217 @@ public void testUnaryCollisionWithoutOption() throws TimeoutException {
     assertEquals("Hello Julien", reply.getMessage());
   }
 
+  @Test
+  public void testUnaryAddService() throws TimeoutException {
+    HttpClient client = vertx.createHttpClient();
+
+    vertx.createHttpServer()
+      .requestHandler(GrpcServer.server(vertx).addService(GreeterGrpcService.of(new GreeterService() {
+        @Override
+        public Future<HelloReply> sayHello(HelloRequest request) {
+          return Future.succeededFuture(HelloReply.newBuilder().setMessage("Hello " + request.getName()).build());
+        }
+      }))).listen(8080, "localhost").await(10, TimeUnit.SECONDS);
+
+    RequestOptions options = new RequestOptions().setHost("localhost").setPort(8080).setURI("/v1/hello/Julien").setMethod(HttpMethod.GET);
+
+    Buffer body = client.request(options).compose(req -> {
+        req.putHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+        req.putHeader(HttpHeaders.ACCEPT, "application/json");
+        return req.send();
+      }).expecting(HttpResponseExpectation.SC_OK)
+      .expecting(HttpResponseExpectation.JSON)
+      .compose(HttpClientResponse::body)
+      .await(10, TimeUnit.SECONDS);
+    assertEquals("Hello Julien", getMessage(body.toString()));
+  }
+
+  @Test
+  public void testServerStreamingAddService() throws TimeoutException {
+    HttpClient client = vertx.createHttpClient();
+
+    vertx.createHttpServer()
+      .requestHandler(GrpcServer.server(vertx).addService(StreamingTranscodingGreeterGrpcService.of(new StreamingTranscodingGreeterService() {
+        @Override
+        protected void sayHelloStreaming(StreamingHelloRequest request, WriteStream<StreamingHelloReply> response) {
+          response.write(StreamingHelloReply.newBuilder().setMessage("Hello " + request.getName() + " 1").build());
+          response.write(StreamingHelloReply.newBuilder().setMessage("Hello " + request.getName() + " 2").build());
+          response.end();
+        }
+      }))).listen(8080, "localhost").await(10, TimeUnit.SECONDS);
+
+    RequestOptions options = new RequestOptions().setHost("localhost").setPort(8080).setURI("/v1/hello/stream/Julien").setMethod(HttpMethod.GET);
+
+    Buffer body = client.request(options).compose(req -> {
+        req.putHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+        req.putHeader(HttpHeaders.ACCEPT, "application/json");
+        return req.send();
+      }).expecting(HttpResponseExpectation.SC_OK)
+      .compose(HttpClientResponse::body)
+      .await(10, TimeUnit.SECONDS);
+
+    JsonArray array = new JsonArray(body);
+    assertEquals(2, array.size());
+    assertEquals("Hello Julien 1", array.getJsonObject(0).getString("message"));
+    assertEquals("Hello Julien 2", array.getJsonObject(1).getString("message"));
+  }
+
+  @Test
+  public void testServerStreaming() throws TimeoutException {
+    HttpClient client = vertx.createHttpClient();
+
+    vertx.createHttpServer()
+      .requestHandler(GrpcServer.server(vertx).callHandler(StreamingTranscodingGreeterGrpcService.SayHelloStreaming, call -> call.handler(request -> {
+        GrpcServerResponse<StreamingHelloRequest, StreamingHelloReply> response = call.response();
+        response.write(StreamingHelloReply.newBuilder().setMessage("Hello " + request.getName() + " 1").build());
+        response.write(StreamingHelloReply.newBuilder().setMessage("Hello " + request.getName() + " 2").build());
+        response.write(StreamingHelloReply.newBuilder().setMessage("Hello " + request.getName() + " 3").build());
+        response.end();
+      }))).listen(8080, "localhost").await(10, TimeUnit.SECONDS);
+
+    RequestOptions options = new RequestOptions().setHost("localhost").setPort(8080).setURI("/v1/hello/stream/Julien").setMethod(HttpMethod.GET);
+
+    HttpClientResponse response = client.request(options).compose(req -> {
+        req.putHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+        req.putHeader(HttpHeaders.ACCEPT, "application/json");
+        return req.send();
+      }).expecting(HttpResponseExpectation.SC_OK)
+      .compose(resp -> resp.body().map(resp))
+      .await(10, TimeUnit.SECONDS);
+
+    assertTrue(response.headers().contains(HttpHeaders.CONTENT_TYPE, "application/json", true));
+    // Streaming responses are chunked, so the length is not known up-front.
+    assertFalse(response.headers().contains(HttpHeaders.CONTENT_LENGTH));
+    JsonArray array = new JsonArray(response.body().result());
+    assertEquals(3, array.size());
+    assertEquals("Hello Julien 1", array.getJsonObject(0).getString("message"));
+    assertEquals("Hello Julien 2", array.getJsonObject(1).getString("message"));
+    assertEquals("Hello Julien 3", array.getJsonObject(2).getString("message"));
+  }
+
+  @Test
+  public void testServerStreamingNdjson() throws TimeoutException {
+    HttpClient client = vertx.createHttpClient();
+
+    vertx.createHttpServer()
+      .requestHandler(GrpcServer.server(vertx).callHandler(StreamingTranscodingGreeterGrpcService.SayHelloStreaming, call -> call.handler(request -> {
+        GrpcServerResponse<StreamingHelloRequest, StreamingHelloReply> response = call.response();
+        response.write(StreamingHelloReply.newBuilder().setMessage("Hello " + request.getName() + " 1").build());
+        response.write(StreamingHelloReply.newBuilder().setMessage("Hello " + request.getName() + " 2").build());
+        response.end();
+      }))).listen(8080, "localhost").await(10, TimeUnit.SECONDS);
+
+    RequestOptions options = new RequestOptions().setHost("localhost").setPort(8080).setURI("/v1/hello/stream/Julien").setMethod(HttpMethod.GET);
+
+    HttpClientResponse response = client.request(options).compose(req -> {
+        req.putHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+        req.putHeader(HttpHeaders.ACCEPT, "application/x-ndjson");
+        return req.send();
+      }).expecting(HttpResponseExpectation.SC_OK)
+      .compose(resp -> resp.body().map(resp))
+      .await(10, TimeUnit.SECONDS);
+
+    assertTrue(response.headers().contains(HttpHeaders.CONTENT_TYPE, "application/x-ndjson", true));
+    String[] lines = response.body().result().toString().split("\n");
+    assertEquals(2, lines.length);
+    assertEquals("Hello Julien 1", new JsonObject(lines[0]).getString("message"));
+    assertEquals("Hello Julien 2", new JsonObject(lines[1]).getString("message"));
+  }
+
+  @Test
+  public void testServerStreamingSse() throws TimeoutException {
+    HttpClient client = vertx.createHttpClient();
+
+    vertx.createHttpServer()
+      .requestHandler(GrpcServer.server(vertx).callHandler(StreamingTranscodingGreeterGrpcService.SayHelloStreaming, call -> call.handler(request -> {
+        GrpcServerResponse<StreamingHelloRequest, StreamingHelloReply> response = call.response();
+        response.write(StreamingHelloReply.newBuilder().setMessage("Hello " + request.getName() + " 1").build());
+        response.write(StreamingHelloReply.newBuilder().setMessage("Hello " + request.getName() + " 2").build());
+        response.end();
+      }))).listen(8080, "localhost").await(10, TimeUnit.SECONDS);
+
+    RequestOptions options = new RequestOptions().setHost("localhost").setPort(8080).setURI("/v1/hello/stream/Julien").setMethod(HttpMethod.GET);
+
+    HttpClientResponse response = client.request(options).compose(req -> {
+        req.putHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+        req.putHeader(HttpHeaders.ACCEPT, "text/event-stream");
+        return req.send();
+      }).expecting(HttpResponseExpectation.SC_OK)
+      .compose(resp -> resp.body().map(resp))
+      .await(10, TimeUnit.SECONDS);
+
+    assertTrue(response.headers().contains(HttpHeaders.CONTENT_TYPE, "text/event-stream", true));
+    String[] events = response.body().result().toString().split("\n\n");
+    assertEquals(2, events.length);
+    assertTrue(events[0].startsWith("data: "));
+    assertTrue(events[1].startsWith("data: "));
+    assertEquals("Hello Julien 1", new JsonObject(events[0].substring("data: ".length())).getString("message"));
+    assertEquals("Hello Julien 2", new JsonObject(events[1].substring("data: ".length())).getString("message"));
+  }
+
+  @Test
+  public void testServerStreamingEmpty() throws TimeoutException {
+    HttpClient client = vertx.createHttpClient();
+
+    vertx.createHttpServer()
+      .requestHandler(GrpcServer.server(vertx).callHandler(StreamingTranscodingGreeterGrpcService.SayHelloStreaming, call -> call.handler(request -> {
+        GrpcServerResponse<StreamingHelloRequest, StreamingHelloReply> response = call.response();
+        response.end();
+      }))).listen(8080, "localhost").await(10, TimeUnit.SECONDS);
+
+    RequestOptions options = new RequestOptions().setHost("localhost").setPort(8080).setURI("/v1/hello/stream/Julien").setMethod(HttpMethod.GET);
+
+    Buffer body = client.request(options).compose(req -> {
+        req.putHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+        req.putHeader(HttpHeaders.ACCEPT, "application/json");
+        return req.send();
+      }).expecting(HttpResponseExpectation.SC_OK)
+      .compose(HttpClientResponse::body)
+      .await(10, TimeUnit.SECONDS);
+
+    assertEquals(0, new JsonArray(body).size());
+  }
+
+  @Test
+  public void testServerStreamingGrpcCollision() throws TimeoutException {
+    HttpClient httpClient = vertx.createHttpClient();
+
+    vertx.createHttpServer()
+      .requestHandler(GrpcServer.server(vertx).callHandler(StreamingTranscodingGreeterGrpcService.SayHelloStreaming, call -> call.handler(request -> {
+        GrpcServerResponse<StreamingHelloRequest, StreamingHelloReply> response = call.response();
+        response.write(StreamingHelloReply.newBuilder().setMessage("Hello " + request.getName()).build());
+        response.end();
+      }))).listen(8080, "localhost").await(10, TimeUnit.SECONDS);
+
+    RequestOptions options = new RequestOptions().setHost("localhost").setPort(8080).setURI("/v1/hello/stream/Julien").setMethod(HttpMethod.GET);
+
+    Buffer httpBody = httpClient.request(options).compose(req -> {
+        req.putHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+        req.putHeader(HttpHeaders.ACCEPT, "application/json");
+        return req.send();
+      }).expecting(HttpResponseExpectation.SC_OK)
+      .compose(HttpClientResponse::body)
+      .await(10, TimeUnit.SECONDS);
+    assertEquals("Hello Julien", new JsonArray(httpBody).getJsonObject(0).getString("message"));
+
+    // The same service method is still reachable over plain gRPC.
+    GrpcClient grpcClient = GrpcClient.client(vertx);
+    StreamingTranscodingGreeterClient greeterClient = StreamingTranscodingGreeterGrpcClient.create(grpcClient, SocketAddress.inetSocketAddress(8080, "localhost"));
+
+    List<String> received = greeterClient
+      .sayHelloStreaming(StreamingHelloRequest.newBuilder().setName("Julien").build())
+      .compose(stream -> {
+        Promise<List<String>> promise = Promise.promise();
+        List<String> replies = new ArrayList<>();
+        stream.handler(reply -> replies.add(reply.getMessage()));
+        stream.endHandler(v -> promise.tryComplete(replies));
+        stream.exceptionHandler(promise::tryFail);
+        return promise.future();
+      })
+      .await(10, TimeUnit.SECONDS);
+    assertEquals(Collections.singletonList("Hello Julien"), received);
+  }
+
   @Test
   public void testUnknownService() throws TimeoutException {
     HttpClient client = vertx.createHttpClient();

vertx-grpc-server/src/main/java/io/vertx/grpc/server/impl/GrpcServerImpl.java

@@ -254,7 +254,14 @@ public GrpcServer addService(Service service) {
       }
       for (ServiceMethod method : service.methods()) {
         Handler handler = service.handler(method);
-        registerMethodCallHandler(service.pathOfMethod(method.methodName()), new ServiceMethodCallHandler<Object, Object>(method, handler));
+        ServiceMethodCallHandler<Object, Object> smch = new ServiceMethodCallHandler<>(method, handler);
+        if (method instanceof MountPoint) {
+          MountPoint<Object, Object> mountPoint = (MountPoint<Object, Object>) method;
+          for (String path : mountPoint.paths()) {
+            registerMethodCallHandler(path, smch);
+          }
+        }
+        registerMethodCallHandler(service.pathOfMethod(method.methodName()), smch);
       }
 
       this.services.add(service);