Switch to Reactor snapshots and remove workaround

Following the 5.2 M1 release we can switch back to Reactor snapshots
and remove the workaround for a fix coming in Reactor Core 3.2.9.

Author: rstoyanchev

build.gradle

@@ -40,7 +40,7 @@ ext {
 	log4jVersion         = "2.11.2"
 	nettyVersion         = "4.1.34.Final"
 	quartzVersion        = "2.3.1"
-	reactorVersion       = "Californium-SR6"
+	reactorVersion       = "Californium-BUILD-SNAPSHOT"
 	rxjavaVersion        = "1.3.8"
 	rxjavaAdapterVersion = "1.2.1"
 	rxjava2Version       = "2.2.8"
@@ -149,6 +149,7 @@ configure(allprojects) { project ->
 
 	repositories {
 		maven { url "https://repo.spring.io/libs-release" }
+		maven { url "https://repo.spring.io/snapshot" } // Reactor
 		maven { url "https://oss.jfrog.org/artifactory/libs-snapshot" } // RSocket
 		mavenLocal()
 	}

spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java

@@ -57,12 +57,6 @@ public abstract class DataBufferUtils {
 
 	private static final Consumer<DataBuffer> RELEASE_CONSUMER = DataBufferUtils::release;
 
-	/**
-	 * Workaround to disable use of pooled buffers:
-	 * https://github.com/reactor/reactor-core/issues/1634.
-	 */
-	private static final DataBufferFactory defaultDataBufferFactory = new DefaultDataBufferFactory();
-
 
 	//---------------------------------------------------------------------
 	// Reading
@@ -141,14 +135,12 @@ public static Flux<DataBuffer> readAsynchronousFileChannel(
 		Assert.isTrue(position >= 0, "'position' must be >= 0");
 		Assert.isTrue(bufferSize > 0, "'bufferSize' must be > 0");
 
-		DataBufferFactory bufferFactoryToUse = defaultDataBufferFactory;
-
 		Flux<DataBuffer> flux = Flux.using(channelSupplier,
 				channel -> Flux.create(sink -> {
 					ReadCompletionHandler handler =
-							new ReadCompletionHandler(channel, sink, position, bufferFactoryToUse, bufferSize);
+							new ReadCompletionHandler(channel, sink, position, bufferFactory, bufferSize);
 					sink.onDispose(handler::dispose);
-					DataBuffer dataBuffer = bufferFactoryToUse.allocateBuffer(bufferSize);
+					DataBuffer dataBuffer = bufferFactory.allocateBuffer(bufferSize);
 					ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, bufferSize);
 					channel.read(byteBuffer, position, dataBuffer, handler);
 				}),

spring-messaging/spring-messaging.gradle

@@ -7,7 +7,7 @@ dependencyManagement {
 	}
 }
 
-def rsocketVersion = "0.12.1-RC3"
+def rsocketVersion = "0.12.1-RC4-SNAPSHOT"
 
 dependencies {
 	compile(project(":spring-beans"))

spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java

@@ -238,9 +238,10 @@ public RSocketStrategies rsocketStrategies() {
 
 
 	/**
-	 * Similar {@link org.springframework.core.io.buffer.LeakAwareDataBufferFactory}
-	 * but extends {@link NettyDataBufferFactory} rather than rely on
-	 * decoration, since {@link PayloadUtils} does instanceof checks.
+	 * Unlike {@link org.springframework.core.io.buffer.LeakAwareDataBufferFactory}
+	 * this one is an instance of {@link NettyDataBufferFactory} which is necessary
+	 * since {@link PayloadUtils} does instanceof checks, and that also allows
+	 * intercepting {@link NettyDataBufferFactory#wrap(ByteBuf)}.
 	 */
 	private static class LeakAwareNettyDataBufferFactory extends NettyDataBufferFactory {
 
@@ -277,32 +278,33 @@ void reset() {
 
 		@Override
 		public NettyDataBuffer allocateBuffer() {
-			return (NettyDataBuffer) record(super.allocateBuffer());
+			return (NettyDataBuffer) recordHint(super.allocateBuffer());
 		}
 
 		@Override
 		public NettyDataBuffer allocateBuffer(int initialCapacity) {
-			return (NettyDataBuffer) record(super.allocateBuffer(initialCapacity));
+			return (NettyDataBuffer) recordHint(super.allocateBuffer(initialCapacity));
 		}
 
 		@Override
 		public NettyDataBuffer wrap(ByteBuf byteBuf) {
 			NettyDataBuffer dataBuffer = super.wrap(byteBuf);
 			if (byteBuf != Unpooled.EMPTY_BUFFER) {
-				record(dataBuffer);
+				recordHint(dataBuffer);
 			}
 			return dataBuffer;
 		}
 
 		@Override
 		public DataBuffer join(List<? extends DataBuffer> dataBuffers) {
-			return record(super.join(dataBuffers));
+			return recordHint(super.join(dataBuffers));
 		}
 
-		private DataBuffer record(DataBuffer buffer) {
-			this.created.add(new DataBufferLeakInfo(buffer, new AssertionError(String.format(
+		private DataBuffer recordHint(DataBuffer buffer) {
+			AssertionError error = new AssertionError(String.format(
 					"DataBuffer leak: {%s} {%s} not released.%nStacktrace at buffer creation: ", buffer,
-					ObjectUtils.getIdentityHexString(((NettyDataBuffer) buffer).getNativeBuffer())))));
+					ObjectUtils.getIdentityHexString(((NettyDataBuffer) buffer).getNativeBuffer())));
+			this.created.add(new DataBufferLeakInfo(buffer, error));
 			return buffer;
 		}
 	}