Skip to content

Commit 5d173aa

Browse files
yschimkeyschimke
authored
ephemeral port for Tcp server (rsocket#293)
returns a more useful NettyContextCloseable instead of Closeable
1 parent 90b504b commit 5d173aa

8 files changed

Lines changed: 51 additions & 32 deletions

File tree

‎rsocket-core/src/main/java/io/rsocket/transport/ServerTransport.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@
2626
/**
2727
* A server contract for writing transports of RSocket.
2828
*/
29-
public interface ServerTransport extends Transport {
29+
public interface ServerTransport<T extends Closeable> extends Transport {
3030

3131
/**
3232
* Starts this server.
@@ -35,7 +35,7 @@ public interface ServerTransport extends Transport {
3535
*
3636
* @return A handle to retrieve information about a started server.
3737
*/
38-
Mono<Closeable> start(ConnectionAcceptor acceptor);
38+
Mono<T> start(ConnectionAcceptor acceptor);
3939

4040
/**
4141
* A contract to accept a new {@code DuplexConnection}.

‎rsocket-examples/src/test/java/io/rsocket/integration/IntegrationTest.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,12 +18,12 @@
1818

1919
import io.reactivex.subscribers.TestSubscriber;
2020
import io.rsocket.AbstractRSocket;
21-
import io.rsocket.Closeable;
2221
import io.rsocket.Payload;
2322
import io.rsocket.Plugins;
2423
import io.rsocket.RSocket;
2524
import io.rsocket.RSocketFactory;
2625
import io.rsocket.transport.netty.client.TcpClientTransport;
26+
import io.rsocket.transport.netty.server.NettyContextCloseable;
2727
import io.rsocket.transport.netty.server.TcpServerTransport;
2828
import io.rsocket.util.PayloadImpl;
2929
import io.rsocket.util.RSocketProxy;
@@ -44,7 +44,7 @@
4444

4545
public class IntegrationTest {
4646

47-
private Closeable server;
47+
private NettyContextCloseable server;
4848
private RSocket client;
4949
private AtomicInteger requestCount;
5050
private CountDownLatch disconnectionCounter;
@@ -82,6 +82,8 @@ public void startup() {
8282
requestCount = new AtomicInteger();
8383
disconnectionCounter = new CountDownLatch(1);
8484

85+
TcpServerTransport serverTransport = TcpServerTransport.create(0);
86+
8587
server = RSocketFactory
8688
.receive()
8789
.acceptor((setup, sendingSocket) -> {
@@ -104,13 +106,15 @@ public Flux<Payload> requestStream(Payload payload) {
104106
}
105107
});
106108
})
107-
.transport(TcpServerTransport.create("localhost", 8000))
109+
.transport(serverTransport)
108110
.start()
111+
// TODO fix the Types through RSocketFactory.Start
112+
.cast(NettyContextCloseable.class)
109113
.block();
110114

111115
client = RSocketFactory
112116
.connect()
113-
.transport(TcpClientTransport.create("localhost", 8000))
117+
.transport(TcpClientTransport.create(server.address()))
114118
.start()
115119
.block();
116120
}

‎rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import io.rsocket.RSocket;
77
import io.rsocket.RSocketFactory;
88
import io.rsocket.transport.netty.client.TcpClientTransport;
9+
import io.rsocket.transport.netty.server.NettyContextCloseable;
910
import io.rsocket.transport.netty.server.TcpServerTransport;
1011
import io.rsocket.util.PayloadImpl;
1112
import io.rsocket.util.RSocketProxy;
@@ -27,22 +28,26 @@
2728
public class TcpIntegrationTest {
2829
private AbstractRSocket handler;
2930

30-
private Closeable server;
31+
private NettyContextCloseable server;
3132

3233
@Before
3334
public void startup() {
34-
server = RSocketFactory
35+
TcpServerTransport serverTransport = TcpServerTransport.create(0);
36+
RSocketFactory.Start<Closeable> transport = RSocketFactory
3537
.receive()
3638
.acceptor((setup, sendingSocket) -> Mono.just(new RSocketProxy(handler)))
37-
.transport(TcpServerTransport.create("localhost", 8000))
39+
.transport(serverTransport);
40+
server = transport
3841
.start()
42+
// TODO fix the Types through RSocketFactory.Start
43+
.cast(NettyContextCloseable.class)
3944
.block();
4045
}
4146

4247
private RSocket buildClient() {
4348
return RSocketFactory
4449
.connect()
45-
.transport(TcpClientTransport.create("localhost", 8000))
50+
.transport(TcpClientTransport.create(server.address()))
4651
.start()
4752
.block();
4853
}

‎rsocket-transport-local/src/main/java/io/rsocket/transport/local/LocalServerTransport.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
import java.util.concurrent.ConcurrentMap;
1111
import java.util.function.Consumer;
1212

13-
public class LocalServerTransport implements ServerTransport {
13+
public class LocalServerTransport implements ServerTransport<Closeable> {
1414
private static final ConcurrentMap<String, ServerDuplexConnectionAcceptor> registry = new ConcurrentHashMap<>();
1515

1616
static ServerDuplexConnectionAcceptor findServer(String name) {

‎rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/client/TcpClientTransport.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import io.rsocket.transport.ClientTransport;
2121
import io.rsocket.transport.netty.RSocketLengthCodec;
2222
import io.rsocket.transport.netty.NettyDuplexConnection;
23+
import java.net.InetSocketAddress;
2324
import org.slf4j.Logger;
2425
import org.slf4j.LoggerFactory;
2526
import reactor.core.publisher.Mono;
@@ -39,11 +40,16 @@ public static TcpClientTransport create(int port) {
3940

4041
}
4142

42-
public static TcpClientTransport create(String bindAddress, int port) {
43+
public static TcpClientTransport create(String bindAddress, int port) {
4344
TcpClient tcpClient = TcpClient.create(bindAddress, port);
4445
return create(tcpClient);
4546
}
4647

48+
public static TcpClientTransport create(InetSocketAddress address) {
49+
TcpClient tcpClient = TcpClient.create(address.getHostString(), address.getPort());
50+
return create(tcpClient);
51+
}
52+
4753
public static TcpClientTransport create(TcpClient client) {
4854
return new TcpClientTransport(client);
4955
}

rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/NettyContextClosable.java renamed to rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/NettyContextCloseable.java

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,21 @@
11
package io.rsocket.transport.netty.server;
22

33
import io.rsocket.Closeable;
4+
import java.net.InetSocketAddress;
45
import reactor.core.publisher.Mono;
56
import reactor.core.publisher.MonoProcessor;
67
import reactor.ipc.netty.NettyContext;
78

89
/**
9-
*
10+
* A {@link Closeable} wrapping a {@link NettyContext}, allowing for close and
11+
* aware of its address.
1012
*/
11-
class NettyContextClosable implements Closeable {
13+
public class NettyContextCloseable implements Closeable {
1214
private NettyContext context;
1315

1416
private MonoProcessor<Void> onClose;
1517

16-
NettyContextClosable(NettyContext context) {
18+
NettyContextCloseable(NettyContext context) {
1719
this.onClose = MonoProcessor.create();
1820
this.context = context;
1921
}
@@ -33,4 +35,11 @@ public Mono<Void> close() {
3335
public Mono<Void> onClose() {
3436
return onClose;
3537
}
38+
39+
/**
40+
* @see NettyContext#address()
41+
*/
42+
public InetSocketAddress address() {
43+
return context.address();
44+
}
3645
}

‎rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/TcpServerTransport.java‎

Lines changed: 9 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,13 @@
1616

1717
package io.rsocket.transport.netty.server;
1818

19-
import io.rsocket.Closeable;
2019
import io.rsocket.transport.ServerTransport;
2120
import io.rsocket.transport.netty.NettyDuplexConnection;
2221
import io.rsocket.transport.netty.RSocketLengthCodec;
2322
import reactor.core.publisher.Mono;
2423
import reactor.ipc.netty.tcp.TcpServer;
2524

26-
public class TcpServerTransport implements ServerTransport {
25+
public class TcpServerTransport implements ServerTransport<NettyContextCloseable> {
2726
TcpServer server;
2827

2928
private TcpServerTransport(TcpServer server) {
@@ -46,16 +45,13 @@ public static TcpServerTransport create(TcpServer server) {
4645
}
4746

4847
@Override
49-
public Mono<Closeable> start(ConnectionAcceptor acceptor) {
50-
return server
51-
.newHandler((in, out) -> {
52-
in.context().addHandler("server-length-codec", new RSocketLengthCodec());
53-
NettyDuplexConnection connection = new NettyDuplexConnection(in, out, in.context());
54-
acceptor.apply(connection).subscribe();
55-
56-
return out.neverComplete();
57-
})
58-
.map(NettyContextClosable::new);
48+
public Mono<NettyContextCloseable> start(ConnectionAcceptor acceptor) {
49+
return server.newHandler((in, out) -> {
50+
in.context().addHandler("server-length-codec", new RSocketLengthCodec());
51+
NettyDuplexConnection connection = new NettyDuplexConnection(in, out, in.context());
52+
acceptor.apply(connection).subscribe();
53+
54+
return out.neverComplete();
55+
}).map(NettyContextCloseable::new);
5956
}
60-
6157
}

‎rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/WebsocketServerTransport.java‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,12 @@
1616

1717
package io.rsocket.transport.netty.server;
1818

19-
import io.rsocket.Closeable;
2019
import io.rsocket.transport.ServerTransport;
2120
import io.rsocket.transport.netty.WebsocketDuplexConnection;
2221
import reactor.core.publisher.Mono;
2322
import reactor.ipc.netty.http.server.HttpServer;
2423

25-
public class WebsocketServerTransport implements ServerTransport {
24+
public class WebsocketServerTransport implements ServerTransport<NettyContextCloseable> {
2625
HttpServer server;
2726

2827
private WebsocketServerTransport(HttpServer server) {
@@ -44,7 +43,7 @@ public static WebsocketServerTransport create(HttpServer server) {
4443
}
4544

4645
@Override
47-
public Mono<Closeable> start(ServerTransport.ConnectionAcceptor acceptor) {
46+
public Mono<NettyContextCloseable> start(ServerTransport.ConnectionAcceptor acceptor) {
4847
return server
4948
.newHandler((request, response) ->
5049
response.sendWebsocket((in, out) -> {
@@ -54,7 +53,7 @@ public Mono<Closeable> start(ServerTransport.ConnectionAcceptor acceptor) {
5453
return out.neverComplete();
5554
})
5655
)
57-
.map(NettyContextClosable::new);
56+
.map(NettyContextCloseable::new);
5857
}
5958

6059
}

0 commit comments

Comments
 (0)