fixes test

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

Author: OlegDokuka

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

@@ -514,6 +514,7 @@ public void accept(long l) {
                     sendProcessor.onNext(RequestNFrameFlyweight.encode(allocator, streamId, n));
                   }
                 })
+            .doFinally(signalType -> channelProcessors.remove(streamId))
             .doOnDiscard(ReferenceCounted.class, DROPPED_ELEMENTS_CONSUMER);
 
     // not chained, as the payload should be enqueued in the Unicast processor before this method

rsocket-core/src/main/java/io/rsocket/frame/decoder/ZeroCopyPayloadDecoder.java

@@ -11,7 +11,6 @@
  * for releasing the payload to free memory when they no long need it.
  */
 public class ZeroCopyPayloadDecoder implements PayloadDecoder {
-
   @Override
   public Payload apply(ByteBuf byteBuf) {
     ByteBuf m;

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

@@ -357,8 +357,7 @@ public void shouldThrownExceptionIfGivenPayloadIsExitsSizeAllowanceWithNoFragmen
                               .hasMessage(INVALID_PAYLOAD_ERROR_MESSAGE))
                   .verify();
               // FIXME: should be removed
-              Assertions.assertThat(rule.connection.getSent())
-                      .allMatch(bb -> bb.release());
+              Assertions.assertThat(rule.connection.getSent()).allMatch(bb -> bb.release());
               rule.assertHasNoLeaks();
             });
   }

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

@@ -175,7 +175,6 @@ public Mono<Payload> requestResponse(Payload payload) {
   @Test
   @Timeout(2_000)
   public void shouldThrownExceptionIfGivenPayloadIsExitsSizeAllowanceWithNoFragmentation() {
-    ByteBufAllocator allocator = rule.alloc();
     final int streamId = 4;
     final AtomicBoolean cancelled = new AtomicBoolean();
     byte[] metadata = new byte[FrameLengthFlyweight.FRAME_LENGTH_MASK];
@@ -226,18 +225,26 @@ protected void hookOnSubscribe(Subscription subscription) {
           .isInstanceOf(IllegalArgumentException.class)
           .hasToString("java.lang.IllegalArgumentException: " + INVALID_PAYLOAD_ERROR_MESSAGE);
       Assertions.assertThat(rule.connection.getSent())
+          .filteredOn(bb -> FrameHeaderFlyweight.frameType(bb) == FrameType.ERROR)
           .hasSize(1)
           .first()
-          .matches(bb -> FrameHeaderFlyweight.frameType(bb) == FrameType.ERROR)
           .matches(bb -> ErrorFrameFlyweight.dataUtf8(bb).contains(INVALID_PAYLOAD_ERROR_MESSAGE))
           .matches(ReferenceCounted::release);
 
       assertThat("Subscription not cancelled.", cancelled.get(), is(true));
+
+      // FIXME: needs to be removed. May have cancel frame as well
+      Assertions.assertThat(rule.connection.getSent())
+          .filteredOn(bb -> bb.refCnt() != 0)
+          .allMatch(ReferenceCounted::release);
       rule.init();
       rule.setAcceptingSocket(acceptingSocket);
     }
-    // FIXME: needs to be removed
-    Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release);
+
+    // FIXME: needs to be removed. May have cancel frame as well
+    Assertions.assertThat(rule.connection.getSent())
+        .filteredOn(bb -> bb.refCnt() != 0)
+        .allMatch(ReferenceCounted::release);
     rule.assertHasNoLeaks();
   }