RSocketRequester extends RequesterResponderSupport implements RSocket { @Override public Mono<Payload> requestResponse(Payload payload) { return new RequestResponseRequesterMono(payload, this); } RSocketRequester(
      DuplexConnection connection,
      PayloadDecoder payloadDecoder,
      StreamIdSupplier streamIdSupplier,
      int mtu,
      int maxFrameLength,
      int maxInboundPayloadSize,
      int keepAliveTickPeriod,
      int keepAliveAckTimeout,
      @Nullable KeepAliveHandler keepAliveHandler,
      RequesterLeaseHandler leaseHandler); @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 int getNextStreamId(); @Override int addAndGetNextStreamId(FrameHandler frameHandler); @Override double availability(); @Override void dispose(); @Override boolean isDisposed(); @Override Mono<Void> onClose();  }
@Test @Timeout(2_000) public void testHandleApplicationException() { rule.connection.clearSendReceiveBuffers(); Publisher<Payload> response = rule.socket.requestResponse(EmptyPayload.INSTANCE); Subscriber<Payload> responseSub = TestSubscriber.create(); response.subscribe(responseSub); int streamId = rule.getStreamIdForRequestType(REQUEST_RESPONSE); rule.connection.addToReceivedBuffer( ErrorFrameCodec.encode(rule.alloc(), streamId, new ApplicationErrorException("error"))); verify(responseSub).onError(any(ApplicationErrorException.class)); Assertions.assertThat(rule.connection.getSent()) .hasSize(1) .allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); }
@Test @Timeout(2_000) public void testHandleValidFrame() { Publisher<Payload> response = rule.socket.requestResponse(EmptyPayload.INSTANCE); Subscriber<Payload> sub = TestSubscriber.create(); response.subscribe(sub); int streamId = rule.getStreamIdForRequestType(REQUEST_RESPONSE); rule.connection.addToReceivedBuffer( PayloadFrameCodec.encodeNextReleasingPayload( rule.alloc(), streamId, EmptyPayload.INSTANCE)); verify(sub).onComplete(); Assertions.assertThat(rule.connection.getSent()).hasSize(1).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); }
@Test @Timeout(2_000) public void testRequestReplyWithCancel() { Mono<Payload> response = rule.socket.requestResponse(EmptyPayload.INSTANCE); try { response.block(Duration.ofMillis(100)); } catch (IllegalStateException ise) { } List<ByteBuf> sent = new ArrayList<>(rule.connection.getSent()); assertThat( "Unexpected frame sent on the connection.", frameType(sent.get(0)), is(REQUEST_RESPONSE)); assertThat("Unexpected frame sent on the connection.", frameType(sent.get(1)), is(CANCEL)); Assertions.assertThat(sent).hasSize(2).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); }