more improvements

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

Author: OlegDokuka

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

@@ -114,6 +114,13 @@ public void request(long n) {
       if (firstLoop) {
         firstLoop = false;
         try {
+          // since in all the scenarios where RequestOperator is used, the
+          // CorePublisher is either UnicastProcessor or UnicastProcessor.next()
+          // Therefore, we are free to propagate max demand immediately up on the first
+          // request. UnicastProcessor is only there to allow sending signals from the
+          // connection to a created stream and does not have to check the correctness
+          // of flow-control. For more info see
+          // https://github.com/rsocket/rsocket/blob/master/Protocol.md#handling-the-unexpected
           this.s.request(Long.MAX_VALUE);
           this.hookOnFirstRequest(n);
         } catch (Throwable throwable) {

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

@@ -1142,15 +1142,22 @@ public void testWorkaround858() {
     rule.assertHasNoLeaks();
   }
 
-  @Test
+  @ParameterizedTest
+  @ValueSource(strings = {"stream", "channel"})
   // see https://github.com/rsocket/rsocket-java/issues/959
-  public void testWorkaround959() {
-    for (int i = 1; i < 100000; i += 2) {
+  public void testWorkaround959(String type) {
+    for (int i = 1; i < 20000; i += 2) {
       ByteBuf buffer = rule.alloc().buffer();
       buffer.writeCharSequence("test", CharsetUtil.UTF_8);
 
       final AssertSubscriber<Payload> assertSubscriber = new AssertSubscriber<>(3);
-      rule.socket.requestStream(ByteBufPayload.create(buffer)).subscribe(assertSubscriber);
+      if (type.equals("stream")) {
+        rule.socket.requestStream(ByteBufPayload.create(buffer)).subscribe(assertSubscriber);
+      } else if (type.equals("channel")) {
+        rule.socket
+            .requestChannel(Flux.just(ByteBufPayload.create(buffer)))
+            .subscribe(assertSubscriber);
+      }
 
       final ByteBuf payloadFrame =
           PayloadFrameCodec.encode(