LocalClientTransport implements ClientTransport { @Override public Mono<DuplexConnection> connect() { return Mono.defer( () -> { ServerTransport.ConnectionAcceptor server = LocalServerTransport.findServer(name); if (server == null) { return Mono.error(new IllegalArgumentException("Could not find server: " + name)); } UnboundedProcessor in = new UnboundedProcessor(); UnboundedProcessor out = new UnboundedProcessor(); MonoProcessor<Void> closeNotifier = MonoProcessor.create(); server .apply(new LocalDuplexConnection(name, allocator, out, in, closeNotifier)) .subscribe(); return Mono.just( (DuplexConnection) new LocalDuplexConnection(name, allocator, in, out, closeNotifier)); }); } private  LocalClientTransport(String name, ByteBufAllocator allocator); static LocalClientTransport create(String name); static LocalClientTransport create(String name, ByteBufAllocator allocator); @Override Mono<DuplexConnection> connect();  }
@DisplayName("connects to server") @Test void connect() { LocalServerTransport serverTransport = LocalServerTransport.createEphemeral(); serverTransport .start(duplexConnection -> Mono.empty()) .flatMap(closeable -> LocalClientTransport.create(serverTransport.getName()).connect()) .as(StepVerifier::create) .expectNextCount(1) .verifyComplete(); }