diff --git a/build.gradle b/build.gradle index 679b93135..342a94954 100644 --- a/build.gradle +++ b/build.gradle @@ -30,8 +30,7 @@ subprojects { dependencyManagement { imports { - // TODO: Upgrade to latest version - mavenBom 'io.projectreactor:reactor-bom:Bismuth-SR11' + mavenBom 'io.projectreactor:reactor-bom:Californium-RELEASE' } dependencies { diff --git a/rsocket-core/build.gradle b/rsocket-core/build.gradle index 0cdb6103c..7743715a3 100644 --- a/rsocket-core/build.gradle +++ b/rsocket-core/build.gradle @@ -27,7 +27,6 @@ dependencies { api 'io.netty:netty-buffer' api 'io.projectreactor:reactor-core' - implementation 'io.projectreactor.addons:reactor-extra' implementation 'org.jctools:jctools-core' implementation 'org.slf4j:slf4j-api' diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketClient.java b/rsocket-core/src/main/java/io/rsocket/RSocketClient.java index 052b5b0e5..44460a938 100644 --- a/rsocket-core/src/main/java/io/rsocket/RSocketClient.java +++ b/rsocket-core/src/main/java/io/rsocket/RSocketClient.java @@ -430,10 +430,9 @@ private void handleStreamZero(FrameType type, Frame frame) { lifecycle.terminate(error); errorConsumer.accept(error); connection.dispose(); + break; case LEASE: - { - break; - } + break; case KEEPALIVE: if (keepAliveHandler != null) { keepAliveHandler.receive(frame); diff --git a/rsocket-core/src/main/java/io/rsocket/fragmentation/FragmentationDuplexConnection.java b/rsocket-core/src/main/java/io/rsocket/fragmentation/FragmentationDuplexConnection.java index e765d54e2..e70f45d9f 100644 --- a/rsocket-core/src/main/java/io/rsocket/fragmentation/FragmentationDuplexConnection.java +++ b/rsocket-core/src/main/java/io/rsocket/fragmentation/FragmentationDuplexConnection.java @@ -18,7 +18,6 @@ import static io.rsocket.fragmentation.FrameReassembler.createFrameReassembler; import static io.rsocket.util.AbstractionLeakingFrameUtils.toAbstractionLeakingFrame; -import static reactor.function.TupleUtils.function; import io.netty.buffer.ByteBufAllocator; import io.netty.buffer.PooledByteBufAllocator; @@ -115,7 +114,7 @@ public Flux receive() { return delegate .receive() .map(AbstractionLeakingFrameUtils::fromAbstractionLeakingFrame) - .concatMap(function(this::toReassembledFrames)); + .concatMap(t2 -> toReassembledFrames(t2.getT1(), t2.getT2())); } @Override @@ -125,7 +124,7 @@ public Mono send(Publisher frames) { return delegate.send( Flux.from(frames) .map(AbstractionLeakingFrameUtils::fromAbstractionLeakingFrame) - .concatMap(function(this::toFragmentedFrames))); + .concatMap(t2 -> toFragmentedFrames(t2.getT1(), t2.getT2()))); } private Flux toFragmentedFrames(int streamId, io.rsocket.framing.Frame frame) { diff --git a/rsocket-examples/src/test/java/io/rsocket/integration/IntegrationTest.java b/rsocket-examples/src/test/java/io/rsocket/integration/IntegrationTest.java index bd9231e43..627b1d7da 100644 --- a/rsocket-examples/src/test/java/io/rsocket/integration/IntegrationTest.java +++ b/rsocket-examples/src/test/java/io/rsocket/integration/IntegrationTest.java @@ -31,7 +31,7 @@ import io.rsocket.plugins.RSocketInterceptor; import io.rsocket.test.TestSubscriber; import io.rsocket.transport.netty.client.TcpClientTransport; -import io.rsocket.transport.netty.server.NettyContextCloseable; +import io.rsocket.transport.netty.server.CloseableChannel; import io.rsocket.transport.netty.server.TcpServerTransport; import io.rsocket.util.DefaultPayload; import io.rsocket.util.RSocketProxy; @@ -83,7 +83,7 @@ public Mono requestResponse(Payload payload) { }; } - private NettyContextCloseable server; + private CloseableChannel server; private RSocket client; private AtomicInteger requestCount; private CountDownLatch disconnectionCounter; diff --git a/rsocket-examples/src/test/java/io/rsocket/integration/InteractionsLoadTest.java b/rsocket-examples/src/test/java/io/rsocket/integration/InteractionsLoadTest.java index 1c488db4f..6c8f0e8fa 100644 --- a/rsocket-examples/src/test/java/io/rsocket/integration/InteractionsLoadTest.java +++ b/rsocket-examples/src/test/java/io/rsocket/integration/InteractionsLoadTest.java @@ -6,7 +6,7 @@ import io.rsocket.RSocketFactory; import io.rsocket.test.SlowTest; import io.rsocket.transport.netty.client.TcpClientTransport; -import io.rsocket.transport.netty.server.NettyContextCloseable; +import io.rsocket.transport.netty.server.CloseableChannel; import io.rsocket.transport.netty.server.TcpServerTransport; import io.rsocket.util.DefaultPayload; import java.time.Duration; @@ -23,7 +23,7 @@ public class InteractionsLoadTest { public void channel() { TcpServerTransport serverTransport = TcpServerTransport.create(0); - NettyContextCloseable server = + CloseableChannel server = RSocketFactory.receive() .acceptor((setup, rsocket) -> Mono.just(new EchoRSocket())) .transport(serverTransport) diff --git a/rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java b/rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java index 9aadb491d..f5d048508 100644 --- a/rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java +++ b/rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java @@ -24,7 +24,7 @@ import io.rsocket.RSocket; import io.rsocket.RSocketFactory; import io.rsocket.transport.netty.client.TcpClientTransport; -import io.rsocket.transport.netty.server.NettyContextCloseable; +import io.rsocket.transport.netty.server.CloseableChannel; import io.rsocket.transport.netty.server.TcpServerTransport; import io.rsocket.util.DefaultPayload; import io.rsocket.util.EmptyPayload; @@ -42,7 +42,7 @@ public class TcpIntegrationTest { private AbstractRSocket handler; - private NettyContextCloseable server; + private CloseableChannel server; @Before public void startup() { diff --git a/rsocket-transport-netty/build.gradle b/rsocket-transport-netty/build.gradle index f50aa9752..71d0d5088 100644 --- a/rsocket-transport-netty/build.gradle +++ b/rsocket-transport-netty/build.gradle @@ -19,11 +19,17 @@ plugins { id 'maven-publish' id 'com.jfrog.artifactory' id 'com.jfrog.bintray' + id "com.google.osdetector" version "1.4.0" +} + +def os_suffix = "" +if (osdetector.classifier in ["linux-x86_64"] || ["osx-x86_64"] || ["windows-x86_64"]) { + os_suffix = ":" + osdetector.classifier } dependencies { api project(':rsocket-core') - api 'io.projectreactor.ipc:reactor-netty' + api 'io.projectreactor.netty:reactor-netty' compileOnly 'com.google.code.findbugs:jsr305' @@ -35,6 +41,7 @@ dependencies { testRuntimeOnly 'ch.qos.logback:logback-classic' testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine' + testRuntimeOnly 'io.netty:netty-tcnative-boringssl-static:2.0.14.Final' + os_suffix } description = 'Reactor Netty RSocket transport implementations (TCP, Websocket)' diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java index dca183bef..6ac2dbe0a 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java @@ -21,50 +21,42 @@ import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.ipc.netty.NettyContext; -import reactor.ipc.netty.NettyInbound; -import reactor.ipc.netty.NettyOutbound; +import reactor.netty.Connection; + +import java.util.Objects; /** An implementation of {@link DuplexConnection} that connects via TCP. */ public final class TcpDuplexConnection implements DuplexConnection { - private final NettyContext context; - - private final NettyInbound in; - - private final NettyOutbound out; + private final Connection connection; /** * Creates a new instance * - * @param in the {@link NettyInbound} to listen on - * @param out the {@link NettyOutbound} to send with - * @param context the {@link NettyContext} to for managing the server + * @param connection the {@link Connection} to for managing the server */ - public TcpDuplexConnection(NettyInbound in, NettyOutbound out, NettyContext context) { - this.in = in; - this.out = out; - this.context = context; + public TcpDuplexConnection(Connection connection) { + this.connection = Objects.requireNonNull(connection, "connection must not be null"); } @Override public void dispose() { - context.dispose(); + connection.dispose(); } @Override public boolean isDisposed() { - return context.isDisposed(); + return connection.isDisposed(); } @Override public Mono onClose() { - return context.onClose(); + return connection.onDispose(); } @Override public Flux receive() { - return in.receive().map(buf -> Frame.from(buf.retain())); + return connection.inbound().receive().map(buf -> Frame.from(buf.retain())); } @Override @@ -74,6 +66,6 @@ public Mono send(Publisher frames) { @Override public Mono sendOne(Frame frame) { - return out.sendObject(frame.content()).then(); + return connection.outbound().sendObject(frame.content()).then(); } } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpUriHandler.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpUriHandler.java index 9821f5fa7..952a1e398 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpUriHandler.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpUriHandler.java @@ -24,7 +24,7 @@ import java.net.URI; import java.util.Objects; import java.util.Optional; -import reactor.ipc.netty.tcp.TcpServer; +import reactor.netty.tcp.TcpServer; /** * An implementation of {@link UriHandler} that creates {@link TcpClientTransport}s and {@link @@ -53,6 +53,9 @@ public Optional buildServer(URI uri) { return Optional.empty(); } - return Optional.of(TcpServerTransport.create(TcpServer.create(uri.getHost(), uri.getPort()))); + return Optional.of(TcpServerTransport.create( + TcpServer.create() + .host(uri.getHost()) + .port(uri.getPort()))); } } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java index 7c71d2692..efb47c2a5 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java @@ -28,9 +28,7 @@ import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.ipc.netty.NettyContext; -import reactor.ipc.netty.NettyInbound; -import reactor.ipc.netty.NettyOutbound; +import reactor.netty.Connection; /** * An implementation of {@link DuplexConnection} that connects via a Websocket. @@ -41,46 +39,38 @@ */ public final class WebsocketDuplexConnection implements DuplexConnection { - private final NettyContext context; - - private final NettyInbound in; - - private final NettyOutbound out; + private final Connection connection; /** * Creates a new instance * - * @param in the {@link NettyInbound} to listen on - * @param out the {@link NettyOutbound} to send with - * @param context the {@link NettyContext} to for managing the server + * @param connection the {@link Connection} to for managing the server */ - public WebsocketDuplexConnection(NettyInbound in, NettyOutbound out, NettyContext context) { - this.in = Objects.requireNonNull(in, "in must not be null"); - this.out = Objects.requireNonNull(out, "out must not be null"); - this.context = Objects.requireNonNull(context, "context must not be null"); + public WebsocketDuplexConnection(Connection connection) { + this.connection = Objects.requireNonNull(connection, "connection must not be null"); } @Override public void dispose() { - context.dispose(); + connection.dispose(); } @Override public boolean isDisposed() { - return context.isDisposed(); + return connection.isDisposed(); } @Override public Mono onClose() { - return context.onClose(); + return connection.onDispose(); } @Override public Flux receive() { - return in.receive() + return connection.inbound().receive() .map( buf -> { - CompositeByteBuf composite = context.channel().alloc().compositeBuffer(); + CompositeByteBuf composite = connection.channel().alloc().compositeBuffer(); ByteBuf length = wrappedBuffer(new byte[FRAME_LENGTH_SIZE]); FrameHeaderFlyweight.encodeLength(length, 0, buf.readableBytes()); composite.addComponents(true, length, buf.retain()); @@ -95,7 +85,7 @@ public Mono send(Publisher frames) { @Override public Mono sendOne(Frame frame) { - return out.sendObject(new BinaryWebSocketFrame(frame.content().skipBytes(FRAME_LENGTH_SIZE))) + return connection.outbound().sendObject(new BinaryWebSocketFrame(frame.content().skipBytes(FRAME_LENGTH_SIZE))) .then(); } } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/client/TcpClientTransport.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/client/TcpClientTransport.java index 36829c907..aca238d31 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/client/TcpClientTransport.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/client/TcpClientTransport.java @@ -24,7 +24,7 @@ import java.net.InetSocketAddress; import java.util.Objects; import reactor.core.publisher.Mono; -import reactor.ipc.netty.tcp.TcpClient; +import reactor.netty.tcp.TcpClient; /** * An implementation of {@link ClientTransport} that connects to a {@link ServerTransport} via TCP. @@ -44,7 +44,7 @@ private TcpClientTransport(TcpClient client) { * @return a new instance */ public static TcpClientTransport create(int port) { - TcpClient tcpClient = TcpClient.create(port); + TcpClient tcpClient = TcpClient.create().port(port); return create(tcpClient); } @@ -59,7 +59,7 @@ public static TcpClientTransport create(int port) { public static TcpClientTransport create(String bindAddress, int port) { Objects.requireNonNull(bindAddress, "bindAddress must not be null"); - TcpClient tcpClient = TcpClient.create(bindAddress, port); + TcpClient tcpClient = TcpClient.create().host(bindAddress).port(port); return create(tcpClient); } @@ -73,7 +73,7 @@ public static TcpClientTransport create(String bindAddress, int port) { public static TcpClientTransport create(InetSocketAddress address) { Objects.requireNonNull(address, "address must not be null"); - TcpClient tcpClient = TcpClient.create(address.getHostString(), address.getPort()); + TcpClient tcpClient = TcpClient.create().addressSupplier(() -> address); return create(tcpClient); } @@ -92,20 +92,9 @@ public static TcpClientTransport create(TcpClient client) { @Override public Mono connect() { - return Mono.create( - sink -> - client - .newHandler( - (in, out) -> { - in.context().addHandler(new RSocketLengthCodec()); - - TcpDuplexConnection connection = - new TcpDuplexConnection(in, out, in.context()); - - sink.success(connection); - return connection.onClose(); - }) - .doOnError(sink::error) - .subscribe()); + return client + .doOnConnected(c -> c.addHandlerLast(new RSocketLengthCodec())) + .connect() + .map(TcpDuplexConnection::new); } } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/client/WebsocketClientTransport.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/client/WebsocketClientTransport.java index 01851f82a..8f4dac7eb 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/client/WebsocketClientTransport.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/client/WebsocketClientTransport.java @@ -30,8 +30,11 @@ import java.util.Map; import java.util.Objects; import java.util.function.Supplier; + import reactor.core.publisher.Mono; -import reactor.ipc.netty.http.client.HttpClient; +import reactor.netty.Connection; +import reactor.netty.http.client.HttpClient; +import reactor.netty.tcp.TcpClient; /** * An implementation of {@link ClientTransport} that connects to a {@link ServerTransport} via a @@ -57,8 +60,8 @@ private WebsocketClientTransport(HttpClient client, String path) { * @return a new instance */ public static WebsocketClientTransport create(int port) { - HttpClient httpClient = HttpClient.create(port); - return create(httpClient, "/"); + TcpClient client = TcpClient.create().port(port); + return create(client); } /** @@ -72,8 +75,8 @@ public static WebsocketClientTransport create(int port) { public static WebsocketClientTransport create(String bindAddress, int port) { Objects.requireNonNull(bindAddress, "bindAddress must not be null"); - HttpClient httpClient = HttpClient.create(bindAddress, port); - return create(httpClient, "/"); + TcpClient client = TcpClient.create().host(bindAddress).port(port); + return create(client); } /** @@ -86,7 +89,8 @@ public static WebsocketClientTransport create(String bindAddress, int port) { public static WebsocketClientTransport create(InetSocketAddress address) { Objects.requireNonNull(address, "address must not be null"); - return create(address.getHostName(), address.getPort()); + TcpClient client = TcpClient.create().addressSupplier(() -> address); + return create(client); } /** @@ -99,8 +103,21 @@ public static WebsocketClientTransport create(InetSocketAddress address) { public static WebsocketClientTransport create(URI uri) { Objects.requireNonNull(uri, "uri must not be null"); - HttpClient httpClient = createClient(uri); - return create(httpClient, uri.getPath()); + TcpClient client = createClient(uri); + return create(HttpClient.from(client), uri.getPath()); + } + + /** + * Creates a new instance + * + * @param client the {@link TcpClient} to use + * @return a new instance + * @throws NullPointerException if {@code client} or {@code path} is {@code null} + */ + public static WebsocketClientTransport create(TcpClient client) { + Objects.requireNonNull(client, "client must not be null"); + + return create(HttpClient.from(client), "/"); } /** @@ -120,21 +137,12 @@ public static WebsocketClientTransport create(HttpClient client, String path) { @Override public Mono connect() { - return Mono.create( - sink -> - client - .ws(path, hb -> transportHeaders.get().forEach(hb::set)) - .flatMap( - response -> - response.receiveWebsocket( - (in, out) -> { - WebsocketDuplexConnection connection = - new WebsocketDuplexConnection(in, out, in.context()); - sink.success(connection); - return connection.onClose(); - })) - .doOnError(sink::error) - .subscribe()); + return client + .headers(headers -> transportHeaders.get().forEach(headers::set)) + .websocket() + .uri(path) + .connect() + .map(WebsocketDuplexConnection::new); } @Override @@ -143,16 +151,16 @@ public void setTransportHeaders(Supplier> transportHeaders) Objects.requireNonNull(transportHeaders, "transportHeaders must not be null"); } - private static HttpClient createClient(URI uri) { + private static TcpClient createClient(URI uri) { if (isSecure(uri)) { - return HttpClient.create( - options -> - options - .sslSupport() - .connectAddress( - () -> InetSocketAddress.createUnresolved(uri.getHost(), getPort(uri, 443)))); + return TcpClient.create() + .secure() + .host(uri.getHost()) + .port(getPort(uri, 443)); } else { - return HttpClient.create(uri.getHost(), getPort(uri, 80)); + return TcpClient.create() + .host(uri.getHost()) + .port(getPort(uri, 80)); } } } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/NettyContextCloseable.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/CloseableChannel.java similarity index 60% rename from rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/NettyContextCloseable.java rename to rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/CloseableChannel.java index ff31bf918..639100b06 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/NettyContextCloseable.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/CloseableChannel.java @@ -20,48 +20,48 @@ import java.net.InetSocketAddress; import java.util.Objects; import reactor.core.publisher.Mono; -import reactor.ipc.netty.NettyContext; +import reactor.netty.DisposableChannel; /** - * An implementation of {@link Closeable} that wraps a {@link NettyContext}, enabling close-ability - * and exposing the {@link NettyContext}'s address. + * An implementation of {@link Closeable} that wraps a {@link DisposableChannel}, enabling close-ability + * and exposing the {@link DisposableChannel}'s address. */ -public final class NettyContextCloseable implements Closeable { +public final class CloseableChannel implements Closeable { - private NettyContext context; + private DisposableChannel channel; /** * Creates a new instance * - * @param context the {@link NettyContext} to wrap + * @param channel the {@link DisposableChannel} to wrap * @throws NullPointerException if {@code context} is {@code null} */ - NettyContextCloseable(NettyContext context) { - this.context = Objects.requireNonNull(context, "context must not be null"); + CloseableChannel(DisposableChannel channel) { + this.channel = Objects.requireNonNull(channel, "channel must not be null"); } /** - * Returns the address that the {@link NettyContext} is listening on. + * Return local server selector channel address. * - * @return the address that the {@link NettyContext} is listening on - * @see NettyContext#address() + * @return local {@link InetSocketAddress} + * @see DisposableChannel#address() */ public InetSocketAddress address() { - return context.address(); + return channel.address(); } @Override public void dispose() { - context.dispose(); + channel.dispose(); } @Override public boolean isDisposed() { - return context.isDisposed(); + return channel.isDisposed(); } @Override public Mono onClose() { - return context.onClose(); + return channel.onDispose(); } } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/TcpServerTransport.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/TcpServerTransport.java index 18d9ddb32..dcf1476a7 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/TcpServerTransport.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/TcpServerTransport.java @@ -20,15 +20,17 @@ import io.rsocket.transport.ServerTransport; import io.rsocket.transport.netty.RSocketLengthCodec; import io.rsocket.transport.netty.TcpDuplexConnection; + import java.net.InetSocketAddress; import java.util.Objects; + import reactor.core.publisher.Mono; -import reactor.ipc.netty.tcp.TcpServer; +import reactor.netty.tcp.TcpServer; /** * An implementation of {@link ServerTransport} that connects to a {@link ClientTransport} via TCP. */ -public final class TcpServerTransport implements ServerTransport { +public final class TcpServerTransport implements ServerTransport { private final TcpServer server; @@ -43,7 +45,7 @@ private TcpServerTransport(TcpServer server) { * @return a new instance */ public static TcpServerTransport create(int port) { - TcpServer server = TcpServer.create(port); + TcpServer server = TcpServer.create().port(port); return create(server); } @@ -58,7 +60,7 @@ public static TcpServerTransport create(int port) { public static TcpServerTransport create(String bindAddress, int port) { Objects.requireNonNull(bindAddress, "bindAddress must not be null"); - TcpServer server = TcpServer.create(bindAddress, port); + TcpServer server = TcpServer.create().host(bindAddress).port(port); return create(server); } @@ -72,8 +74,7 @@ public static TcpServerTransport create(String bindAddress, int port) { public static TcpServerTransport create(InetSocketAddress address) { Objects.requireNonNull(address, "address must not be null"); - TcpServer server = TcpServer.create(address.getHostName(), address.getPort()); - return create(server); + return create(address.getHostName(), address.getPort()); } /** @@ -90,16 +91,19 @@ public static TcpServerTransport create(TcpServer server) { } @Override - public Mono start(ConnectionAcceptor acceptor) { + public Mono start(ConnectionAcceptor acceptor) { Objects.requireNonNull(acceptor, "acceptor must not be null"); return server - .newHandler( - (in, out) -> { - in.context().addHandler(new RSocketLengthCodec()); - TcpDuplexConnection connection = new TcpDuplexConnection(in, out, in.context()); - return acceptor.apply(connection).then(out.neverComplete()); - }) - .map(NettyContextCloseable::new); + .doOnConnection(c -> { + c.addHandlerLast(new RSocketLengthCodec()); + TcpDuplexConnection connection = new TcpDuplexConnection(c); + acceptor + .apply(connection) + .then(Mono.never()) + .subscribe(c.disposeSubscriber()); + }) + .bind() + .map(CloseableChannel::new); } } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/WebsocketRouteTransport.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/WebsocketRouteTransport.java index 2bfd15d5f..27bafd3de 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/WebsocketRouteTransport.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/WebsocketRouteTransport.java @@ -24,10 +24,11 @@ import java.util.function.Consumer; import org.reactivestreams.Publisher; import reactor.core.publisher.Mono; -import reactor.ipc.netty.http.server.HttpServer; -import reactor.ipc.netty.http.server.HttpServerRoutes; -import reactor.ipc.netty.http.websocket.WebsocketInbound; -import reactor.ipc.netty.http.websocket.WebsocketOutbound; +import reactor.netty.Connection; +import reactor.netty.http.server.HttpServer; +import reactor.netty.http.server.HttpServerRoutes; +import reactor.netty.http.websocket.WebsocketInbound; +import reactor.netty.http.websocket.WebsocketOutbound; /** * An implementation of {@link ServerTransport} that connects via Websocket and listens on specified @@ -61,12 +62,13 @@ public Mono start(ConnectionAcceptor acceptor) { Objects.requireNonNull(acceptor, "acceptor must not be null"); return server - .newRouter( + .route( routes -> { routesBuilder.accept(routes); routes.ws(path, newHandler(acceptor)); }) - .map(NettyContextCloseable::new); + .bind() + .map(CloseableChannel::new); } /** @@ -82,7 +84,7 @@ static BiFunction> newHandl Objects.requireNonNull(acceptor, "acceptor must not be null"); return (in, out) -> { - WebsocketDuplexConnection connection = new WebsocketDuplexConnection(in, out, in.context()); + WebsocketDuplexConnection connection = new WebsocketDuplexConnection((Connection)in); return acceptor.apply(connection).then(out.neverComplete()); }; } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/WebsocketServerTransport.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/WebsocketServerTransport.java index be6f5b04e..b6ef5eaea 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/WebsocketServerTransport.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/WebsocketServerTransport.java @@ -25,14 +25,14 @@ import java.util.Objects; import java.util.function.Supplier; import reactor.core.publisher.Mono; -import reactor.ipc.netty.http.server.HttpServer; +import reactor.netty.http.server.HttpServer; /** * An implementation of {@link ServerTransport} that connects to a {@link ClientTransport} via a * Websocket. */ public final class WebsocketServerTransport - implements ServerTransport, TransportHeaderAware { + implements ServerTransport, TransportHeaderAware { private final HttpServer server; @@ -49,7 +49,7 @@ private WebsocketServerTransport(HttpServer server) { * @return a new instance */ public static WebsocketServerTransport create(int port) { - HttpServer httpServer = HttpServer.create(port); + HttpServer httpServer = HttpServer.create().port(port); return create(httpServer); } @@ -64,7 +64,7 @@ public static WebsocketServerTransport create(int port) { public static WebsocketServerTransport create(String bindAddress, int port) { Objects.requireNonNull(bindAddress, "bindAddress must not be null"); - HttpServer httpServer = HttpServer.create(bindAddress, port); + HttpServer httpServer = HttpServer.create().host(bindAddress).port(port); return create(httpServer); } @@ -101,15 +101,16 @@ public void setTransportHeaders(Supplier> transportHeaders) } @Override - public Mono start(ConnectionAcceptor acceptor) { + public Mono start(ConnectionAcceptor acceptor) { Objects.requireNonNull(acceptor, "acceptor must not be null"); return server - .newHandler( + .handle( (request, response) -> { transportHeaders.get().forEach(response::addHeader); return response.sendWebsocket(WebsocketRouteTransport.newHandler(acceptor)); }) - .map(NettyContextCloseable::new); + .bind() + .map(CloseableChannel::new); } } diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/SetupRejectionTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/SetupRejectionTest.java index c3234a0c8..b524af7a1 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/SetupRejectionTest.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/SetupRejectionTest.java @@ -9,7 +9,7 @@ import io.rsocket.transport.ServerTransport; import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.transport.netty.client.WebsocketClientTransport; -import io.rsocket.transport.netty.server.NettyContextCloseable; +import io.rsocket.transport.netty.server.CloseableChannel; import io.rsocket.transport.netty.server.TcpServerTransport; import io.rsocket.transport.netty.server.WebsocketServerTransport; import io.rsocket.util.DefaultPayload; @@ -34,14 +34,14 @@ public class SetupRejectionTest { @ParameterizedTest @MethodSource(value = "transports") void rejectSetupTcp( - Function> serverTransport, + Function> serverTransport, Function clientTransport) { String errorMessage = "error"; RejectingAcceptor acceptor = new RejectingAcceptor(errorMessage); Mono serverRequester = acceptor.requesterRSocket(); - NettyContextCloseable nettyCtx = + CloseableChannel channel = RSocketFactory.receive() .acceptor(acceptor) .transport(serverTransport.apply(new InetSocketAddress(0))) @@ -53,7 +53,7 @@ void rejectSetupTcp( RSocket clientRequester = RSocketFactory.connect() .errorConsumer(errorConsumer) - .transport(clientTransport.apply(nettyCtx.address())) + .transport(clientTransport.apply(channel.address())) .start() .block(); @@ -73,13 +73,13 @@ void rejectSetupTcp( err -> err instanceof RejectedSetupException && errorMessage.equals(err.getMessage())) .verify(Duration.ofSeconds(5)); - nettyCtx.dispose(); + channel.dispose(); } static Stream transports() { - Function> tcpServer = + Function> tcpServer = TcpServerTransport::create; - Function> wsServer = + Function> wsServer = WebsocketServerTransport::create; Function tcpClient = TcpClientTransport::create; Function wsClient = WebsocketClientTransport::create; diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/WebsocketSecureTransportTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/WebsocketSecureTransportTest.java index 2afe8b1af..8b54d7189 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/WebsocketSecureTransportTest.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/WebsocketSecureTransportTest.java @@ -16,14 +16,20 @@ package io.rsocket.transport.netty; +import io.netty.handler.ssl.SslContextBuilder; import io.netty.handler.ssl.util.InsecureTrustManagerFactory; +import io.netty.handler.ssl.util.SelfSignedCertificate; import io.rsocket.test.TransportTest; import io.rsocket.transport.netty.client.WebsocketClientTransport; import io.rsocket.transport.netty.server.WebsocketServerTransport; import java.net.InetSocketAddress; +import java.security.cert.CertificateException; import java.time.Duration; -import reactor.ipc.netty.http.client.HttpClient; -import reactor.ipc.netty.http.server.HttpServer; + +import reactor.core.Exceptions; +import reactor.netty.http.client.HttpClient; +import reactor.netty.http.server.HttpServer; +import reactor.netty.tcp.TcpServer; final class WebsocketSecureTransportTest implements TransportTest { @@ -32,18 +38,27 @@ final class WebsocketSecureTransportTest implements TransportTest { () -> new InetSocketAddress("localhost", 0), (address, server) -> WebsocketClientTransport.create( - HttpClient.create( - options -> - options - .connectAddress(server::address) - .sslSupport( - c -> c.trustManager(InsecureTrustManagerFactory.INSTANCE))), + HttpClient.create() + .addressSupplier(server::address) + .secure(ssl -> ssl.sslContext( + SslContextBuilder.forClient() + .trustManager(InsecureTrustManagerFactory.INSTANCE))), String.format( "https://%s:%d/", server.address().getHostName(), server.address().getPort())), - address -> - WebsocketServerTransport.create( - HttpServer.create(options -> options.listenAddress(address).sslSelfSigned()))); + address -> { + try { + SelfSignedCertificate ssc = new SelfSignedCertificate(); + HttpServer server = HttpServer.from( + TcpServer.create() + .addressSupplier(() -> address) + .secure(ssl -> ssl.sslContext( + SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey())))); + return WebsocketServerTransport.create(server); + } catch (CertificateException e) { + throw Exceptions.propagate(e); + } + }); @Override public Duration getTimeout() { diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/client/TcpClientTransportTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/client/TcpClientTransportTest.java index a0954e707..388001fb6 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/client/TcpClientTransportTest.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/client/TcpClientTransportTest.java @@ -24,7 +24,7 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; -import reactor.ipc.netty.tcp.TcpClient; +import reactor.netty.tcp.TcpClient; import reactor.test.StepVerifier; final class TcpClientTransportTest { diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/client/WebsocketClientTransportTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/client/WebsocketClientTransportTest.java index 1fcea3049..202c5b3f3 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/client/WebsocketClientTransportTest.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/client/WebsocketClientTransportTest.java @@ -26,7 +26,7 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; -import reactor.ipc.netty.http.client.HttpClient; +import reactor.netty.http.client.HttpClient; import reactor.test.StepVerifier; final class WebsocketClientTransportTest { diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/NettyContextCloseableTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/CloseableChannelTest.java similarity index 77% rename from rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/NettyContextCloseableTest.java rename to rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/CloseableChannelTest.java index 48ffcb2d4..0e03317c7 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/NettyContextCloseableTest.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/CloseableChannelTest.java @@ -23,21 +23,21 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; -import reactor.ipc.netty.NettyContext; -import reactor.ipc.netty.tcp.TcpServer; +import reactor.netty.DisposableChannel; +import reactor.netty.tcp.TcpServer; import reactor.test.StepVerifier; -final class NettyContextCloseableTest { +final class CloseableChannelTest { - private final Mono nettyContext = - TcpServer.create().newHandler((in, out) -> Mono.empty()); + private final Mono channel = + TcpServer.create().handle((in, out) -> Mono.empty()).bind(); @DisplayName("returns the address of the context") @Test void address() { - nettyContext - .map(NettyContextCloseable::new) - .map(NettyContextCloseable::address) + channel + .map(CloseableChannel::new) + .map(CloseableChannel::address) .as(StepVerifier::create) .expectNextCount(1) .verifyComplete(); @@ -46,8 +46,8 @@ void address() { @DisplayName("creates instance") @Test void constructor() { - nettyContext - .map(NettyContextCloseable::new) + channel + .map(CloseableChannel::new) .as(StepVerifier::create) .expectNextCount(1) .verifyComplete(); @@ -57,8 +57,8 @@ void constructor() { @Test void constructorNullContext() { assertThatNullPointerException() - .isThrownBy(() -> new NettyContextCloseable(null)) - .withMessage("context must not be null"); + .isThrownBy(() -> new CloseableChannel(null)) + .withMessage("channel must not be null"); } @Disabled( @@ -67,8 +67,8 @@ void constructorNullContext() { @DisplayName("disposes context") @Test void dispose() { - nettyContext - .map(NettyContextCloseable::new) + channel + .map(CloseableChannel::new) .delayUntil( closeable -> { closeable.dispose(); diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/TcpServerTransportTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/TcpServerTransportTest.java index d5adc7ab9..15a216b96 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/TcpServerTransportTest.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/TcpServerTransportTest.java @@ -23,7 +23,7 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; -import reactor.ipc.netty.tcp.TcpServer; +import reactor.netty.tcp.TcpServer; import reactor.test.StepVerifier; final class TcpServerTransportTest { diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketRouteTransportTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketRouteTransportTest.java index 3e4d2c9a0..66822890a 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketRouteTransportTest.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketRouteTransportTest.java @@ -22,7 +22,7 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; -import reactor.ipc.netty.http.server.HttpServer; +import reactor.netty.http.server.HttpServer; import reactor.test.StepVerifier; final class WebsocketRouteTransportTest { @@ -75,7 +75,7 @@ void newHandlerNullAcceptor() { @Test void start() { WebsocketRouteTransport serverTransport = - new WebsocketRouteTransport(HttpServer.create(0), routes -> {}, "/test-path"); + new WebsocketRouteTransport(HttpServer.create(), routes -> {}, "/test-path"); serverTransport .start(duplexConnection -> Mono.empty()) @@ -90,7 +90,7 @@ void startNullAcceptor() { assertThatNullPointerException() .isThrownBy( () -> - new WebsocketRouteTransport(HttpServer.create(0), routes -> {}, "/test-path") + new WebsocketRouteTransport(HttpServer.create(), routes -> {}, "/test-path") .start(null)) .withMessage("acceptor must not be null"); } diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketServerTransportTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketServerTransportTest.java index 5213d27ec..d1a6b374e 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketServerTransportTest.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketServerTransportTest.java @@ -24,7 +24,7 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; -import reactor.ipc.netty.http.server.HttpServer; +import reactor.netty.http.server.HttpServer; import reactor.test.StepVerifier; final class WebsocketServerTransportTest {