TcpServerTransport implements ServerTransport<CloseableChannel> { @Override public Mono<CloseableChannel> start(ConnectionAcceptor acceptor) { Objects.requireNonNull(acceptor, "acceptor must not be null"); return server .doOnConnection( c -> { c.addHandlerLast(new RSocketLengthCodec(maxFrameLength)); acceptor .apply(new TcpDuplexConnection(c)) .then(Mono.<Void>never()) .subscribe(c.disposeSubscriber()); }) .bind() .map(CloseableChannel::new); } private  TcpServerTransport(TcpServer server, int maxFrameLength); static TcpServerTransport create(int port); static TcpServerTransport create(String bindAddress, int port); static TcpServerTransport create(InetSocketAddress address); static TcpServerTransport create(TcpServer server); static TcpServerTransport create(TcpServer server, int maxFrameLength); @Override int maxFrameLength(); @Override Mono<CloseableChannel> start(ConnectionAcceptor acceptor);  }
@DisplayName("starts server") @Test void start() { InetSocketAddress address = InetSocketAddress.createUnresolved("localhost", 0); TcpServerTransport serverTransport = TcpServerTransport.create(address); serverTransport .start(duplexConnection -> Mono.empty()) .as(StepVerifier::create) .expectNextCount(1) .verifyComplete(); }