@Test public void testFireForget() { Payload incoming = Mockito.mock(Payload.class); Payload transformed = ByteBufPayload.create("transformed"); RSocket mock = Mockito.mock(RSocket.class); Mockito.when(mock.fireAndForget(Mockito.any(Payload.class))).thenReturn(Mono.empty()); DefaultProteusSocket socket = new DefaultProteusSocket(payload -> transformed, () -> mock); socket.fireAndForget(incoming).block(); Mockito.verify(incoming, Mockito.times(0)).release(); }
@Override public Mono<Void> fireAndForget(Payload payload) { return Mono.defer( () -> { Payload transformedPayload = payloadTransformer.apply(payload); if (transformedPayload != null && payload.refCnt() > 0) { quietRelease(payload); } return rSocketSupplier.get().fireAndForget(transformedPayload); }); }
DefaultProteusSocket implements ProteusSocket { @Override public Mono<Void> fireAndForget(Payload payload) { return Mono.defer( () -> { Payload transformedPayload = payloadTransformer.apply(payload); if (transformedPayload != null && payload.refCnt() > 0) { quietRelease(payload); } return rSocketSupplier.get().fireAndForget(transformedPayload); }); } }
DefaultProteusSocket implements ProteusSocket { @Override public Mono<Void> fireAndForget(Payload payload) { return Mono.defer( () -> { Payload transformedPayload = payloadTransformer.apply(payload); if (transformedPayload != null && payload.refCnt() > 0) { quietRelease(payload); } return rSocketSupplier.get().fireAndForget(transformedPayload); }); } DefaultProteusSocket(
      Function<Payload, Payload> payloadTransformer, Supplier<RSocket> rSocketSupplier); }
DefaultProteusSocket implements ProteusSocket { @Override public Mono<Void> fireAndForget(Payload payload) { return Mono.defer( () -> { Payload transformedPayload = payloadTransformer.apply(payload); if (transformedPayload != null && payload.refCnt() > 0) { quietRelease(payload); } return rSocketSupplier.get().fireAndForget(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(); }
DefaultProteusSocket implements ProteusSocket { @Override public Mono<Void> fireAndForget(Payload payload) { return Mono.defer( () -> { Payload transformedPayload = payloadTransformer.apply(payload); if (transformedPayload != null && payload.refCnt() > 0) { quietRelease(payload); } return rSocketSupplier.get().fireAndForget(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();  }