fixes incorrect requestN behaviour on the requester side

Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java

@@ -317,7 +317,7 @@ private Flux<Payload> handleChannel(Flux<Payload> request) {
         (s, flux) -> {
           Payload payload = s.get();
           if (payload != null) {
-            return handleChannel(payload, flux.skip(1));
+            return handleChannel(payload, flux);
           } else {
             return flux;
           }
@@ -334,13 +334,20 @@ private Flux<? extends Payload> handleChannel(Payload initialPayload, Flux<Paylo
     final BaseSubscriber<Payload> upstreamSubscriber =
         new BaseSubscriber<Payload>() {
 
+          boolean first = true;
+
           @Override
           protected void hookOnSubscribe(Subscription subscription) {
             // noops
           }
 
           @Override
           protected void hookOnNext(Payload payload) {
+            if (first) {
+              // need to skip first since we have already sent it
+              first = false;
+              return;
+            }
             final ByteBuf frame =
                 PayloadFrameFlyweight.encode(allocator, streamId, false, false, true, payload);
 

rsocket-test/src/main/java/io/rsocket/test/TransportTest.java

@@ -229,7 +229,7 @@ default void requestChannel3() {
         .verify(getTimeout());
 
     Assertions.assertThat(requested.get())
-        .isEqualTo(257L); // 257 because of eager behavior of limitRate
+        .isEqualTo(256L); // 256 because of eager behavior of limitRate
   }
 
   @DisplayName("makes 1 requestChannel request with 512 payloads")