DefaultProteusSocket implements ProteusSocket { @Override public Flux<Payload> requestChannel(Publisher<Payload> payloads) { Flux<Payload> transformed = Flux.from(payloads) .map( payload -> { Payload transformedPayload = payloadTransformer.apply(payload); if (transformedPayload != null) { quietRelease(payload); } return transformedPayload; }); return rSocketSupplier.get().requestChannel(transformed); } DefaultProteusSocket(
      Function<Payload, Payload> payloadTransformer, Supplier<RSocket> rSocketSupplier); @Override Mono<Void> fireAndForget(Payload payload); @Override Mono<Payload> requestResponse(Payload payload); @Override Flux<Payload> requestStream(Payload payload); @Override Flux<Payload> requestChannel(Publisher<Payload> payloads); @Override Mono<Void> metadataPush(Payload payload); @Override void dispose(); @Override boolean isDisposed(); @Override Mono<Void> onClose();  }
@Test public void testRequestChannel() { Payload incoming = Mockito.mock(Payload.class); Payload transformed = ByteBufPayload.create("transformed"); Payload outgoing = ByteBufPayload.create("outgoing"); DefaultProteusSocket socket = new DefaultProteusSocket( payload -> transformed, () -> new AbstractRSocket() { @Override public Flux<Payload> requestChannel(Publisher<Payload> payloads) { return Flux.from(payloads).thenMany(Flux.just(outgoing)); } }); Payload block = socket.requestChannel(Mono.just(incoming)).blockLast(); Assert.assertTrue(block == outgoing); Mockito.verify(incoming, Mockito.times(0)).release(); }