Support server-streaming in transcoding

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

@@ -212,6 +212,40 @@ 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 per-request from the HTTP `Accept` header so clients can pick the encoding that fits their consumption pattern without any server configuration:
+
+|===
+| `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"}
+----
+
+If the RPC terminates with a non-OK gRPC status before any message has been written, the response is finished with the corresponding HTTP status code and no body. If messages have already been written, the stream ends (with the closing `]` for JSON array mode). Mid-stream errors cannot be signalled on the body since the HTTP status was already sent.
+
+Client-streaming and bidirectional-streaming RPCs are not supported by transcoding.
+
 === Transcoding error handling
 
 If an error occurs during transcoding, the server will return an HTTP error response with the appropriate status code.

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

@@ -89,7 +89,8 @@ public static Service of(GreeterService service) {
     "SayHello",
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder(examples.grpc.HelloRequest.newBuilder()),
-    SayHello_OPTIONS
+    SayHello_OPTIONS,
+    false
   );
 
   private final Invoker invoker = new Invoker(this, all());

vertx-grpc-protoc-plugin2/src/main/resources/grpc-service.mustache

@@ -117,7 +117,8 @@ public class {{grpcServiceFqn}} extends {{serviceFqn}} implements Service {
     "{{methodName}}",
     GrpcMessageEncoder.encoder(),
     GrpcMessageDecoder.decoder({{inputType}}.newBuilder()),
-    {{methodName}}_OPTIONS
+    {{methodName}}_OPTIONS,
+    {{isManyOutput}}
   );
 {{/transcodingMethods}}
 

vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/TranscodingServiceMethod.java

@@ -24,7 +24,16 @@ static <Req, Resp> TranscodingServiceMethod<Req, Resp> server(ServiceName servic
                                                                 GrpcMessageEncoder<Resp> encoder,
                                                                 GrpcMessageDecoder<Req> decoder,
                                                                 MethodTranscodingOptions options) {
-    return new TranscodingServiceMethodImpl<>(serviceName, methodName, encoder, decoder, options);
+    return new TranscodingServiceMethodImpl<>(serviceName, methodName, encoder, decoder, options, false);
+  }
+
+  static <Req, Resp> TranscodingServiceMethod<Req, Resp> server(ServiceName serviceName,
+                                                                String methodName,
+                                                                GrpcMessageEncoder<Resp> encoder,
+                                                                GrpcMessageDecoder<Req> decoder,
+                                                                MethodTranscodingOptions options,
+                                                                boolean streaming) {
+    return new TranscodingServiceMethodImpl<>(serviceName, methodName, encoder, decoder, options, streaming);
   }
 
   static <Req, Resp> TranscodingServiceMethod<Req, Resp> server(ServiceMethod<Req, Resp> serviceMethod,
@@ -34,4 +43,13 @@ static <Req, Resp> TranscodingServiceMethod<Req, Resp> server(ServiceMethod<Req,
 
   MethodTranscodingOptions options();
 
+  /**
+   * @return whether the response side of the RPC is streaming (server-streaming or bidi). When {@code true} the
+   * transcoder emits a JSON array of messages (one element per gRPC message) on the HTTP response, using chunked
+   * transfer encoding instead of a single content-length-framed JSON object.
+   */
+  default boolean streaming() {
+    return false;
+  }
+
 }

vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/impl/MessageWeaver.java

@@ -35,21 +35,22 @@ private MessageWeaver() {
   }
 
   /**
-   * Weaves HTTP variable bindings and request body into a gRPC message.
+   * Weaves HTTP variable bindings and request body into a gRPC message. The return type is always a {@link JsonObject}
+   * because a protobuf {@code Message} is always object-shaped in JSON form.
    *
    * @param message The original message buffer
    * @param bindings The HTTP variable bindings
    * @param transcodingRequestBody The transcoding request body path
    * @param descriptor The protobuf message descriptor, used to identify repeated fields
-   * @return The modified buffer with weaved content
+   * @return The woven message as a JsonObject
    * @throws DecodeException If JSON decoding fails
    */
-  public static Buffer weaveRequestMessage(Buffer message, List<HttpVariableBinding> bindings, String transcodingRequestBody, Descriptors.Descriptor descriptor) throws DecodeException {
+  public static JsonObject weaveRequestMessage(Buffer message, List<HttpVariableBinding> bindings, String transcodingRequestBody, Descriptors.Descriptor descriptor) throws DecodeException {
     boolean hasBindings = bindings != null && !bindings.isEmpty();
     boolean hasBody = transcodingRequestBody != null && !transcodingRequestBody.isEmpty();
 
     if (!hasBindings && !hasBody) {
-      return new JsonObject().toBuffer();
+      return new JsonObject();
     }
 
     JsonObject result = new JsonObject();
@@ -72,7 +73,7 @@ public static Buffer weaveRequestMessage(Buffer message, List<HttpVariableBindin
       applyBindings(result, bindings, descriptor);
     }
 
-    return result.toBuffer();
+    return result;
   }
 
   /**
@@ -138,32 +139,34 @@ private static void applyAtPath(JsonObject root, String[] path, Object value) {
   }
 
   /**
-   * Extracts a response message portion based on the transcoding path.
+   * Extracts a response message portion based on the {@code response_body} selector. The result may be any JSON value
+   * the selector lands on: a {@link JsonObject}, a {@link JsonArray} (for selectors pointing at a repeated proto field),
+   * or a scalar ({@link String}, {@link Number}, {@link Boolean}, or {@code null}). Callers should use
+   * {@link io.vertx.core.json.Json#encodeToBuffer(Object)} to serialize the result uniformly.
    *
    * @param message The original message buffer
    * @param transcodingResponseBody The path to extract from the response
-   * @return The modified buffer with the extracted content
+   * @return The selected JSON value (object, array, scalar, or null)
    */
-  public static Buffer weaveResponseMessage(Buffer message, String transcodingResponseBody) throws DecodeException {
+  public static Object weaveResponseMessage(Buffer message, String transcodingResponseBody) throws DecodeException {
     Objects.requireNonNull(message, "Message cannot be null");
 
     if (transcodingResponseBody == null || transcodingResponseBody.isEmpty() || transcodingResponseBody.equals(ROOT_LEVEL)) {
-      return message;
+      return message.toJsonObject();
     }
 
     JsonObject json = message.toJsonObject();
     String[] path = transcodingResponseBody.split("\\.");
 
-    JsonObject current = json;
-    for (String field : path) {
-      Object value = current.getValue(field);
-      if (value instanceof JsonObject) {
-        current = (JsonObject) value;
-      } else {
-        throw new IllegalArgumentException("Path segment '" + field + "' in transcodingResponseBody does not refer to a JSON object");
+    Object current = json;
+    for (int i = 0; i < path.length; i++) {
+      String field = path[i];
+      if (!(current instanceof JsonObject)) {
+        throw new IllegalArgumentException("Path segment '" + path[i - 1] + "' in transcodingResponseBody does not refer to a JSON object");
       }
+      current = ((JsonObject) current).getValue(field);
     }
 
-    return current.toBuffer();
+    return current;
   }
 }

vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/impl/TranscodingGrpcOutboundStream.java

@@ -8,36 +8,88 @@
 import io.vertx.core.http.HttpServerRequest;
 import io.vertx.core.http.HttpServerResponse;
 import io.vertx.core.internal.ContextInternal;
-import io.vertx.core.internal.buffer.BufferInternal;
+import io.vertx.core.json.Json;
 import io.vertx.grpc.common.CodecException;
 import io.vertx.grpc.common.GrpcStatus;
 import io.vertx.grpc.common.JsonWireFormat;
 import io.vertx.grpc.common.WireFormat;
+import io.vertx.grpc.common.impl.GrpcHeadersFrame;
 import io.vertx.grpc.common.impl.GrpcMessageDeframer;
 import io.vertx.grpc.common.impl.GrpcMessageFrame;
 import io.vertx.grpc.server.GrpcProtocol;
 import io.vertx.grpc.server.impl.HttpGrpcOutboundStream;
 
 public class TranscodingGrpcOutboundStream extends HttpGrpcOutboundStream {
 
+  /**
+   * Wire framing for server-streaming responses. Selected per-request from the {@code Accept} header so clients can opt into a format suited to their consumption pattern without
+   * server reconfiguration. Mirrors Envoy's {@code GrpcJsonTranscoder.PrintOptions} (stream_newline_delimited, stream_sse_style_delimited).
+   */
+  private enum StreamFormat {
+    /** Default: top-level JSON array, {@code application/json}. Envoy-compatible; not parseable until the stream ends. */
+    JSON_ARRAY("application/json"),
+    /** Newline-delimited JSON: one message per line. Each line is independently parseable. */
+    NDJSON("application/x-ndjson"),
+    /** Server-Sent Events: {@code data: <msg>\n\n} framing, consumable by browser {@code EventSource}. */
+    SSE("text/event-stream");
+
+    final String mediaType;
+
+    StreamFormat(String mediaType) {
+      this.mediaType = mediaType;
+    }
+  }
+
+  private static final Buffer OPEN_ARRAY = Buffer.buffer("[");
+  private static final Buffer EMPTY_ARRAY = Buffer.buffer("[]");
+  private static final Buffer COMMA = Buffer.buffer(",");
+  private static final Buffer CLOSE_ARRAY = Buffer.buffer("]");
+  private static final Buffer NEWLINE = Buffer.buffer("\n");
+  private static final Buffer SSE_PREFIX = Buffer.buffer("data: ");
+  private static final Buffer SSE_SUFFIX = Buffer.buffer("\n\n");
+
   private Promise<Void> head;
   private final ContextInternal context;
   private final HttpServerResponse httpResponse;
   private final String transcodingResponseBody;
+  private final boolean streaming;
+  private final StreamFormat streamFormat;
+  private boolean firstMessageWritten;
 
-  public TranscodingGrpcOutboundStream(ContextInternal context, HttpServerRequest httpRequest,
-                                       String transcodingResponseBody, GrpcMessageDeframer deframer) {
+  public TranscodingGrpcOutboundStream(
+    ContextInternal context,
+    HttpServerRequest httpRequest,
+    String transcodingResponseBody,
+    GrpcMessageDeframer deframer,
+    boolean streaming
+  ) {
     super(httpRequest, GrpcProtocol.TRANSCODING, deframer);
 
     this.context = context;
     this.httpResponse = httpRequest.response();
     this.transcodingResponseBody = transcodingResponseBody;
+    this.streaming = streaming;
+    this.streamFormat = streaming ? negotiateStreamFormat(httpRequest.getHeader(HttpHeaders.ACCEPT)) : StreamFormat.JSON_ARRAY;
+  }
+
+  private static StreamFormat negotiateStreamFormat(String acceptHeader) {
+    if (acceptHeader == null) {
+      return StreamFormat.JSON_ARRAY;
+    }
+    String accept = acceptHeader.toLowerCase();
+    if (accept.contains("text/event-stream")) {
+      return StreamFormat.SSE;
+    }
+    if (accept.contains("application/x-ndjson") || accept.contains("application/jsonl")) {
+      return StreamFormat.NDJSON;
+    }
+    return StreamFormat.JSON_ARRAY;
   }
 
   @Override
   protected String contentType(WireFormat wireFormat) {
     if (wireFormat instanceof JsonWireFormat) {
-      return protocol.mediaType();
+      return streaming ? streamFormat.mediaType : protocol.mediaType();
     }
     throw new UnsupportedOperationException();
   }
@@ -47,15 +99,20 @@ protected void encodeGrpcHeaders(MultiMap grpcHeaders, MultiMap httpHeaders, Str
   }
 
   @Override
-  public Future<Void> writeEnd() {
-    if (status != GrpcStatus.OK) {
-      httpResponse.setStatusCode(GrpcTranscodingError.fromHttp2Code(status.code).getHttpStatusCode());
+  protected Future<Void> writeHeaders(GrpcHeadersFrame frame) {
+    if (streaming) {
+      // Flush headers up-front with chunked transfer so each message can be emitted as soon as it arrives.
+      // The unary path defers writeHead until the first message so Content-Length can be set on the response.
+      httpResponse.setChunked(true);
     }
-    return super.writeEnd();
+    return super.writeHeaders(frame);
   }
 
   @Override
   public Future<Void> writeHead() {
+    if (streaming) {
+      return httpResponse.writeHead();
+    }
     head = context.promise();
     return head.future();
   }
@@ -68,9 +125,16 @@ public Future<Void> writeMessage(GrpcMessageFrame frame) {
     } catch (CodecException e) {
       return context.failedFuture(e);
     }
+    if (streaming) {
+      return writeStreamingMessage(payload);
+    }
+    return writeUnaryMessage(payload);
+  }
+
+  private Future<Void> writeUnaryMessage(Buffer payload) {
     Future<Void> res;
     try {
-      BufferInternal transcoded = (BufferInternal) MessageWeaver.weaveResponseMessage(payload, transcodingResponseBody);
+      Buffer transcoded = Json.encodeToBuffer(MessageWeaver.weaveResponseMessage(payload, transcodingResponseBody));
       httpResponse.putHeader(HttpHeaders.CONTENT_LENGTH, Integer.toString(transcoded.length()));
       httpResponse.putHeader(HttpHeaders.CONTENT_TYPE, GrpcProtocol.TRANSCODING.mediaType());
       res = httpResponse.write(transcoded);
@@ -83,4 +147,53 @@ public Future<Void> writeMessage(GrpcMessageFrame frame) {
     }
     return res;
   }
+
+  private Future<Void> writeStreamingMessage(Buffer payload) {
+    Buffer transcoded;
+    try {
+      transcoded = Json.encodeToBuffer(MessageWeaver.weaveResponseMessage(payload, transcodingResponseBody));
+    } catch (Exception e) {
+      httpResponse.setStatusCode(500).end();
+      return context.failedFuture(e);
+    }
+    Buffer chunk;
+    switch (streamFormat) {
+      case JSON_ARRAY: {
+        Buffer prefix = firstMessageWritten ? COMMA : OPEN_ARRAY;
+        chunk = Buffer.buffer(prefix.length() + transcoded.length()).appendBuffer(prefix).appendBuffer(transcoded);
+        break;
+      }
+      case NDJSON: {
+        chunk = Buffer.buffer(transcoded.length() + NEWLINE.length()).appendBuffer(transcoded).appendBuffer(NEWLINE);
+        break;
+      }
+    case SSE: {
+        chunk = Buffer.buffer(SSE_PREFIX.length() + transcoded.length() + SSE_SUFFIX.length())
+          .appendBuffer(SSE_PREFIX).appendBuffer(transcoded).appendBuffer(SSE_SUFFIX);
+        break;
+      }
+      default:
+        throw new AssertionError(streamFormat);
+    }
+    firstMessageWritten = true;
+    return httpResponse.write(chunk);
+  }
+
+  @Override
+  public Future<Void> writeEnd() {
+    if (streaming) {
+      if (!firstMessageWritten && status != null && status != GrpcStatus.OK) {
+        httpResponse.setStatusCode(GrpcTranscodingError.fromHttp2Code(status.code).getHttpStatusCode());
+        return httpResponse.end();
+      }
+      if (streamFormat == StreamFormat.JSON_ARRAY) {
+        return httpResponse.end(firstMessageWritten ? CLOSE_ARRAY : EMPTY_ARRAY);
+      }
+      return httpResponse.end();
+    }
+    if (status != GrpcStatus.OK) {
+      httpResponse.setStatusCode(GrpcTranscodingError.fromHttp2Code(status.code).getHttpStatusCode());
+    }
+    return super.writeEnd();
+  }
 }

vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/impl/TranscodingMessageDecoder.java

@@ -1,7 +1,7 @@
 package io.vertx.grpc.transcoding.impl;
 
-import io.vertx.core.buffer.Buffer;
 import io.vertx.core.json.DecodeException;
+import io.vertx.core.json.JsonObject;
 import io.vertx.grpc.common.CodecException;
 import io.vertx.grpc.common.GrpcMessage;
 import io.vertx.grpc.common.GrpcMessageDecoder;
@@ -26,13 +26,13 @@ public TranscodingMessageDecoder(GrpcMessageDecoder<Req> messageDecoder, WireFor
 
   @Override
   public Req decode(GrpcMessage msg) throws CodecException {
-    Buffer transcoded;
+    JsonObject transcoded;
     try {
       transcoded = MessageWeaver.weaveRequestMessage(msg.payload(), bindings, transcodingRequestBody, messageDecoder.messageDescriptor());
     } catch (DecodeException e) {
       throw new CodecException(e);
     }
-    return messageDecoder.decode(GrpcMessage.message("identity", format, transcoded));
+    return messageDecoder.decode(GrpcMessage.message("identity", format, transcoded.toBuffer()));
   }
   @Override
   public boolean accepts(WireFormat format) {