improves tests

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

Author: OlegDokuka

rsocket-core/src/test/java/io/rsocket/internal/UnboundedProcessorTest.java

@@ -16,33 +16,28 @@
 
 package io.rsocket.internal;
 
-import static org.assertj.core.api.Assertions.assertThat;
+import java.time.Duration;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
-import io.netty.buffer.UnpooledByteBufAllocator;
 import io.netty.util.CharsetUtil;
 import io.netty.util.ReferenceCountUtil;
 import io.rsocket.buffer.LeaksTrackingByteBufAllocator;
 import io.rsocket.internal.subscriber.AssertSubscriber;
-import java.time.Duration;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Disabled;
-import org.junit.jupiter.api.DisplayName;
-import org.junit.jupiter.api.RepeatedTest;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 import reactor.core.Fuseable;
 import reactor.core.publisher.Hooks;
-import reactor.core.publisher.Operators;
 import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 import reactor.test.util.RaceTestUtils;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 public class UnboundedProcessorTest {
 
   @BeforeAll
@@ -138,7 +133,7 @@ public void ensureUnboundedProcessorDisposesQueueProperly(boolean withFusionEnab
             assertSubscriber.request(1);
           });
 
-      assertSubscriber.values().forEach(ReferenceCountUtil::safeRelease);
+      assertSubscriber.values().forEach(ReferenceCountUtil::release);
 
       allocator.assertHasNoLeaks();
     }
@@ -167,7 +162,6 @@ public void smokeTest1(boolean withFusionEnabled) {
       unboundedProcessor.subscribe(assertSubscriber);
 
       RaceTestUtils.race(
-          Schedulers.boundedElastic(),
           () -> {
             unboundedProcessor.onNext(buffer1);
             unboundedProcessor.onNextPrioritized(buffer2);
@@ -284,10 +278,11 @@ public void smokeTest3(boolean withFusionEnabled) {
     }
   }
 
-  @Test
-  @DisplayName(
-      "Ensures that racing between onNext | dispose | subscribe(cancelled) | terminal will not cause any issues and leaks")
-  public void smokeTest31() {
+  @ParameterizedTest(
+      name =
+          "Ensures that racing between onNext | dispose | subscribe(cancelled) | terminal will not cause any issues and leaks; mode[fusionEnabled={0}]")
+  @ValueSource(booleans = {true, false})
+  public void smokeTest31(boolean withFusionEnabled) {
     final LeaksTrackingByteBufAllocator allocator =
         LeaksTrackingByteBufAllocator.instrument(ByteBufAllocator.DEFAULT);
     final RuntimeException runtimeException = new RuntimeException("test");
@@ -300,7 +295,8 @@ public void smokeTest31() {
       final ByteBuf buffer4 = allocator.buffer(4);
 
       final AssertSubscriber<ByteBuf> assertSubscriber =
-          new AssertSubscriber<ByteBuf>(0).requestedFusionMode(Fuseable.ANY);
+          new AssertSubscriber<ByteBuf>(0)
+              .requestedFusionMode(withFusionEnabled ? Fuseable.ANY : Fuseable.NONE);
 
       RaceTestUtils.race(
           Schedulers.boundedElastic(),
@@ -329,44 +325,43 @@ public void smokeTest31() {
     }
   }
 
-  @RepeatedTest(
+  @ParameterizedTest(
       name =
-          "Ensures that racing between onNext + dispose | downstream async drain should not cause any issues and leaks",
-      value = 10000)
-  @Timeout(60)
-  public void ensuresAsyncFusionAndDisposureHasNoDeadlock() {
+          "Ensures that racing between onNext + dispose | downstream async drain should not cause any issues and leaks; mode[fusionEnabled={0}]")
+  @ValueSource(booleans = {true, false})
+  public void ensuresAsyncFusionAndDisposureHasNoDeadlock(boolean withFusionEnabled) {
     final LeaksTrackingByteBufAllocator allocator =
-        LeaksTrackingByteBufAllocator.instrument(UnpooledByteBufAllocator.DEFAULT);
-    final UnboundedProcessor<ByteBuf> unboundedProcessor = new UnboundedProcessor<>();
-
-    final ByteBuf buffer1 = allocator.buffer(1);
-    final ByteBuf buffer2 = allocator.buffer(2);
-    final ByteBuf buffer3 = allocator.buffer(3);
-    final ByteBuf buffer4 = allocator.buffer(4);
-    final ByteBuf buffer5 = allocator.buffer(5);
-    final ByteBuf buffer6 = allocator.buffer(6);
-
-    final AssertSubscriber<ByteBuf> assertSubscriber =
-        new AssertSubscriber<>(Operators.enableOnDiscard(null, ReferenceCountUtil::safeRelease));
-
-    unboundedProcessor.subscribe(assertSubscriber);
-
-    RaceTestUtils.race(
-        () -> {
-          unboundedProcessor.onNext(buffer1);
-          unboundedProcessor.onNext(buffer2);
-          unboundedProcessor.onNext(buffer3);
-          unboundedProcessor.onNext(buffer4);
-          unboundedProcessor.onNext(buffer5);
-          unboundedProcessor.onNext(buffer6);
-          unboundedProcessor.dispose();
-        },
-        unboundedProcessor::dispose);
-
-    assertSubscriber
-        .await(Duration.ofSeconds(50))
-        .values()
-        .forEach(ReferenceCountUtil::safeRelease);
+        LeaksTrackingByteBufAllocator.instrument(ByteBufAllocator.DEFAULT);
+
+    for (int i = 0; i < 10000; i++) {
+      final UnboundedProcessor<ByteBuf> unboundedProcessor = new UnboundedProcessor<>();
+      final ByteBuf buffer1 = allocator.buffer(1);
+      final ByteBuf buffer2 = allocator.buffer(2);
+      final ByteBuf buffer3 = allocator.buffer(3);
+      final ByteBuf buffer4 = allocator.buffer(4);
+      final ByteBuf buffer5 = allocator.buffer(5);
+      final ByteBuf buffer6 = allocator.buffer(6);
+
+      final AssertSubscriber<ByteBuf> assertSubscriber =
+          new AssertSubscriber<ByteBuf>()
+              .requestedFusionMode(withFusionEnabled ? Fuseable.ANY : Fuseable.NONE);
+
+      unboundedProcessor.subscribe(assertSubscriber);
+
+      RaceTestUtils.race(
+          () -> {
+            unboundedProcessor.onNext(buffer1);
+            unboundedProcessor.onNext(buffer2);
+            unboundedProcessor.onNext(buffer3);
+            unboundedProcessor.onNext(buffer4);
+            unboundedProcessor.onNext(buffer5);
+            unboundedProcessor.onNext(buffer6);
+            unboundedProcessor.dispose();
+          },
+          unboundedProcessor::dispose);
+
+      assertSubscriber.await(Duration.ofSeconds(50)).values().forEach(ReferenceCountUtil::release);
+    }
 
     allocator.assertHasNoLeaks();
   }