fixes incorrect request propagation

Initially that issue was hidden because both sides uses limitRate which does prefetch 256 elements in advance so it is almost impossible to track underflow in request

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

Author: OlegDokuka

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

@@ -382,7 +382,10 @@ protected void hookOnSubscribe(Subscription subscription) {
           protected void hookOnNext(Payload payload) {
             if (first) {
               // need to skip first since we have already sent it
+              // no need to release it since it was released earlier on the request establishment
+              // phase
               first = false;
+              request(1);
               return;
             }
             if (!PayloadValidationUtils.isValid(mtu, payload)) {

rsocket-core/src/test/java/io/rsocket/core/RSocketTest.java

@@ -34,6 +34,7 @@
 import io.rsocket.test.util.TestSubscriber;
 import io.rsocket.util.DefaultPayload;
 import io.rsocket.util.EmptyPayload;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.concurrent.atomic.AtomicReference;
 import org.assertj.core.api.Assertions;
@@ -120,6 +121,32 @@ public Mono<Payload> requestResponse(Payload payload) {
     rule.assertServerError("CustomRSocketException (0x501): Deliberate Custom exception.");
   }
 
+  @Test(timeout = 2000)
+  public void testRequestPropagatesCorrectlyForRequestChannel() {
+    rule.setRequestAcceptor(
+        new AbstractRSocket() {
+          @Override
+          public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
+            return Flux.from(payloads)
+                // specifically limits request to 3 in order to prevent 256 request from limitRate
+                // hidden on the responder side
+                .limitRequest(3);
+          }
+        });
+
+    Flux.range(0, 3)
+        .map(i -> DefaultPayload.create("" + i))
+        .as(rule.crs::requestChannel)
+        .as(publisher -> StepVerifier.create(publisher, 3))
+        .expectSubscription()
+        .expectNextCount(3)
+        .expectComplete()
+        .verify(Duration.ofMillis(5000));
+
+    rule.assertNoClientErrors();
+    rule.assertNoServerErrors();
+  }
+
   @Test(timeout = 2000)
   public void testStream() throws Exception {
     Flux<Payload> responses = rule.crs.requestStream(DefaultPayload.create("Payload In"));