From 82cf15c60fb0276cc863f5642bf237b3c1a4710e Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Thu, 20 Aug 2026 14:47:52 +0100 Subject: [PATCH] ProtobufJsonEncoder actually supports streaming Closes gh-37158 --- .../codec/protobuf/ProtobufJsonEncoder.java | 59 ++++++++++++++----- .../protobuf/ProtobufJsonEncoderTests.java | 14 ++++- 2 files changed, 56 insertions(+), 17 deletions(-) diff --git a/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufJsonEncoder.java b/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufJsonEncoder.java index 4acc85b67f6..524f536562f 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufJsonEncoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufJsonEncoder.java @@ -54,6 +54,8 @@ import org.springframework.util.MimeType; */ public class ProtobufJsonEncoder implements HttpMessageEncoder { + private static final byte[] NEWLINE_SEPARATOR = {'\n'}; + private static final byte[] EMPTY_BYTES = new byte[0]; private static final ResolvableType MESSAGE_TYPE = ResolvableType.forClass(Message.class); @@ -117,22 +119,33 @@ public class ProtobufJsonEncoder implements HttpMessageEncoder { .map(value -> encodeValue(value, bufferFactory, elementType, mimeType, hints)) .flux(); } - JsonArrayJoinHelper helper = new JsonArrayJoinHelper(); - // Do not prepend JSON array prefix until first signal is known, onNext vs onError - // Keeps response not committed for error handling - return Flux.from(inputStream) - .map(value -> { - byte[] prefix = helper.getPrefix(); - byte[] delimiter = helper.getDelimiter(); - DataBuffer delimiterBuffer = bufferFactory.wrap(delimiter); - DataBuffer dataBuffer = encodeValue(value, bufferFactory, MESSAGE_TYPE, mimeType, hints); - return (prefix.length > 0 ? - bufferFactory.join(List.of(bufferFactory.wrap(prefix), delimiterBuffer, dataBuffer)) : - bufferFactory.join(List.of(delimiterBuffer, dataBuffer))); - }) - .switchIfEmpty(Mono.fromCallable(() -> bufferFactory.wrap(helper.getPrefix()))) - .concatWith(Mono.fromCallable(() -> bufferFactory.wrap(helper.getSuffix()))); + byte[] separator = getStreamingMediaTypeSeparator(mimeType); + if (separator != null) { + return Flux.from(inputStream) + .map(value -> { + DataBuffer dataBuffer = encodeValue(value, bufferFactory, MESSAGE_TYPE, mimeType, hints); + return bufferFactory.join(List.of(dataBuffer, bufferFactory.wrap(separator))); + }); + } + else { + JsonArrayJoinHelper helper = new JsonArrayJoinHelper(); + + // Do not prepend JSON array prefix until first signal is known, onNext vs onError + // Keeps response not committed for error handling + return Flux.from(inputStream) + .map(value -> { + byte[] prefix = helper.getPrefix(); + byte[] delimiter = helper.getDelimiter(); + DataBuffer delimiterBuffer = bufferFactory.wrap(delimiter); + DataBuffer dataBuffer = encodeValue(value, bufferFactory, MESSAGE_TYPE, mimeType, hints); + return (prefix.length > 0 ? + bufferFactory.join(List.of(bufferFactory.wrap(prefix), delimiterBuffer, dataBuffer)) : + bufferFactory.join(List.of(delimiterBuffer, dataBuffer))); + }) + .switchIfEmpty(Mono.fromCallable(() -> bufferFactory.wrap(helper.getPrefix()))) + .concatWith(Mono.fromCallable(() -> bufferFactory.wrap(helper.getSuffix()))); + } } @Override @@ -153,6 +166,22 @@ public class ProtobufJsonEncoder implements HttpMessageEncoder { } } + /** + * Return the separator to use for the given mime type. + *

By default, this method returns new line {@code "\n"} if the given + * mime type is one of the configured {@link #getStreamingMediaTypes + * streaming} mime types. + * @since 7.0.10 + */ + protected byte @Nullable [] getStreamingMediaTypeSeparator(@Nullable MimeType mimeType) { + for (MediaType streamingMediaType : getStreamingMediaTypes()) { + if (streamingMediaType.isCompatibleWith(mimeType)) { + return NEWLINE_SEPARATOR; + } + } + return null; + } + private static class JsonArrayJoinHelper { diff --git a/spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufJsonEncoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufJsonEncoderTests.java index 2924b0aa731..2b08b6cc2e0 100644 --- a/spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufJsonEncoderTests.java +++ b/spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufJsonEncoderTests.java @@ -92,7 +92,7 @@ class ProtobufJsonEncoderTests extends AbstractEncoderTests } @Test - void encodeStream() { + void encodeNonStream() { Flux input = Flux.just(this.msg1, this.msg2); ResolvableType inputType = forClass(Msg.class); @@ -104,7 +104,7 @@ class ProtobufJsonEncoderTests extends AbstractEncoderTests } @Test - void encodeEmptyFlux() { + void encodeNonStreamEmpty() { Flux input = Flux.empty(); ResolvableType inputType = forClass(Msg.class); Flux result = this.encoder.encode(input, this.bufferFactory, inputType, @@ -115,6 +115,16 @@ class ProtobufJsonEncoderTests extends AbstractEncoderTests .verifyComplete(); } + @Test + void encodeStream() { + Flux input = Flux.just(this.msg1, this.msg2); + ResolvableType inputType = forClass(Msg.class); + + testEncode(input, inputType, MediaType.APPLICATION_NDJSON, null, step -> step + .assertNext(buffer -> assertBufferEqualsJson(buffer, "{\"foo\":\"Foo\",\"blah\":{\"blah\":123}}\n")) + .assertNext(buffer -> assertBufferEqualsJson(buffer, "{\"foo\":\"Bar\",\"blah\":{\"blah\":456}}\n")) + .verifyComplete()); + } private void assertBufferEqualsJson(DataBuffer actual, String expected) { byte[] bytes = DataBufferTestUtils.dumpBytes(actual);