RSocketRequester extends RequesterResponderSupport implements RSocket { @Override public Flux<Payload> requestStream(Payload payload) { return new RequestStreamRequesterFlux(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 testStreamInitialN() { Flux<Payload> stream = rule.socket.requestStream(EmptyPayload.INSTANCE); BaseSubscriber<Payload> subscriber = new BaseSubscriber<Payload>() { @Override protected void hookOnSubscribe(Subscription subscription) { } }; stream.subscribe(subscriber); Assertions.assertThat(rule.connection.getSent()).isEmpty(); subscriber.request(5); List<ByteBuf> sent = new ArrayList<>(rule.connection.getSent()); assertThat("sent frame count", sent.size(), is(1)); ByteBuf f = sent.get(0); assertThat("initial frame", frameType(f), is(REQUEST_STREAM)); assertThat("initial request n", RequestStreamFrameCodec.initialRequestN(f), is(5L)); assertThat("should be released", f.release(), is(true)); rule.assertHasNoLeaks(); }