fixes extra Payload release on racing complete and cancel (#894)

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

Author: OlegDokuka

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

@@ -315,7 +315,9 @@ void hookOnCancel() {
                               if (receivers.remove(streamId, receiver)) {
                                 sendProcessor.onNext(CancelFrameCodec.encode(allocator, streamId));
                               } else {
-                                payload.release();
+                                if (this.firstRequest) {
+                                  payload.release();
+                                }
                               }
                             }
 
@@ -405,7 +407,9 @@ void hookOnCancel() {
                               if (receivers.remove(streamId, receiver)) {
                                 sendProcessor.onNext(CancelFrameCodec.encode(allocator, streamId));
                               } else {
-                                payload.release();
+                                if (this.firstRequest) {
+                                  payload.release();
+                                }
                               }
                             }
 

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

@@ -76,11 +76,12 @@ public void request(long n) {
       }
       return;
     }
-    this.firstRequest = false;
 
     if (WIP.getAndIncrement(this) != 0) {
       return;
     }
+
+    this.firstRequest = false;
     int missed = 1;
 
     boolean firstLoop = true;

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

@@ -616,7 +616,15 @@ private static Stream<Arguments> racingCases() {
                 }),
         Arguments.of(
             (Function<ClientSocketRule, Publisher<Payload>>)
-                (rule) -> rule.socket.requestResponse(EmptyPayload.INSTANCE),
+                (rule) -> {
+                  ByteBuf data = rule.allocator.buffer();
+                  data.writeCharSequence("testData", CharsetUtil.UTF_8);
+
+                  ByteBuf metadata = rule.allocator.buffer();
+                  metadata.writeCharSequence("testMetadata", CharsetUtil.UTF_8);
+                  Payload requestPayload = ByteBufPayload.create(data, metadata);
+                  return rule.socket.requestResponse(requestPayload);
+                },
             (BiConsumer<AssertSubscriber<Payload>, ClientSocketRule>)
                 (as, rule) -> {
                   ByteBufAllocator allocator = rule.alloc();
@@ -630,6 +638,32 @@ private static Stream<Arguments> racingCases() {
                       PayloadFrameCodec.encode(
                           allocator, streamId, false, false, true, metadata, data);
 
+                  RaceTestUtils.race(as::cancel, () -> rule.connection.addToReceivedBuffer(frame));
+                }),
+        Arguments.of(
+            (Function<ClientSocketRule, Publisher<Payload>>)
+                (rule) -> {
+                  ByteBuf data = rule.allocator.buffer();
+                  data.writeCharSequence("testData", CharsetUtil.UTF_8);
+
+                  ByteBuf metadata = rule.allocator.buffer();
+                  metadata.writeCharSequence("testMetadata", CharsetUtil.UTF_8);
+                  Payload requestPayload = ByteBufPayload.create(data, metadata);
+                  return rule.socket.requestStream(requestPayload);
+                },
+            (BiConsumer<AssertSubscriber<Payload>, ClientSocketRule>)
+                (as, rule) -> {
+                  ByteBufAllocator allocator = rule.alloc();
+                  ByteBuf metadata = allocator.buffer();
+                  metadata.writeCharSequence("abc", CharsetUtil.UTF_8);
+                  ByteBuf data = allocator.buffer();
+                  data.writeCharSequence("def", CharsetUtil.UTF_8);
+                  as.request(Long.MAX_VALUE);
+                  int streamId = rule.getStreamIdForRequestType(REQUEST_STREAM);
+                  ByteBuf frame =
+                      PayloadFrameCodec.encode(
+                          allocator, streamId, false, true, true, metadata, data);
+
                   RaceTestUtils.race(as::cancel, () -> rule.connection.addToReceivedBuffer(frame));
                 }));
   }