DefaultProteusSocket implements ProteusSocket { @Override public Flux<Payload> requestStream(Payload payload) { return Flux.defer( () -> { Payload transformedPayload = payloadTransformer.apply(payload); if (transformedPayload != null) { quietRelease(payload); } return rSocketSupplier.get().requestStream(transformedPayload); }); } 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 testRequestStream() { Payload incoming = Mockito.mock(Payload.class); Payload transformed = ByteBufPayload.create("transformed"); Payload outgoing = ByteBufPayload.create("outgoing"); RSocket mock = Mockito.mock(RSocket.class); Mockito.when(mock.requestStream(Mockito.any(Payload.class))).thenReturn(Flux.just(outgoing)); DefaultProteusSocket socket = new DefaultProteusSocket(payload -> transformed, () -> mock); Payload block = socket.requestStream(incoming).blockLast(); Assert.assertTrue(block == outgoing); Mockito.verify(incoming, Mockito.times(0)).release(); }