fixes test

Signed-off-by: Oleh Dokuka <shadowgun@i.ua>
rsocket · Apr 21, 2020 · 8a3103e · 8a3103e
1 parent 5492447
commit 8a3103e
Show file tree

Hide file tree

Showing 4 changed files with 13 additions and 7 deletions.
diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java b/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

diff --git a/rsocket-core/src/main/java/io/rsocket/frame/decoder/ZeroCopyPayloadDecoder.java b/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;

diff --git a/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java b/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();
             });
   }

diff --git a/rsocket-core/src/test/java/io/rsocket/core/RSocketResponderTest.java b/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();
   }