fixes integration test

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

Author: OlegDokuka

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

@@ -19,7 +19,6 @@
 import io.rsocket.transport.ServerTransport;
 import io.rsocket.util.DefaultPayload;
 import java.time.Duration;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.UnicastProcessor;
@@ -47,7 +46,6 @@ void responderRejectSetup() {
   }
 
   @Test
-  @Disabled("FIXME: needs to be revised")
   void requesterStreamsTerminatedOnZeroErrorFrame() {
     LeaksTrackingByteBufAllocator allocator =
         LeaksTrackingByteBufAllocator.instrument(ByteBufAllocator.DEFAULT);

rsocket-test/src/main/java/io/rsocket/test/LeaksTrackingByteBufAllocator.java

@@ -60,7 +60,8 @@ public LeaksTrackingByteBufAllocator assertHasNoLeaks() {
 
       final Duration awaitZeroRefCntDuration = this.awaitZeroRefCntDuration;
       if (!unreleased.isEmpty() && !awaitZeroRefCntDuration.isZero()) {
-        long endTimeInMillis = System.currentTimeMillis() + awaitZeroRefCntDuration.toMillis();
+        final long startTime = System.currentTimeMillis();
+        final long endTimeInMillis = startTime + awaitZeroRefCntDuration.toMillis();
         boolean hasUnreleased;
         while (System.currentTimeMillis() <= endTimeInMillis) {
           hasUnreleased = false;
@@ -72,14 +73,21 @@ public LeaksTrackingByteBufAllocator assertHasNoLeaks() {
           }
 
           if (!hasUnreleased) {
-            break;
+            System.out.println("all the buffers are released...");
+            return this;
           }
 
-          parkNanos(100);
+          System.out.println("await buffers to be released");
+          for (int i = 0; i < 100; i++) {
+            System.gc();
+            parkNanos(1000);
+            System.gc();
+          }
         }
       }
 
       Assertions.assertThat(unreleased).allMatch(bb -> bb.refCnt() == 0);
+      System.out.println("all the buffers are released...");
     } finally {
       tracker.clear();
     }

rsocket-test/src/main/java/io/rsocket/test/TransportTest.java

@@ -101,6 +101,7 @@ default void setUp() {
   default void close() {
     getTransportPair().responder.awaitAllInteractionTermination(getTimeout());
     getTransportPair().dispose();
+    getTransportPair().awaitClosed();
     getTransportPair().byteBufAllocator.assertHasNoLeaks();
     Hooks.resetOnOperatorDebug();
   }
@@ -611,6 +612,10 @@ public String expectedPayloadMetadata() {
       return metadata;
     }
 
+    public void awaitClosed() {
+      server.onClose().and(client.onClose()).block(Duration.ofMinutes(1));
+    }
+
     private static class AsyncDuplexConnection implements DuplexConnection {
 
       private final DuplexConnection duplexConnection;