Upgrade to RSocket 0.12.1-RC3-SNAPSHOT

Closes gh-22629

Author: rstoyanchev

build.gradle

@@ -150,6 +150,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-messaging/spring-messaging.gradle

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

spring-messaging/src/main/java/org/springframework/messaging/rsocket/PayloadUtils.java

@@ -17,7 +17,6 @@
 package org.springframework.messaging.rsocket;
 
 import io.netty.buffer.ByteBuf;
-import io.rsocket.Frame;
 import io.rsocket.Payload;
 import io.rsocket.util.ByteBufPayload;
 import io.rsocket.util.DefaultPayload;
@@ -27,7 +26,6 @@
 import org.springframework.core.io.buffer.DefaultDataBuffer;
 import org.springframework.core.io.buffer.NettyDataBuffer;
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
-import org.springframework.util.Assert;
 
 /**
  * Static utility methods to create {@link Payload} from {@link DataBuffer}s
@@ -53,11 +51,9 @@ public static DataBuffer retainDataAndReleasePayload(Payload payload, DataBuffer
 				ByteBuf byteBuf = payload.sliceData().retain();
 				return ((NettyDataBufferFactory) bufferFactory).wrap(byteBuf);
 			}
-
-			Assert.isTrue(!(payload instanceof ByteBufPayload) && !(payload instanceof Frame),
-					"NettyDataBufferFactory expected, actual: " + bufferFactory.getClass().getSimpleName());
-
-			return bufferFactory.wrap(payload.getData());
+			else {
+				return bufferFactory.wrap(payload.getData());
+			}
 		}
 		finally {
 			if (payload.refCnt() > 0) {

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

@@ -28,9 +28,9 @@
 import io.netty.buffer.Unpooled;
 import io.netty.util.ReferenceCounted;
 import io.rsocket.AbstractRSocket;
-import io.rsocket.Frame;
 import io.rsocket.RSocket;
 import io.rsocket.RSocketFactory;
+import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.plugins.RSocketInterceptor;
 import io.rsocket.transport.netty.client.TcpClientTransport;
 import io.rsocket.transport.netty.server.CloseableChannel;
@@ -89,15 +89,15 @@ public static void setupOnce() {
 		context = new AnnotationConfigApplicationContext(ServerConfig.class);
 
 		server = RSocketFactory.receive()
-				.frameDecoder(Frame::retain) // zero copy
+				.frameDecoder(PayloadDecoder.ZERO_COPY)
 				.addServerPlugin(payloadInterceptor) // intercept responding
 				.acceptor(context.getBean(MessageHandlerAcceptor.class))
 				.transport(TcpServerTransport.create("localhost", 7000))
 				.start()
 				.block();
 
 		client = RSocketFactory.connect()
-				.frameDecoder(Frame::retain) // zero copy
+				.frameDecoder(PayloadDecoder.ZERO_COPY)
 				.addClientPlugin(payloadInterceptor) // intercept outgoing requests
 				.dataMimeType(MimeTypeUtils.TEXT_PLAIN_VALUE)
 				.transport(TcpClientTransport.create("localhost", 7000))

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

@@ -19,9 +19,9 @@
 import java.time.Duration;
 
 import io.netty.buffer.PooledByteBufAllocator;
-import io.rsocket.Frame;
 import io.rsocket.RSocket;
 import io.rsocket.RSocketFactory;
+import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.transport.netty.client.TcpClientTransport;
 import io.rsocket.transport.netty.server.CloseableChannel;
 import io.rsocket.transport.netty.server.TcpServerTransport;
@@ -71,15 +71,15 @@ public static void setupOnce() {
 
 		server = RSocketFactory.receive()
 				.addServerPlugin(interceptor)
-				.frameDecoder(Frame::retain)  // as per https://github.com/rsocket/rsocket-java#zero-copy
+				.frameDecoder(PayloadDecoder.ZERO_COPY)
 				.acceptor(context.getBean(MessageHandlerAcceptor.class))
 				.transport(TcpServerTransport.create("localhost", 7000))
 				.start()
 				.block();
 
 		client = RSocketFactory.connect()
 				.dataMimeType(MimeTypeUtils.TEXT_PLAIN_VALUE)
-				.frameDecoder(Frame::retain)  // as per https://github.com/rsocket/rsocket-java#zero-copy
+				.frameDecoder(PayloadDecoder.ZERO_COPY)
 				.transport(TcpClientTransport.create("localhost", 7000))
 				.start()
 				.block();
@@ -105,6 +105,7 @@ public void fireAndForget() {
 				.expectNext("Hello 1")
 				.expectNext("Hello 2")
 				.expectNext("Hello 3")
+				.thenAwait(Duration.ofMillis(50))
 				.thenCancel()
 				.verify(Duration.ofSeconds(5));
 

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

@@ -21,9 +21,9 @@
 
 import io.netty.buffer.PooledByteBufAllocator;
 import io.rsocket.Closeable;
-import io.rsocket.Frame;
 import io.rsocket.RSocket;
 import io.rsocket.RSocketFactory;
+import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.transport.netty.client.TcpClientTransport;
 import io.rsocket.transport.netty.server.TcpServerTransport;
 import io.rsocket.util.DefaultPayload;
@@ -64,7 +64,7 @@ public static void setupOnce() {
 		context = new AnnotationConfigApplicationContext(RSocketConfig.class);
 
 		server = RSocketFactory.receive()
-				.frameDecoder(Frame::retain)  // as per https://github.com/rsocket/rsocket-java#zero-copy
+				.frameDecoder(PayloadDecoder.ZERO_COPY)
 				.acceptor(context.getBean("serverAcceptor", MessageHandlerAcceptor.class))
 				.transport(TcpServerTransport.create("localhost", 7000))
 				.start()
@@ -108,7 +108,7 @@ private static void connectAndVerify(String destination) {
 			rsocket = RSocketFactory.connect()
 					.setupPayload(DefaultPayload.create("", destination))
 					.dataMimeType("text/plain")
-					.frameDecoder(Frame::retain)  // as per https://github.com/rsocket/rsocket-java#zero-copy
+					.frameDecoder(PayloadDecoder.ZERO_COPY)
 					.acceptor(context.getBean("clientAcceptor", MessageHandlerAcceptor.class))
 					.transport(TcpClientTransport.create("localhost", 7000))
 					.start()