improves leak tracker

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>
Signed-off-by: Oleh Dokuka <oleh.dokuka@icloud.com>
Signed-off-by: OlegDokuka <odokuka@vmware.com>

Author: OlegDokuka

rsocket-core/src/test/java/io/rsocket/buffer/LeaksTrackingByteBufAllocator.java

@@ -5,20 +5,25 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.CompositeByteBuf;
+import io.netty.util.IllegalReferenceCountException;
 import io.netty.util.ResourceLeakDetector;
 import java.lang.reflect.Field;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import org.assertj.core.api.Assertions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Additional Utils which allows to decorate a ByteBufAllocator and track/assertOnLeaks all created
  * ByteBuffs
  */
 public class LeaksTrackingByteBufAllocator implements ByteBufAllocator {
+  static final Logger LOGGER = LoggerFactory.getLogger(LeaksTrackingByteBufAllocator.class);
 
   /**
    * Allows to instrument any given the instance of ByteBufAllocator
@@ -83,6 +88,7 @@ public LeaksTrackingByteBufAllocator assertHasNoLeaks() {
             return this;
           }
 
+          LOGGER.debug(tag + " await buffers to be released");
           for (int i = 0; i < 100; i++) {
             System.gc();
             parkNanos(1000);
@@ -91,22 +97,31 @@ public LeaksTrackingByteBufAllocator assertHasNoLeaks() {
         }
       }
 
-      Assertions.assertThat(unreleased)
-          .allMatch(
-              bb -> {
-                final boolean checkResult = bb.refCnt() == 0;
-
-                if (!checkResult) {
-                  try {
-                    System.out.println(tag + " " + resolveTrackingInfo(bb));
-                  } catch (Exception e) {
-                    e.printStackTrace();
-                  }
-                }
-
-                return checkResult;
-              },
-              tag);
+      Set<ByteBuf> collected = new HashSet<>();
+      for (ByteBuf buf : unreleased) {
+        if (buf.refCnt() != 0) {
+          try {
+            collected.add(buf);
+          } catch (IllegalReferenceCountException ignored) {
+            // fine to ignore if throws because of refCnt
+          }
+        }
+      }
+
+      Assertions.assertThat(
+              collected
+                  .stream()
+                  .filter(bb -> bb.refCnt() != 0)
+                  .peek(
+                      bb -> {
+                        try {
+                          LOGGER.debug(tag + " " + resolveTrackingInfo(bb));
+                        } catch (Exception e) {
+                          e.printStackTrace();
+                        }
+                      }))
+          .describedAs("[" + tag + "] all buffers expected to be released but got ")
+          .isEmpty();
     } finally {
       tracker.clear();
     }

rsocket-test/build.gradle

@@ -28,6 +28,8 @@ dependencies {
     implementation 'io.projectreactor:reactor-test'
     implementation 'org.assertj:assertj-core'
     implementation 'org.mockito:mockito-core'
+    implementation 'org.awaitility:awaitility'
+    implementation 'org.slf4j:slf4j-api'
 }
 
 jar {

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

@@ -5,20 +5,25 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.CompositeByteBuf;
+import io.netty.util.IllegalReferenceCountException;
 import io.netty.util.ResourceLeakDetector;
 import java.lang.reflect.Field;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import org.assertj.core.api.Assertions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Additional Utils which allows to decorate a ByteBufAllocator and track/assertOnLeaks all created
  * ByteBuffs
  */
 public class LeaksTrackingByteBufAllocator implements ByteBufAllocator {
+  static final Logger LOGGER = LoggerFactory.getLogger(LeaksTrackingByteBufAllocator.class);
 
   /**
    * Allows to instrument any given the instance of ByteBufAllocator
@@ -83,7 +88,7 @@ public LeaksTrackingByteBufAllocator assertHasNoLeaks() {
             return this;
           }
 
-          System.out.println(tag + " await buffers to be released");
+          LOGGER.debug(tag + " await buffers to be released");
           for (int i = 0; i < 100; i++) {
             System.gc();
             parkNanos(1000);
@@ -92,22 +97,31 @@ public LeaksTrackingByteBufAllocator assertHasNoLeaks() {
         }
       }
 
-      Assertions.assertThat(unreleased)
-          .allMatch(
-              bb -> {
-                final boolean checkResult = bb.refCnt() == 0;
-
-                if (!checkResult) {
-                  try {
-                    System.out.println(tag + " " + resolveTrackingInfo(bb));
-                  } catch (Exception e) {
-                    e.printStackTrace();
-                  }
-                }
-
-                return checkResult;
-              },
-              tag);
+      Set<ByteBuf> collected = new HashSet<>();
+      for (ByteBuf buf : unreleased) {
+        if (buf.refCnt() != 0) {
+          try {
+            collected.add(buf);
+          } catch (IllegalReferenceCountException ignored) {
+            // fine to ignore if throws because of refCnt
+          }
+        }
+      }
+
+      Assertions.assertThat(
+              collected
+                  .stream()
+                  .filter(bb -> bb.refCnt() != 0)
+                  .peek(
+                      bb -> {
+                        try {
+                          LOGGER.debug(tag + " " + resolveTrackingInfo(bb));
+                        } catch (Exception e) {
+                          e.printStackTrace();
+                        }
+                      }))
+          .describedAs("[" + tag + "] all buffers expected to be released but got ")
+          .isEmpty();
     } finally {
       tracker.clear();
     }