Make JsonObjectDecoder use allocator property on DataBuffer.

spring-web-reactive/src/main/java/org/springframework/core/codec/support/JsonObjectDecoder.java

@@ -55,23 +55,21 @@ public class JsonObjectDecoder extends AbstractDecoder<DataBuffer> {
 
 	private static final int ST_DECODING_ARRAY_STREAM = 2;
 
-	private final DataBufferAllocator allocator;
-
 	private final int maxObjectLength;
 
 	private final boolean streamArrayElements;
 
-	public JsonObjectDecoder(DataBufferAllocator allocator) {
+	public JsonObjectDecoder() {
 		// 1 MB
-		this(allocator, 1024 * 1024);
+		this(1024 * 1024);
 	}
 
-	public JsonObjectDecoder(DataBufferAllocator allocator, int maxObjectLength) {
-		this(allocator, maxObjectLength, true);
+	public JsonObjectDecoder(int maxObjectLength) {
+		this(maxObjectLength, true);
 	}
 
-	public JsonObjectDecoder(DataBufferAllocator allocator, boolean streamArrayElements) {
-		this(allocator, 1024 * 1024, streamArrayElements);
+	public JsonObjectDecoder(boolean streamArrayElements) {
+		this(1024 * 1024, streamArrayElements);
 	}
 
 
@@ -84,11 +82,10 @@ public class JsonObjectDecoder extends AbstractDecoder<DataBuffer> {
 	 * is an array, each of its entries is passed through the pipeline individually
 	 * and immediately after it was fully received, allowing for arrays with
 	 */
-	public JsonObjectDecoder(DataBufferAllocator allocator, int maxObjectLength,
+	public JsonObjectDecoder(int maxObjectLength,
 			boolean streamArrayElements) {
 		super(new MimeType("application", "json", StandardCharsets.UTF_8),
 				new MimeType("application", "*+json", StandardCharsets.UTF_8));
-		this.allocator = allocator;
 		if (maxObjectLength < 1) {
 			throw new IllegalArgumentException("maxObjectLength must be a positive int");
 		}
@@ -133,6 +130,7 @@ public class JsonObjectDecoder extends AbstractDecoder<DataBuffer> {
 					return Flux.error(new IllegalStateException("object length exceeds " +
 							maxObjectLength + ": " + this.writerIndex + " bytes discarded"));
 				}
+				DataBufferAllocator allocator = b.allocator();
 				for (/* use current index */; this.index < this.writerIndex; this.index++) {
 					byte c = this.input.getByte(this.index);
 					if (this.state == ST_DECODING_NORMAL) {

spring-web-reactive/src/main/java/org/springframework/web/client/reactive/WebClient.java

@@ -89,7 +89,7 @@ public final class WebClient {
 		this.messageEncoders = Arrays.asList(new ByteBufferEncoder(allocator), new StringEncoder(allocator),
 				new JacksonJsonEncoder(allocator));
 		this.messageDecoders = Arrays.asList(new ByteBufferDecoder(), new StringDecoder(allocator),
-				new JacksonJsonDecoder(new JsonObjectDecoder(allocator)));
+				new JacksonJsonDecoder(new JsonObjectDecoder()));
 	}
 
 	/**

spring-web-reactive/src/main/java/org/springframework/web/reactive/method/annotation/RequestMappingHandlerAdapter.java

@@ -101,7 +101,7 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin
 
 			List<Decoder<?>> decoders = Arrays.asList(new ByteBufferDecoder(),
 					new StringDecoder(allocator),
-					new JacksonJsonDecoder(new JsonObjectDecoder(allocator)));
+					new JacksonJsonDecoder(new JsonObjectDecoder()));
 
 			this.argumentResolvers.add(new RequestParamArgumentResolver());
 			this.argumentResolvers.add(new RequestBodyArgumentResolver(decoders, this.conversionService));

spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectDecoderTests.java

@@ -20,11 +20,10 @@ import java.nio.charset.StandardCharsets;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.test.TestSubscriber;
 
 import org.springframework.core.io.buffer.DataBuffer;
 
-import reactor.core.test.TestSubscriber;
-
 /**
  * @author Sebastien Deleuze
  */
@@ -33,7 +32,7 @@ public class JsonObjectDecoderTests extends AbstractAllocatingTestCase {
 
 	@Test
 	public void decodeSingleChunkToJsonObject()  {
-		JsonObjectDecoder decoder = new JsonObjectDecoder(allocator);
+		JsonObjectDecoder decoder = new JsonObjectDecoder();
 		Flux<DataBuffer> source =
 				Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
 		Flux<String> output =
@@ -45,7 +44,7 @@ public class JsonObjectDecoderTests extends AbstractAllocatingTestCase {
 
 	@Test
 	public void decodeMultipleChunksToJsonObject() throws InterruptedException {
-		JsonObjectDecoder decoder = new JsonObjectDecoder(allocator);
+		JsonObjectDecoder decoder = new JsonObjectDecoder();
 		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"foo\": \"foofoo\""),
 				stringBuffer(", \"bar\": \"barbar\"}"));
 		Flux<String> output =
@@ -57,7 +56,7 @@ public class JsonObjectDecoderTests extends AbstractAllocatingTestCase {
 
 	@Test
 	public void decodeSingleChunkToArray() throws InterruptedException {
-		JsonObjectDecoder decoder = new JsonObjectDecoder(allocator);
+		JsonObjectDecoder decoder = new JsonObjectDecoder();
 		Flux<DataBuffer> source = Flux.just(stringBuffer(
 				"[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]"));
 		Flux<String> output =
@@ -70,7 +69,7 @@ public class JsonObjectDecoderTests extends AbstractAllocatingTestCase {
 
 	@Test
 	public void decodeMultipleChunksToArray() throws InterruptedException {
-		JsonObjectDecoder decoder = new JsonObjectDecoder(allocator);
+		JsonObjectDecoder decoder = new JsonObjectDecoder();
 		Flux<DataBuffer> source =
 				Flux.just(stringBuffer("[{\"foo\": \"foofoo\", \"bar\""), stringBuffer(
 						": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]"));