Skip to content

Commit ce56846

Browse files
rstoyanchevrstoyanchev
committed
Refine JSON encoding of non-streaming Flux
Closes gh-28398
1 parent 496c1dc commit ce56846

File tree

2 files changed

+109
-58
lines changed

2 files changed

+109
-58
lines changed

Diff for: ‎spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java

+90-39
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import com.fasterxml.jackson.databind.SequenceWriter;
3636
import com.fasterxml.jackson.databind.exc.InvalidDefinitionException;
3737
import com.fasterxml.jackson.databind.ser.FilterProvider;
38+
import org.apache.commons.logging.Log;
3839
import org.reactivestreams.Publisher;
3940
import reactor.core.publisher.Flux;
4041
import reactor.core.publisher.Mono;
@@ -70,6 +71,8 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
7071

7172
private static final byte[] NEWLINE_SEPARATOR = {'\n'};
7273

74+
private static final byte[] EMPTY_BYTES = new byte[0];
75+
7376
private static final Map<String, JsonEncoding> ENCODINGS;
7477

7578
static {
@@ -150,45 +153,47 @@ public Flux<DataBuffer> encode(Publisher<?> inputStream, DataBufferFactory buffe
150153
.map(value -> encodeValue(value, bufferFactory, elementType, mimeType, hints))
151154
.flux();
152155
}
153-
else {
156+
157+
try {
158+
ObjectMapper mapper = selectObjectMapper(elementType, mimeType);
159+
if (mapper == null) {
160+
throw new IllegalStateException("No ObjectMapper for " + elementType);
161+
}
162+
ObjectWriter writer = createObjectWriter(mapper, elementType, mimeType, null, hints);
163+
ByteArrayBuilder byteBuilder = new ByteArrayBuilder(writer.getFactory()._getBufferRecycler());
164+
JsonEncoding encoding = getJsonEncoding(mimeType);
165+
JsonGenerator generator = mapper.getFactory().createGenerator(byteBuilder, encoding);
166+
SequenceWriter sequenceWriter = writer.writeValues(generator);
167+
154168
byte[] separator = getStreamingMediaTypeSeparator(mimeType);
155-
if (separator != null) { // streaming
156-
try {
157-
ObjectMapper mapper = selectObjectMapper(elementType, mimeType);
158-
if (mapper == null) {
159-
throw new IllegalStateException("No ObjectMapper for " + elementType);
160-
}
161-
ObjectWriter writer = createObjectWriter(mapper, elementType, mimeType, null, hints);
162-
ByteArrayBuilder byteBuilder = new ByteArrayBuilder(writer.getFactory()._getBufferRecycler());
163-
JsonEncoding encoding = getJsonEncoding(mimeType);
164-
JsonGenerator generator = mapper.getFactory().createGenerator(byteBuilder, encoding);
165-
SequenceWriter sequenceWriter = writer.writeValues(generator);
166-
167-
return Flux.from(inputStream)
168-
.map(value -> encodeStreamingValue(value, bufferFactory, hints, sequenceWriter, byteBuilder,
169-
separator))
170-
.doAfterTerminate(() -> {
171-
try {
172-
byteBuilder.release();
173-
generator.close();
174-
}
175-
catch (IOException ex) {
176-
logger.error("Could not close Encoder resources", ex);
177-
}
178-
});
179-
}
180-
catch (IOException ex) {
181-
return Flux.error(ex);
182-
}
169+
Flux<DataBuffer> dataBufferFlux;
170+
171+
if (separator != null) {
172+
dataBufferFlux = Flux.from(inputStream).map(value -> encodeStreamingValue(
173+
value, bufferFactory, hints, sequenceWriter, byteBuilder, EMPTY_BYTES, separator));
183174
}
184-
else { // non-streaming
185-
ResolvableType listType = ResolvableType.forClassWithGenerics(List.class, elementType);
186-
return Flux.from(inputStream)
187-
.collectList()
188-
.map(list -> encodeValue(list, bufferFactory, listType, mimeType, hints))
189-
.flux();
175+
else {
176+
JsonArrayJoinHelper helper = new JsonArrayJoinHelper();
177+
return Flux.concat(
178+
helper.getPrefix(bufferFactory, hints, logger),
179+
Flux.from(inputStream).map(value -> encodeStreamingValue(
180+
value, bufferFactory, hints, sequenceWriter, byteBuilder, helper.getDelimiter(), EMPTY_BYTES)),
181+
helper.getSuffix(bufferFactory, hints, logger));
190182
}
191183

184+
return dataBufferFlux
185+
.doAfterTerminate(() -> {
186+
try {
187+
byteBuilder.release();
188+
generator.close();
189+
}
190+
catch (IOException ex) {
191+
logger.error("Could not close Encoder resources", ex);
192+
}
193+
});
194+
}
195+
catch (IOException ex) {
196+
return Flux.error(ex);
192197
}
193198
}
194199

@@ -247,8 +252,10 @@ public DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory,
247252
}
248253
}
249254

250-
private DataBuffer encodeStreamingValue(Object value, DataBufferFactory bufferFactory, @Nullable Map<String, Object> hints,
251-
SequenceWriter sequenceWriter, ByteArrayBuilder byteArrayBuilder, byte[] separator) {
255+
private DataBuffer encodeStreamingValue(
256+
Object value, DataBufferFactory bufferFactory, @Nullable Map<String, Object> hints,
257+
SequenceWriter sequenceWriter, ByteArrayBuilder byteArrayBuilder,
258+
byte[] prefix, byte[] suffix) {
252259

253260
logValue(hints, value);
254261

@@ -280,9 +287,14 @@ private DataBuffer encodeStreamingValue(Object value, DataBufferFactory bufferFa
280287
offset = 0;
281288
length = bytes.length;
282289
}
283-
DataBuffer buffer = bufferFactory.allocateBuffer(length + separator.length);
290+
DataBuffer buffer = bufferFactory.allocateBuffer(length + prefix.length + suffix.length);
291+
if (prefix.length != 0) {
292+
buffer.write(prefix);
293+
}
284294
buffer.write(bytes, offset, length);
285-
buffer.write(separator);
295+
if (suffix.length != 0) {
296+
buffer.write(suffix);
297+
}
286298
Hints.touchDataBuffer(buffer, hints, logger);
287299

288300
return buffer;
@@ -385,4 +397,43 @@ protected <A extends Annotation> A getAnnotation(MethodParameter parameter, Clas
385397
return parameter.getMethodAnnotation(annotType);
386398
}
387399

400+
401+
private static class JsonArrayJoinHelper {
402+
403+
private static final byte[] COMMA_SEPARATOR = {','};
404+
405+
private static final byte[] OPEN_BRACKET = {'['};
406+
407+
private static final byte[] CLOSE_BRACKET = {']'};
408+
409+
410+
private boolean afterFirstItem = false;
411+
412+
public byte[] getDelimiter() {
413+
if (this.afterFirstItem) {
414+
return COMMA_SEPARATOR;
415+
}
416+
this.afterFirstItem = true;
417+
return EMPTY_BYTES;
418+
}
419+
420+
public Mono<DataBuffer> getPrefix(DataBufferFactory factory, @Nullable Map<String, Object> hints, Log logger) {
421+
return wrapBytes(OPEN_BRACKET, factory, hints, logger);
422+
}
423+
424+
public Mono<DataBuffer> getSuffix(DataBufferFactory factory, @Nullable Map<String, Object> hints, Log logger) {
425+
return wrapBytes(CLOSE_BRACKET, factory, hints, logger);
426+
}
427+
428+
private Mono<DataBuffer> wrapBytes(
429+
byte[] bytes, DataBufferFactory bufferFactory, @Nullable Map<String, Object> hints, Log logger) {
430+
431+
return Mono.fromCallable(() -> {
432+
DataBuffer buffer = bufferFactory.wrap(bytes);
433+
Hints.touchDataBuffer(buffer, hints, logger);
434+
return buffer;
435+
});
436+
}
437+
}
438+
388439
}

Diff for: ‎spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java

+19-19
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@
3333

3434
import org.springframework.core.ResolvableType;
3535
import org.springframework.core.io.buffer.DataBuffer;
36-
import org.springframework.core.io.buffer.DataBufferUtils;
3736
import org.springframework.core.testfixture.codec.AbstractEncoderTests;
3837
import org.springframework.http.MediaType;
3938
import org.springframework.http.codec.ServerSentEvent;
@@ -138,11 +137,11 @@ public void encodeNonStream() {
138137
);
139138

140139
testEncode(input, Pojo.class, step -> step
141-
.consumeNextWith(expectString("[" +
142-
"{\"foo\":\"foo\",\"bar\":\"bar\"}," +
143-
"{\"foo\":\"foofoo\",\"bar\":\"barbar\"}," +
144-
"{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}]")
145-
.andThen(DataBufferUtils::release))
140+
.consumeNextWith(expectString("["))
141+
.consumeNextWith(expectString("{\"foo\":\"foo\",\"bar\":\"bar\"}"))
142+
.consumeNextWith(expectString(",{\"foo\":\"foofoo\",\"bar\":\"barbar\"}"))
143+
.consumeNextWith(expectString(",{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}"))
144+
.consumeNextWith(expectString("]"))
146145
.verifyComplete());
147146
}
148147

@@ -151,8 +150,10 @@ public void encodeWithType() {
151150
Flux<ParentClass> input = Flux.just(new Foo(), new Bar());
152151

153152
testEncode(input, ParentClass.class, step -> step
154-
.consumeNextWith(expectString("[{\"type\":\"foo\"},{\"type\":\"bar\"}]")
155-
.andThen(DataBufferUtils::release))
153+
.consumeNextWith(expectString("["))
154+
.consumeNextWith(expectString("{\"type\":\"foo\"}"))
155+
.consumeNextWith(expectString(",{\"type\":\"bar\"}"))
156+
.consumeNextWith(expectString("]"))
156157
.verifyComplete());
157158
}
158159

@@ -169,12 +170,9 @@ public void encodeAsStreamWithCustomStreamingType() {
169170
);
170171

171172
testEncode(input, ResolvableType.forClass(Pojo.class), barMediaType, null, step -> step
172-
.consumeNextWith(expectString("{\"foo\":\"foo\",\"bar\":\"bar\"}\n")
173-
.andThen(DataBufferUtils::release))
174-
.consumeNextWith(expectString("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n")
175-
.andThen(DataBufferUtils::release))
176-
.consumeNextWith(expectString("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n")
177-
.andThen(DataBufferUtils::release))
173+
.consumeNextWith(expectString("{\"foo\":\"foo\",\"bar\":\"bar\"}\n"))
174+
.consumeNextWith(expectString("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n"))
175+
.consumeNextWith(expectString("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n"))
178176
.verifyComplete()
179177
);
180178
}
@@ -191,7 +189,7 @@ public void fieldLevelJsonView() {
191189
Map<String, Object> hints = singletonMap(JSON_VIEW_HINT, MyJacksonView1.class);
192190

193191
testEncode(input, type, null, hints, step -> step
194-
.consumeNextWith(expectString("{\"withView1\":\"with\"}").andThen(DataBufferUtils::release))
192+
.consumeNextWith(expectString("{\"withView1\":\"with\"}"))
195193
.verifyComplete()
196194
);
197195
}
@@ -208,7 +206,7 @@ public void classLevelJsonView() {
208206
Map<String, Object> hints = singletonMap(JSON_VIEW_HINT, MyJacksonView3.class);
209207

210208
testEncode(input, type, null, hints, step -> step
211-
.consumeNextWith(expectString("{\"withoutView\":\"without\"}").andThen(DataBufferUtils::release))
209+
.consumeNextWith(expectString("{\"withoutView\":\"without\"}"))
212210
.verifyComplete()
213211
);
214212
}
@@ -226,7 +224,7 @@ public void jacksonValue() {
226224
ResolvableType type = ResolvableType.forClass(MappingJacksonValue.class);
227225

228226
testEncode(Mono.just(jacksonValue), type, null, Collections.emptyMap(), step -> step
229-
.consumeNextWith(expectString("{\"withView1\":\"with\"}").andThen(DataBufferUtils::release))
227+
.consumeNextWith(expectString("{\"withView1\":\"with\"}"))
230228
.verifyComplete()
231229
);
232230
}
@@ -250,7 +248,7 @@ public void jacksonValueUnwrappedBeforeObjectMapperSelection() {
250248
String ls = System.lineSeparator(); // output below is different between Unix and Windows
251249
testEncode(Mono.just(jacksonValue), type, halMediaType, Collections.emptyMap(), step -> step
252250
.consumeNextWith(expectString("{" + ls + " \"withView1\" : \"with\"" + ls + "}")
253-
.andThen(DataBufferUtils::release))
251+
)
254252
.verifyComplete()
255253
);
256254
}
@@ -265,7 +263,9 @@ public void encodeWithFlushAfterWriteOff() {
265263
ResolvableType.forClass(Pojo.class), MimeTypeUtils.APPLICATION_JSON, Collections.emptyMap());
266264

267265
StepVerifier.create(result)
268-
.consumeNextWith(expectString("[{\"foo\":\"foo\",\"bar\":\"bar\"}]"))
266+
.consumeNextWith(expectString("["))
267+
.consumeNextWith(expectString("{\"foo\":\"foo\",\"bar\":\"bar\"}"))
268+
.consumeNextWith(expectString("]"))
269269
.expectComplete()
270270
.verify(Duration.ofSeconds(5));
271271
}

0 commit comments

Comments
 (0)