From 2f95859715a49fbd7a6b58c1c0ed4e0f4c369843 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Thu, 21 Mar 2019 19:31:06 -0400 Subject: [PATCH] Upgrade to RSocket 0.12.1-RC3-SNAPSHOT Closes gh-22629 --- build.gradle | 1 + spring-messaging/spring-messaging.gradle | 2 +- .../messaging/rsocket/PayloadUtils.java | 10 +++------- .../messaging/rsocket/RSocketBufferLeakTests.java | 6 +++--- .../rsocket/RSocketClientToServerIntegrationTests.java | 7 ++++--- .../rsocket/RSocketServerToClientIntegrationTests.java | 6 +++--- 6 files changed, 15 insertions(+), 17 deletions(-) diff --git a/build.gradle b/build.gradle index 3d458ddc9d..0870c61985 100644 --- a/build.gradle +++ b/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() } diff --git a/spring-messaging/spring-messaging.gradle b/spring-messaging/spring-messaging.gradle index fd469281c6..d1a8f0cf31 100644 --- a/spring-messaging/spring-messaging.gradle +++ b/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")) diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/PayloadUtils.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/PayloadUtils.java index 25f69c0ff8..c83facc15c 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/PayloadUtils.java +++ b/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.DataBufferFactory; 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 @@ abstract class PayloadUtils { 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) { diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java index 86c12f754b..f2eed26541 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java @@ -28,9 +28,9 @@ import io.netty.buffer.PooledByteBufAllocator; 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,7 +89,7 @@ public class RSocketBufferLeakTests { 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)) @@ -97,7 +97,7 @@ public class RSocketBufferLeakTests { .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)) diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java index 318335b43b..190219dd9e 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java @@ -19,9 +19,9 @@ package org.springframework.messaging.rsocket; 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,7 +71,7 @@ public class RSocketClientToServerIntegrationTests { 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() @@ -79,7 +79,7 @@ public class RSocketClientToServerIntegrationTests { 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 class RSocketClientToServerIntegrationTests { .expectNext("Hello 1") .expectNext("Hello 2") .expectNext("Hello 3") + .thenAwait(Duration.ofMillis(50)) .thenCancel() .verify(Duration.ofSeconds(5)); diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java index ed47f44d00..3a2c2c6817 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java @@ -21,9 +21,9 @@ import java.util.Collections; 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 class RSocketServerToClientIntegrationTests { 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 @@ public class RSocketServerToClientIntegrationTests { 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()