From 71c5440339c19a8c0fdd677b4058adb1a53a0b96 Mon Sep 17 00:00:00 2001 From: Daniel Fiala Date: Sun, 17 May 2026 19:09:02 +0200 Subject: [PATCH] Enable JSON transcoding for server-streaming RPCs and refine response formatting. Motivation: - Extend JSON transcoding support for server-streaming RPCs to enhance compatibility with streaming APIs. - Improve response flexibility with support for JSON Array, NDJSON, and SSE formats based on the `Accept` header. Changes: - Added `streaming` and `StreamFormat` fields to `TranscodingGrpcOutboundStream` for response format negotiation. - Implemented JSON Array, NDJSON, and SSE response formatting for server-streaming. - Updated `MessageWeaver` to handle `JsonObject` outputs for request/response weaving. - Modified tests to validate new streaming response formats and HTTP transcoding logic. - Refactored templates (`grpc-service.mustache`) and tests to support method-level `isManyOutput` configurations. Signed-off-by: Daniel Fiala --- .../src/main/asciidoc/transcoding.adoc | 34 +++ .../examples/grpc/GreeterGrpcService.java | 3 +- .../src/main/resources/grpc-service.mustache | 3 +- .../transcoding/TranscodingServiceMethod.java | 20 +- .../grpc/transcoding/impl/MessageWeaver.java | 37 ++-- .../impl/TranscodingGrpcOutboundStream.java | 131 +++++++++++- .../impl/TranscodingMessageDecoder.java | 6 +- .../impl/TranscodingServiceMethodImpl.java | 17 +- .../transcoding/MessageWeaverBenchmark.java | 12 +- .../tests/transcoding/MessageWeaverTest.java | 147 ++++++++----- .../transcoding/ServerTranscodingTest.java | 196 ++++++++++++++++++ 11 files changed, 514 insertions(+), 92 deletions(-) diff --git a/vertx-grpc-docs/src/main/asciidoc/transcoding.adoc b/vertx-grpc-docs/src/main/asciidoc/transcoding.adoc index 1ef302817..bb5a433b3 100644 --- a/vertx-grpc-docs/src/main/asciidoc/transcoding.adoc +++ b/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: \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. diff --git a/vertx-grpc-docs/src/main/java/examples/grpc/GreeterGrpcService.java b/vertx-grpc-docs/src/main/java/examples/grpc/GreeterGrpcService.java index 5dc83dd7a..ce09d1ad6 100644 --- a/vertx-grpc-docs/src/main/java/examples/grpc/GreeterGrpcService.java +++ b/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()); diff --git a/vertx-grpc-protoc-plugin2/src/main/resources/grpc-service.mustache b/vertx-grpc-protoc-plugin2/src/main/resources/grpc-service.mustache index 203e2f6d5..9b6b28fd0 100644 --- a/vertx-grpc-protoc-plugin2/src/main/resources/grpc-service.mustache +++ b/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}} diff --git a/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/TranscodingServiceMethod.java b/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/TranscodingServiceMethod.java index 5cac1ce66..cd6a34bae 100644 --- a/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/TranscodingServiceMethod.java +++ b/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/TranscodingServiceMethod.java @@ -24,7 +24,16 @@ static TranscodingServiceMethod server(ServiceName servic GrpcMessageEncoder encoder, GrpcMessageDecoder decoder, MethodTranscodingOptions options) { - return new TranscodingServiceMethodImpl<>(serviceName, methodName, encoder, decoder, options); + return new TranscodingServiceMethodImpl<>(serviceName, methodName, encoder, decoder, options, false); + } + + static TranscodingServiceMethod server(ServiceName serviceName, + String methodName, + GrpcMessageEncoder encoder, + GrpcMessageDecoder decoder, + MethodTranscodingOptions options, + boolean streaming) { + return new TranscodingServiceMethodImpl<>(serviceName, methodName, encoder, decoder, options, streaming); } static TranscodingServiceMethod server(ServiceMethod serviceMethod, @@ -34,4 +43,13 @@ static TranscodingServiceMethod server(ServiceMethod bindings, String transcodingRequestBody, Descriptors.Descriptor descriptor) throws DecodeException { + public static JsonObject weaveRequestMessage(Buffer message, List 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\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 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 writeEnd() { - if (status != GrpcStatus.OK) { - httpResponse.setStatusCode(GrpcTranscodingError.fromHttp2Code(status.code).getHttpStatusCode()); + protected Future 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 writeHead() { + if (streaming) { + return httpResponse.writeHead(); + } head = context.promise(); return head.future(); } @@ -68,9 +125,16 @@ public Future writeMessage(GrpcMessageFrame frame) { } catch (CodecException e) { return context.failedFuture(e); } + if (streaming) { + return writeStreamingMessage(payload); + } + return writeUnaryMessage(payload); + } + + private Future writeUnaryMessage(Buffer payload) { Future 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 writeMessage(GrpcMessageFrame frame) { } return res; } + + private Future 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 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(); + } } diff --git a/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/impl/TranscodingMessageDecoder.java b/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/impl/TranscodingMessageDecoder.java index ca4b5d8ea..0da3fd76d 100644 --- a/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/impl/TranscodingMessageDecoder.java +++ b/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 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) { diff --git a/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/impl/TranscodingServiceMethodImpl.java b/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/impl/TranscodingServiceMethodImpl.java index 38ebbc7c5..04e6058d4 100644 --- a/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/impl/TranscodingServiceMethodImpl.java +++ b/vertx-grpc-transcoding/src/main/java/io/vertx/grpc/transcoding/impl/TranscodingServiceMethodImpl.java @@ -27,19 +27,25 @@ public class TranscodingServiceMethodImpl implements TranscodingServiceMet private final GrpcMessageEncoder encoder; private final GrpcMessageDecoder decoder; private final MethodTranscodingOptions options; + private final boolean streaming; private final PathMatcher pathMatcher; public TranscodingServiceMethodImpl(ServiceName serviceName, String methodName, GrpcMessageEncoder encoder, GrpcMessageDecoder decoder) { - this(serviceName, methodName, encoder, decoder, null); + this(serviceName, methodName, encoder, decoder, null, false); } public TranscodingServiceMethodImpl(ServiceName serviceName, String methodName, GrpcMessageEncoder encoder, GrpcMessageDecoder decoder, MethodTranscodingOptions options) { + this(serviceName, methodName, encoder, decoder, options, false); + } + + public TranscodingServiceMethodImpl(ServiceName serviceName, String methodName, GrpcMessageEncoder encoder, GrpcMessageDecoder decoder, MethodTranscodingOptions options, boolean streaming) { this.serviceName = serviceName; this.methodName = methodName; this.encoder = encoder; this.decoder = decoder; this.options = options; + this.streaming = streaming; // Init if (options != null) { @@ -96,12 +102,12 @@ public GrpcInvocation accept(HttpServerRequest httpRequest, WireFormat format) { io.vertx.core.internal.ContextInternal context = ((HttpServerRequestInternal) httpRequest).context(); TranscodingMessageDecoder messageDecoder = new TranscodingMessageDecoder<>(decoder, format, res.getBodyFieldPath(), bindings); TranscodingMessageDeframer deframer = new TranscodingMessageDeframer(format); - HttpGrpcOutboundStream protocolHandler = new TranscodingGrpcOutboundStream(context, httpRequest, options.getResponseBody(), deframer); + HttpGrpcOutboundStream protocolHandler = new TranscodingGrpcOutboundStream(context, httpRequest, options.getResponseBody(), deframer, streaming); return new GrpcInvocation(deframer, protocolHandler, messageDecoder); } else if (options == null) { io.vertx.core.internal.ContextInternal context = ((HttpServerRequestInternal) httpRequest).context(); TranscodingMessageDeframer deframer = new TranscodingMessageDeframer(format); - HttpGrpcOutboundStream protocolHandler = new TranscodingGrpcOutboundStream(context, httpRequest, null, deframer); + HttpGrpcOutboundStream protocolHandler = new TranscodingGrpcOutboundStream(context, httpRequest, null, deframer, streaming); return new GrpcInvocation(deframer, protocolHandler, decoder); } @@ -132,4 +138,9 @@ public GrpcMessageEncoder encoder() { public MethodTranscodingOptions options() { return options; } + + @Override + public boolean streaming() { + return streaming; + } } diff --git a/vertx-grpc-transcoding/src/test/java/io/vertx/benchmarks/transcoding/MessageWeaverBenchmark.java b/vertx-grpc-transcoding/src/test/java/io/vertx/benchmarks/transcoding/MessageWeaverBenchmark.java index 135afce91..18f781609 100644 --- a/vertx-grpc-transcoding/src/test/java/io/vertx/benchmarks/transcoding/MessageWeaverBenchmark.java +++ b/vertx-grpc-transcoding/src/test/java/io/vertx/benchmarks/transcoding/MessageWeaverBenchmark.java @@ -96,7 +96,7 @@ private HttpVariableBinding createBinding(String path, String value) { @Benchmark public void benchmarkWeaveRequestSimpleMessageNoBindings(Blackhole blackhole) { - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( simpleMessage, null, null, @@ -107,7 +107,7 @@ public void benchmarkWeaveRequestSimpleMessageNoBindings(Blackhole blackhole) { @Benchmark public void benchmarkWeaveRequestSimpleMessageWithBindings(Blackhole blackhole) { - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( simpleMessage, simpleBindings, null, @@ -118,7 +118,7 @@ public void benchmarkWeaveRequestSimpleMessageWithBindings(Blackhole blackhole) @Benchmark public void benchmarkWeaveRequestComplexMessageWithBindings(Blackhole blackhole) { - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( complexMessage, complexBindings, null, @@ -129,7 +129,7 @@ public void benchmarkWeaveRequestComplexMessageWithBindings(Blackhole blackhole) @Benchmark public void benchmarkWeaveRequestWithTranscodingPath(Blackhole blackhole) { - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( complexMessage, complexBindings, complexTranscodingPath, @@ -140,7 +140,7 @@ public void benchmarkWeaveRequestWithTranscodingPath(Blackhole blackhole) { @Benchmark public void benchmarkWeaveResponseSimpleMessage(Blackhole blackhole) { - Buffer result = MessageWeaver.weaveResponseMessage( + Object result = MessageWeaver.weaveResponseMessage( simpleMessage, simpleTranscodingPath ); @@ -149,7 +149,7 @@ public void benchmarkWeaveResponseSimpleMessage(Blackhole blackhole) { @Benchmark public void benchmarkWeaveResponseComplexMessage(Blackhole blackhole) { - Buffer result = MessageWeaver.weaveResponseMessage( + Object result = MessageWeaver.weaveResponseMessage( complexMessage, complexTranscodingPath ); diff --git a/vertx-grpc-transcoding/src/test/java/io/vertx/tests/transcoding/MessageWeaverTest.java b/vertx-grpc-transcoding/src/test/java/io/vertx/tests/transcoding/MessageWeaverTest.java index 0796ee142..f37f8b006 100644 --- a/vertx-grpc-transcoding/src/test/java/io/vertx/tests/transcoding/MessageWeaverTest.java +++ b/vertx-grpc-transcoding/src/test/java/io/vertx/tests/transcoding/MessageWeaverTest.java @@ -82,14 +82,14 @@ public void testPassThroughRequest() { .put("B", new JsonObject() .put("y", "b"))); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(message.encode()), new ArrayList<>(), "*", TEST_DESCRIPTOR ); - assertEquals(message, new JsonObject(result.toString())); + assertEquals(message, result); } @Test @@ -109,14 +109,14 @@ public void testLevel0Bindings() { .put("_y", "b") .put("_z", "c"); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(original.encode()), bindings, "*", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -140,14 +140,14 @@ public void testLevel1Bindings() { .put("z", "f") .put("_x", "c")); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(original.encode()), bindings, "*", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -179,14 +179,14 @@ public void testLevel2Bindings() { .put("u", "g") .put("_x", "c"))); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(original.encode()), bindings, "*", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -195,14 +195,14 @@ public void testTranscodingRequestBodyWildcard() { .put("field1", "value1") .put("field2", "value2"); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(message.encode()), new ArrayList<>(), "*", TEST_DESCRIPTOR ); - assertEquals(message, new JsonObject(result.toString())); + assertEquals(message, result); } @Test @@ -213,14 +213,14 @@ public void testTranscodingRequestBodyPath() { JsonObject expected = new JsonObject().put("nested", new JsonObject().put("data", message)); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(message.encode()), new ArrayList<>(), "nested.data", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -229,12 +229,12 @@ public void testResponsePassThrough() { .put("field1", "value1") .put("field2", "value2"); - Buffer result = MessageWeaver.weaveResponseMessage( + Object result = MessageWeaver.weaveResponseMessage( Buffer.buffer(message.encode()), null ); - assertEquals(message, new JsonObject(result.toString())); + assertEquals(message, result); } @Test @@ -243,12 +243,12 @@ public void testResponseBodyWildcard() { .put("field1", "value1") .put("field2", "value2"); - Buffer result = MessageWeaver.weaveResponseMessage( + Object result = MessageWeaver.weaveResponseMessage( Buffer.buffer(message.encode()), "*" ); - assertEquals(message, new JsonObject(result.toString())); + assertEquals(message, result); } @Test @@ -261,22 +261,67 @@ public void testResponseBodyPath() { .put("response", new JsonObject() .put("data", nested)); - Buffer result = MessageWeaver.weaveResponseMessage( + Object result = MessageWeaver.weaveResponseMessage( Buffer.buffer(message.encode()), "response.data" ); - assertEquals(nested, new JsonObject(result.toString())); + assertEquals(nested, result); + } + + @Test + public void testResponseBodyRepeatedField() { + // When response_body points to a repeated proto field, the selected value is a JSON array. + JsonArray users = new JsonArray() + .add(new JsonObject().put("id", 1).put("name", "alice")) + .add(new JsonObject().put("id", 2).put("name", "bob")); + + JsonObject message = new JsonObject().put("users", users); + + Object result = MessageWeaver.weaveResponseMessage( + Buffer.buffer(message.encode()), + "users" + ); + + assertEquals(users, result); + } + + @Test + public void testResponseBodyScalarField() { + // response_body pointing to a scalar leaf, e.g. an RPC returning just a string field as the body. + JsonObject message = new JsonObject() + .put("response", new JsonObject().put("token", "abc123")); + + Object result = MessageWeaver.weaveResponseMessage( + Buffer.buffer(message.encode()), + "response.token" + ); + + assertEquals("abc123", result); + } + + @Test + public void testResponseBodyNullLeaf() { + // Field absent or explicitly null yields null, doesn't throw. + JsonObject message = new JsonObject().put("response", new JsonObject().putNull("token")); + + Object result = MessageWeaver.weaveResponseMessage( + Buffer.buffer(message.encode()), + "response.token" + ); + + assertEquals(null, result); } @Test public void testResponseBodyInvalidPath() { + // Path attempts to descend into a scalar, should throw. JsonObject message = new JsonObject() .put("field1", "value1"); assertThrows(IllegalArgumentException.class, () -> MessageWeaver.weaveResponseMessage( Buffer.buffer(message.encode()), - "invalid.path" + "field1.invalid" )); } @@ -289,14 +334,14 @@ public void testRepeatedBindingsLevel0() { JsonObject expected = new JsonObject() .put("x", new JsonArray().add("a").add("b").add("c")); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( null, bindings, null, TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -309,14 +354,14 @@ public void testRepeatedBindingsLevel1() { .put("A", new JsonObject() .put("x", new JsonArray().add("b").add("c").add("d"))); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( null, bindings, null, TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -333,14 +378,14 @@ public void testRepeatedBindingsWithBody() { .put("x", new JsonArray().add("b").add("c").add("d")) .put("y", "e")); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(body.encode()), bindings, "*", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -353,14 +398,14 @@ public void testRepeatedBindingsMixedWithSingle() { .put("x", new JsonArray().add("a").add("b")) .put("y", "c"); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( null, bindings, null, TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -371,38 +416,38 @@ public void testRepeatedBindingsTwoValues() { JsonObject expected = new JsonObject() .put("keys", new JsonArray().add("A").add("B")); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( null, bindings, null, TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test public void testEmptyMessageReturnsEmptyObject() { - Buffer fromNull = MessageWeaver.weaveRequestMessage(null, new ArrayList<>(), null, TEST_DESCRIPTOR); - assertEquals(new JsonObject(), new JsonObject(fromNull.toString())); + JsonObject fromNull = MessageWeaver.weaveRequestMessage(null, new ArrayList<>(), null, TEST_DESCRIPTOR); + assertEquals(new JsonObject(), fromNull); - Buffer fromEmpty = MessageWeaver.weaveRequestMessage(Buffer.buffer(), new ArrayList<>(), null, TEST_DESCRIPTOR); - assertEquals(new JsonObject(), new JsonObject(fromEmpty.toString())); + JsonObject fromEmpty = MessageWeaver.weaveRequestMessage(Buffer.buffer(), new ArrayList<>(), null, TEST_DESCRIPTOR); + assertEquals(new JsonObject(), fromEmpty); - Buffer fromEmptyBody = MessageWeaver.weaveRequestMessage(Buffer.buffer(), new ArrayList<>(), "", TEST_DESCRIPTOR); - assertEquals(new JsonObject(), new JsonObject(fromEmptyBody.toString())); + JsonObject fromEmptyBody = MessageWeaver.weaveRequestMessage(Buffer.buffer(), new ArrayList<>(), "", TEST_DESCRIPTOR); + assertEquals(new JsonObject(), fromEmptyBody); } @Test public void testUnsetBodyIgnoresHttpBody() { JsonObject httpBody = new JsonObject().put("field1", "value1"); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(httpBody.encode()), new ArrayList<>(), null, TEST_DESCRIPTOR ); - assertEquals(new JsonObject(), new JsonObject(result.toString())); + assertEquals(new JsonObject(), result); } @Test @@ -417,14 +462,14 @@ public void testBindingOverridesBodyScalar() { .put("y", "from-binding") .put("A", new JsonObject().put("y", "nested-body")); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(body.encode()), bindings, "*", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -437,14 +482,14 @@ public void testBindingOverridesNestedBodyScalar() { JsonObject expected = new JsonObject() .put("A", new JsonObject().put("y", "from-binding").put("x", new JsonArray().add("k"))); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(body.encode()), bindings, "*", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -457,14 +502,14 @@ public void testBindingCreatesSubtreeAbsentFromBody() { .put("y", "root-body") .put("A", new JsonObject().put("y", "from-binding")); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(body.encode()), bindings, "*", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -477,14 +522,14 @@ public void testRepeatedBindingAppendedToBodyValues() { JsonObject expected = new JsonObject() .put("x", new JsonArray().add("a").add("b").add("c")); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(body.encode()), bindings, "*", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -498,14 +543,14 @@ public void testBindingAndBodyAtSpecificPath() { .put("x", new JsonArray().add("b")) .put("y", "from-binding")); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(body.encode()), bindings, "A", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -521,14 +566,14 @@ public void testBindingsDoNotDescendIntoArrayElements() { .add(new JsonObject().put("A", new JsonObject().put("x", "in-array")))) .put("A", new JsonObject().put("y", "from-binding")); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(body.encode()), bindings, "*", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test @@ -540,14 +585,14 @@ public void testDeepBodyPath() { .put("b", new JsonObject() .put("c", body))); - Buffer result = MessageWeaver.weaveRequestMessage( + JsonObject result = MessageWeaver.weaveRequestMessage( Buffer.buffer(body.encode()), new ArrayList<>(), "a.b.c", TEST_DESCRIPTOR ); - assertEquals(expected, new JsonObject(result.toString())); + assertEquals(expected, result); } @Test diff --git a/vertx-grpc-transcoding/src/test/java/io/vertx/tests/transcoding/ServerTranscodingTest.java b/vertx-grpc-transcoding/src/test/java/io/vertx/tests/transcoding/ServerTranscodingTest.java index 362b9a869..a63db9a16 100644 --- a/vertx-grpc-transcoding/src/test/java/io/vertx/tests/transcoding/ServerTranscodingTest.java +++ b/vertx-grpc-transcoding/src/test/java/io/vertx/tests/transcoding/ServerTranscodingTest.java @@ -20,6 +20,7 @@ import io.vertx.core.buffer.Buffer; import io.vertx.core.http.*; import io.vertx.core.internal.buffer.BufferInternal; +import io.vertx.core.json.JsonArray; import io.vertx.core.json.JsonObject; import io.vertx.ext.unit.TestContext; import io.vertx.grpc.common.*; @@ -59,6 +60,8 @@ public static MethodTranscodingOptions create(String selector, HttpMethod httpMe public static GrpcMessageDecoder ECHO_REQUEST_BODY_DECODER = GrpcMessageDecoder.decoder(EchoRequestBody.newBuilder()); public static GrpcMessageEncoder ECHO_RESPONSE_ENCODER = GrpcMessageEncoder.encoder(); public static GrpcMessageEncoder ECHO_RESPONSE_BODY_ENCODER = GrpcMessageEncoder.encoder(); + public static GrpcMessageDecoder STREAMING_REQUEST_DECODER = GrpcMessageDecoder.decoder(StreamingRequest.newBuilder()); + public static GrpcMessageEncoder STREAMING_RESPONSE_ENCODER = GrpcMessageEncoder.encoder(); public static final ServiceName TEST_SERVICE_NAME = ServiceName.create(TestServiceGrpc.SERVICE_NAME); @@ -91,6 +94,17 @@ public static MethodTranscodingOptions create(String selector, HttpMethod httpMe public static final TranscodingServiceMethod UNARY_CALL_WITH_REPEATED_QUERY = TranscodingServiceMethod.server(TEST_SERVICE_NAME, "UnaryCallWithRepeatedQuery", ECHO_RESPONSE_ENCODER, ECHO_REQUEST_DECODER, UNARY_TRANSCODING_WITH_REPEATED_QUERY); + public static final MethodTranscodingOptions STREAMING_TRANSCODING = new MethodTranscodingOptions().setHttpMethod(HttpMethod.POST).setPath("/stream").setBody("*"); + public static final TranscodingServiceMethod STREAMING_CALL = TranscodingServiceMethod.server(TEST_SERVICE_NAME, "StreamingCall", + STREAMING_RESPONSE_ENCODER, STREAMING_REQUEST_DECODER, STREAMING_TRANSCODING, true); + + // Unary RPC whose HttpRule selects a repeated field as the response body. The HTTP body should be the unwrapped JSON + // array of the field, not the surrounding message object. + public static final MethodTranscodingOptions UNARY_TRANSCODING_WITH_ARRAY_RESPONSE_BODY = new MethodTranscodingOptions().setHttpMethod(HttpMethod.POST).setPath("/keylist").setBody("*").setResponseBody("keys"); + public static GrpcMessageEncoder ECHO_REQUEST_ENCODER = GrpcMessageEncoder.encoder(); + public static final TranscodingServiceMethod UNARY_CALL_WITH_ARRAY_RESPONSE_BODY = TranscodingServiceMethod.server(TEST_SERVICE_NAME, "UnaryCallWithArrayResponseBody", + ECHO_REQUEST_ENCODER, ECHO_REQUEST_DECODER, UNARY_TRANSCODING_WITH_ARRAY_RESPONSE_BODY); + private static final CharSequence USER_AGENT = HttpHeaders.createOptimized("X-User-Agent"); private static final String CONTENT_TYPE = "application/json"; @@ -239,6 +253,36 @@ public void setUp(TestContext should) { response.end(responseMsg); }); }); + grpcServer.callHandler(UNARY_CALL_WITH_ARRAY_RESPONSE_BODY, request -> { + request.handler(requestMsg -> { + GrpcServerResponse response = request.response(); + EchoRequest responseMsg = EchoRequest.newBuilder() + .addKeys("alice") + .addKeys("bob") + .addKeys("carol") + .build(); + response.end(responseMsg); + }); + }); + grpcServer.callHandler(STREAMING_CALL, request -> { + request.handler(requestMsg -> { + GrpcServerResponse response = request.response(); + if (requestMsg.getResponseSizeList().isEmpty()) { + response.end(); + return; + } + for (int size : requestMsg.getResponseSizeList()) { + if (size < 0) { + response.status(GrpcStatus.INVALID_ARGUMENT).end(); + return; + } + char[] value = new char[size]; + Arrays.fill(value, 'a'); + response.write(StreamingResponse.newBuilder().setPayload(new String(value)).build()); + } + response.end(); + }); + }); httpServer = vertx.createHttpServer(new HttpServerOptions().setPort(port)).requestHandler(grpcServer); httpServer.listen().onComplete(should.asyncAssertSuccess()); } @@ -534,6 +578,158 @@ public void testRepeatedQueryParamsSingleValue(TestContext should) { }))); } + @Test + public void testServerStreaming(TestContext should) { + httpClient.request(HttpMethod.POST, "/stream").compose(req -> { + String body = encode(StreamingRequest.newBuilder().addResponseSize(1).addResponseSize(2).addResponseSize(3).build()).toString(); + req.headers().addAll(HEADERS); + return req.send(body).compose(response -> response.body().map(response)); + }).onComplete(should.asyncAssertSuccess(response -> should.verify(v -> { + assertEquals(200, response.statusCode()); + MultiMap headers = response.headers(); + assertTrue(headers.contains(HttpHeaders.CONTENT_TYPE, CONTENT_TYPE, true)); + // Chunked transfer means no Content-Length header. + assertFalse(headers.contains(HttpHeaders.CONTENT_LENGTH)); + JsonArray array = new JsonArray(response.body().result()); + assertEquals(3, array.size()); + assertEquals("a", array.getJsonObject(0).getString("payload")); + assertEquals("aa", array.getJsonObject(1).getString("payload")); + assertEquals("aaa", array.getJsonObject(2).getString("payload")); + }))); + } + + @Test + public void testServerStreamingEmpty(TestContext should) { + httpClient.request(HttpMethod.POST, "/stream").compose(req -> { + String body = encode(StreamingRequest.newBuilder().build()).toString(); + req.headers().addAll(HEADERS); + return req.send(body).compose(response -> response.body().map(response)); + }).onComplete(should.asyncAssertSuccess(response -> should.verify(v -> { + assertEquals(200, response.statusCode()); + JsonArray array = new JsonArray(response.body().result()); + assertEquals(0, array.size()); + }))); + } + + @Test + public void testServerStreamingErrorBeforeFirstMessage(TestContext should) { + httpClient.request(HttpMethod.POST, "/stream").compose(req -> { + String body = encode(StreamingRequest.newBuilder().addResponseSize(-1).build()).toString(); + req.headers().addAll(HEADERS); + return req.send(body).compose(response -> response.body().map(response)); + }).onComplete(should.asyncAssertSuccess(response -> should.verify(v -> { + assertEquals(400, response.statusCode()); + }))); + } + + @Test + public void testServerStreamingNdjson(TestContext should) { + httpClient.request(HttpMethod.POST, "/stream").compose(req -> { + String body = encode(StreamingRequest.newBuilder().addResponseSize(1).addResponseSize(2).addResponseSize(3).build()).toString(); + req.headers().addAll(HEADERS); + req.headers().set(HttpHeaders.ACCEPT, "application/x-ndjson"); + return req.send(body).compose(response -> response.body().map(response)); + }).onComplete(should.asyncAssertSuccess(response -> should.verify(v -> { + assertEquals(200, response.statusCode()); + assertTrue(response.headers().contains(HttpHeaders.CONTENT_TYPE, "application/x-ndjson", true)); + String[] lines = response.body().result().toString().split("\n"); + assertEquals(3, lines.length); + assertEquals("a", new JsonObject(lines[0]).getString("payload")); + assertEquals("aa", new JsonObject(lines[1]).getString("payload")); + assertEquals("aaa", new JsonObject(lines[2]).getString("payload")); + }))); + } + + @Test + public void testServerStreamingSse(TestContext should) { + httpClient.request(HttpMethod.POST, "/stream").compose(req -> { + String body = encode(StreamingRequest.newBuilder().addResponseSize(1).addResponseSize(2).addResponseSize(3).build()).toString(); + req.headers().addAll(HEADERS); + req.headers().set(HttpHeaders.ACCEPT, "text/event-stream"); + return req.send(body).compose(response -> response.body().map(response)); + }).onComplete(should.asyncAssertSuccess(response -> should.verify(v -> { + assertEquals(200, response.statusCode()); + assertTrue(response.headers().contains(HttpHeaders.CONTENT_TYPE, "text/event-stream", true)); + String[] events = response.body().result().toString().split("\n\n"); + assertEquals(3, events.length); + for (int i = 0; i < events.length; i++) { + assertTrue("event[" + i + "] should start with 'data: ': " + events[i], events[i].startsWith("data: ")); + String json = events[i].substring("data: ".length()); + assertEquals("a".repeat(i + 1), new JsonObject(json).getString("payload")); + } + }))); + } + + @Test + public void testServerStreamingErrorMidStream(TestContext should) { + // Server writes one message, then fails with INVALID_ARGUMENT. Headers have already been flushed with HTTP 200, + // so the gRPC error cannot change the status code. The JSON array is just closed and the body ends. + httpClient.request(HttpMethod.POST, "/stream").compose(req -> { + String body = encode(StreamingRequest.newBuilder().addResponseSize(1).addResponseSize(-1).build()).toString(); + req.headers().addAll(HEADERS); + return req.send(body).compose(response -> response.body().map(response)); + }).onComplete(should.asyncAssertSuccess(response -> should.verify(v -> { + assertEquals(200, response.statusCode()); + JsonArray array = new JsonArray(response.body().result()); + assertEquals(1, array.size()); + assertEquals("a", array.getJsonObject(0).getString("payload")); + }))); + } + + @Test + public void testServerStreamingUnknownAcceptFallsBackToJsonArray(TestContext should) { + httpClient.request(HttpMethod.POST, "/stream").compose(req -> { + String body = encode(StreamingRequest.newBuilder().addResponseSize(1).build()).toString(); + req.headers().addAll(HEADERS); + req.headers().set(HttpHeaders.ACCEPT, "application/xml"); + return req.send(body).compose(response -> response.body().map(response)); + }).onComplete(should.asyncAssertSuccess(response -> should.verify(v -> { + assertEquals(200, response.statusCode()); + assertTrue(response.headers().contains(HttpHeaders.CONTENT_TYPE, CONTENT_TYPE, true)); + JsonArray array = new JsonArray(response.body().result()); + assertEquals(1, array.size()); + }))); + } + + @Test + public void testServerStreamingWildcardAcceptFallsBackToJsonArray(TestContext should) { + httpClient.request(HttpMethod.POST, "/stream").compose(req -> { + String body = encode(StreamingRequest.newBuilder().addResponseSize(1).build()).toString(); + req.headers().addAll(HEADERS); + req.headers().set(HttpHeaders.ACCEPT, "*/*"); + return req.send(body).compose(response -> response.body().map(response)); + }).onComplete(should.asyncAssertSuccess(response -> should.verify(v -> { + assertEquals(200, response.statusCode()); + assertTrue(response.headers().contains(HttpHeaders.CONTENT_TYPE, CONTENT_TYPE, true)); + JsonArray array = new JsonArray(response.body().result()); + assertEquals(1, array.size()); + }))); + } + + @Test + public void testResponseBodyRepeatedFieldUnwrapped(TestContext should) { + httpClient.request(HttpMethod.POST, "/keylist").compose(req -> { + String body = encode(EchoRequest.newBuilder().build()).toString(); + req.headers().addAll(HEADERS); + return req.send(body).compose(response -> response.body().map(response)); + }).onComplete(should.asyncAssertSuccess(response -> should.verify(v -> { + assertEquals(200, response.statusCode()); + assertTrue(response.headers().contains(HttpHeaders.CONTENT_TYPE, CONTENT_TYPE, true)); + JsonArray expected = new JsonArray().add("alice").add("bob").add("carol"); + assertEquals(expected, new JsonArray(response.body().result())); + }))); + } + + @Test + public void testServerStreamingMalformedBody(TestContext should) { + httpClient.request(HttpMethod.POST, "/stream").compose(req -> { + req.headers().addAll(HEADERS); + return req.send("not-json").compose(response -> response.body().map(response)); + }).onComplete(should.asyncAssertSuccess(response -> should.verify(v -> { + assertEquals(400, response.statusCode()); + }))); + } + @Test public void testHttp2Proto() { ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", port)