diff --git a/spring-web/src/main/java/org/springframework/http/codec/AbstractJacksonEncoder.java b/spring-web/src/main/java/org/springframework/http/codec/AbstractJacksonEncoder.java index 4a11813550cb..bd79f247fa98 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/AbstractJacksonEncoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/AbstractJacksonEncoder.java @@ -32,6 +32,7 @@ import tools.jackson.core.JsonEncoding; import tools.jackson.core.JsonGenerator; import tools.jackson.core.exc.JacksonIOException; +import tools.jackson.core.util.BufferRecycler; import tools.jackson.core.util.ByteArrayBuilder; import tools.jackson.databind.JavaType; import tools.jackson.databind.ObjectMapper; @@ -159,7 +160,8 @@ public Flux encode(Publisher inputStream, DataBufferFactory buffe } ObjectWriter writer = createObjectWriter(mapper, elementType, mimeType, hintsToUse); - ByteArrayBuilder byteBuilder = new ByteArrayBuilder(writer.generatorFactory()._getBufferRecycler()); + BufferRecycler recycler = writer.generatorFactory()._getBufferRecycler(); + ByteArrayBuilder byteBuilder = new ByteArrayBuilder(recycler); JsonEncoding encoding = getJsonEncoding(mimeType); JsonGenerator generator = mapper.createGenerator(byteBuilder, encoding); SequenceWriter sequenceWriter = writer.writeValues(generator); @@ -204,6 +206,11 @@ public Flux encode(Publisher inputStream, DataBufferFactory buffe catch (JacksonIOException ex) { logger.error("Could not close Encoder resources", ex); } + finally { + // Release strictly after the generator and builder are done: the + // recycler returns to a concurrent pool and may be reused at once. + recycler.releaseToPool(); + } }); } catch (JacksonIOException ex) { @@ -223,7 +230,8 @@ public DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory, ObjectWriter writer = createObjectWriter(mapper, valueType, mimeType, hints); - ByteArrayBuilder byteBuilder = new ByteArrayBuilder(writer.generatorFactory()._getBufferRecycler()); + BufferRecycler recycler = writer.generatorFactory()._getBufferRecycler(); + ByteArrayBuilder byteBuilder = new ByteArrayBuilder(recycler); try { JsonEncoding encoding = getJsonEncoding(mimeType); @@ -249,6 +257,7 @@ public DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory, } finally { byteBuilder.release(); + recycler.releaseToPool(); } } diff --git a/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java b/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java index bef38edcf570..d19f0243c541 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java @@ -28,6 +28,7 @@ import com.fasterxml.jackson.core.JsonEncoding; import com.fasterxml.jackson.core.JsonGenerator; import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.util.BufferRecycler; import com.fasterxml.jackson.core.util.ByteArrayBuilder; import com.fasterxml.jackson.databind.JavaType; import com.fasterxml.jackson.databind.ObjectMapper; @@ -171,7 +172,8 @@ public Flux encode(Publisher inputStream, DataBufferFactory buffe } ObjectWriter writer = createObjectWriter(mapper, elementType, mimeType, null, hintsToUse); - ByteArrayBuilder byteBuilder = new ByteArrayBuilder(writer.getFactory()._getBufferRecycler()); + BufferRecycler recycler = writer.getFactory()._getBufferRecycler(); + ByteArrayBuilder byteBuilder = new ByteArrayBuilder(recycler); JsonEncoding encoding = getJsonEncoding(mimeType); JsonGenerator generator = mapper.getFactory().createGenerator(byteBuilder, encoding); SequenceWriter sequenceWriter = writer.writeValues(generator); @@ -216,6 +218,11 @@ public Flux encode(Publisher inputStream, DataBufferFactory buffe catch (IOException ex) { logger.error("Could not close Encoder resources", ex); } + finally { + // Release strictly after the generator and builder are done: the + // recycler returns to a concurrent pool and may be reused at once. + recycler.releaseToPool(); + } }); } catch (IOException ex) { @@ -247,7 +254,8 @@ public DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory, writer = writer.with(filters); } - ByteArrayBuilder byteBuilder = new ByteArrayBuilder(writer.getFactory()._getBufferRecycler()); + BufferRecycler recycler = writer.getFactory()._getBufferRecycler(); + ByteArrayBuilder byteBuilder = new ByteArrayBuilder(recycler); try { JsonEncoding encoding = getJsonEncoding(mimeType); @@ -276,6 +284,7 @@ public DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory, } finally { byteBuilder.release(); + recycler.releaseToPool(); } } diff --git a/spring-web/src/test/java/org/springframework/http/codec/json/JacksonJsonEncoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/json/JacksonJsonEncoderTests.java index 6b900fe5dbf7..a597817b8a73 100644 --- a/spring-web/src/test/java/org/springframework/http/codec/json/JacksonJsonEncoderTests.java +++ b/spring-web/src/test/java/org/springframework/http/codec/json/JacksonJsonEncoderTests.java @@ -21,6 +21,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import com.fasterxml.jackson.annotation.JsonFilter; import com.fasterxml.jackson.annotation.JsonTypeInfo; @@ -29,6 +30,10 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; +import tools.jackson.core.json.JsonFactory; +import tools.jackson.core.util.BufferRecycler; +import tools.jackson.core.util.JsonRecyclerPools; +import tools.jackson.core.util.RecyclerPool; import tools.jackson.databind.SerializationFeature; import tools.jackson.databind.json.JsonMapper; import tools.jackson.databind.ser.FilterProvider; @@ -37,6 +42,7 @@ import org.springframework.core.ResolvableType; import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferUtils; import org.springframework.core.testfixture.codec.AbstractEncoderTests; import org.springframework.http.MediaType; import org.springframework.http.codec.json.JacksonViewBean.MyJacksonView1; @@ -305,6 +311,41 @@ void encodeAscii() { ); } + @Test + void releasesBufferRecyclerToPool() { + CountingRecyclerPool pool = new CountingRecyclerPool(); + JsonFactory factory = JsonFactory.builder().recyclerPool(pool).build(); + JacksonJsonEncoder encoder = new JacksonJsonEncoder(JsonMapper.builder(factory).build()); + + Pojo pojo = new Pojo("foo", "bar"); + ResolvableType type = ResolvableType.forClass(Pojo.class); + + // encodeValue (Mono path) + drain(encoder.encode(Mono.just(pojo), this.bufferFactory, type, APPLICATION_JSON, null)); + // Flux joined into a JSON array (non-streaming path) + drain(encoder.encode(Flux.just(pojo, pojo), this.bufferFactory, type, APPLICATION_JSON, null)); + // Flux with a streaming media type (NDJSON separator path) + drain(encoder.encode(Flux.just(pojo, pojo), this.bufferFactory, type, APPLICATION_NDJSON, null)); + + // Error termination of the streaming path must still release via doAfterTerminate + encoder.encode(Flux.error(new IllegalStateException("boom")), this.bufferFactory, type, APPLICATION_NDJSON, null) + .as(StepVerifier::create) + .expectError(IllegalStateException.class) + .verify(); + + assertThat(pool.acquired).hasPositiveValue(); + assertThat(pool.released).hasValue(pool.acquired.get()); + } + + private static void drain(Flux output) { + output.as(StepVerifier::create) + .thenConsumeWhile(buffer -> { + DataBufferUtils.release(buffer); + return true; + }) + .verifyComplete(); + } + @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type") private static class ParentClass { @@ -322,4 +363,26 @@ private static class Bar extends ParentClass { record JacksonFilteredBean(String property1, String property2) { } + @SuppressWarnings("serial") + private static final class CountingRecyclerPool implements RecyclerPool { + + final AtomicInteger acquired = new AtomicInteger(); + + final AtomicInteger released = new AtomicInteger(); + + private final RecyclerPool delegate = JsonRecyclerPools.newConcurrentDequePool(); + + @Override + public BufferRecycler acquirePooled() { + this.acquired.incrementAndGet(); + return this.delegate.acquirePooled(); + } + + @Override + public void releasePooled(BufferRecycler recycler) { + this.released.incrementAndGet(); + this.delegate.releasePooled(recycler); + } + } + }