RSocketResponder extends RequesterResponderSupport implements RSocket { @Override public boolean isDisposed() { return getDuplexConnection().isDisposed(); } RSocketResponder(
      DuplexConnection connection,
      RSocket requestHandler,
      PayloadDecoder payloadDecoder,
      ResponderLeaseHandler leaseHandler,
      int mtu,
      int maxFrameLength,
      int maxInboundPayloadSize); @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 @Disabled("Reactor 3.4.0 should fix that. No need to do anything on our side") public void testWorkaround858() { ByteBuf buffer = rule.alloc().buffer(); buffer.writeCharSequence("test", CharsetUtil.UTF_8); TestPublisher<Payload> testPublisher = TestPublisher.create(); rule.setAcceptingSocket( new RSocket() { @Override public Flux<Payload> requestChannel(Publisher<Payload> payloads) { Flux.from(payloads).doOnNext(ReferenceCounted::release).subscribe(); return testPublisher.flux(); } }); rule.connection.addToReceivedBuffer( RequestChannelFrameCodec.encodeReleasingPayload( rule.alloc(), 1, false, 1, ByteBufPayload.create(buffer))); rule.connection.addToReceivedBuffer( ErrorFrameCodec.encode(rule.alloc(), 1, new RuntimeException("test"))); Assertions.assertThat(rule.connection.getSent()) .hasSize(1) .first() .matches(bb -> FrameHeaderCodec.frameType(bb) == REQUEST_N) .matches(ReferenceCounted::release); Assertions.assertThat(rule.socket.isDisposed()).isFalse(); testPublisher.assertWasCancelled(); rule.assertHasNoLeaks(); }