RSocketRequester extends RequesterResponderSupport implements RSocket { @Override public Flux<Payload> requestChannel(Publisher<Payload> payloads) { return new RequestChannelRequesterFlux(payloads, 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 public void testCorrectFrameOrder() { MonoProcessor<Object> delayer = MonoProcessor.create(); BaseSubscriber<Payload> subscriber = new BaseSubscriber<Payload>() { @Override protected void hookOnSubscribe(Subscription subscription) {} }; rule.socket .requestChannel( Flux.concat(Flux.just(0).delayUntil(i -> delayer), Flux.range(1, 999)) .map(i -> DefaultPayload.create(i + ""))) .subscribe(subscriber); subscriber.request(1); subscriber.request(Long.MAX_VALUE); delayer.onComplete(); Iterator<ByteBuf> iterator = rule.connection.getSent().iterator(); ByteBuf initialFrame = iterator.next(); Assertions.assertThat(FrameHeaderCodec.frameType(initialFrame)).isEqualTo(REQUEST_CHANNEL); Assertions.assertThat(RequestChannelFrameCodec.initialRequestN(initialFrame)) .isEqualTo(Long.MAX_VALUE); Assertions.assertThat(RequestChannelFrameCodec.data(initialFrame).toString(CharsetUtil.UTF_8)) .isEqualTo("0"); Assertions.assertThat(initialFrame.release()).isTrue(); Assertions.assertThat(iterator.hasNext()).isFalse(); rule.assertHasNoLeaks(); }
@Test public void simpleOnDiscardRequestChannelTest() { AssertSubscriber<Payload> assertSubscriber = AssertSubscriber.create(1); TestPublisher<Payload> testPublisher = TestPublisher.create(); Flux<Payload> payloadFlux = rule.socket.requestChannel(testPublisher); payloadFlux.subscribe(assertSubscriber); testPublisher.next( ByteBufPayload.create("d", "m"), ByteBufPayload.create("d1", "m1"), ByteBufPayload.create("d2", "m2")); assertSubscriber.cancel(); Assertions.assertThat(rule.connection.getSent()).allMatch(ByteBuf::release); rule.assertHasNoLeaks(); }
@Test public void simpleOnDiscardRequestChannelTest2() { ByteBufAllocator allocator = rule.alloc(); AssertSubscriber<Payload> assertSubscriber = AssertSubscriber.create(1); TestPublisher<Payload> testPublisher = TestPublisher.create(); Flux<Payload> payloadFlux = rule.socket.requestChannel(testPublisher); payloadFlux.subscribe(assertSubscriber); testPublisher.next(ByteBufPayload.create("d", "m")); int streamId = rule.getStreamIdForRequestType(REQUEST_CHANNEL); testPublisher.next(ByteBufPayload.create("d1", "m1"), ByteBufPayload.create("d2", "m2")); rule.connection.addToReceivedBuffer( ErrorFrameCodec.encode( allocator, streamId, new CustomRSocketException(0x00000404, "test"))); Assertions.assertThat(rule.connection.getSent()).allMatch(ByteBuf::release); rule.assertHasNoLeaks(); }