make sure all destroyHandler are handled asynchronously

reactor · Apr 25, 2019 · f58a528 · f58a528
1 parent 5000e53
commit f58a528
Show file tree

Hide file tree

Showing 5 changed files with 13 additions and 18 deletions.
diff --git a/src/main/java/reactor/pool/AffinityPool.java b/src/main/java/reactor/pool/AffinityPool.java
@@ -125,8 +125,7 @@ void doAcquire(Borrower<POOLABLE> borrower) {
 
             //TODO test this scenario
             if (poolConfig.evictionPredicate.test(element.poolable, element)) {
-                destroyPoolable(element).subscribe(); //this returns a permit
-                allocateOrPend(subPool, borrower);
+                destroyPoolable(element).subscribe(null, t -> allocateOrPend(subPool, borrower), () -> allocateOrPend(subPool, borrower)); //this returns a permit
             }
             else {
                 borrower.stopPendingCountdown();
@@ -275,7 +274,7 @@ public void dispose() {
             toClose.clear();
 
             while(!availableElements.isEmpty()) {
-                destroyPoolable(availableElements.poll()).block();
+                destroyPoolable(availableElements.poll()).subscribe();
             }
         }
     }
@@ -522,7 +521,7 @@ public void onError(Throwable throwable) {
                 return;
             }
 
-            pool.destroyPoolable(slot).subscribe(); //TODO manage further errors?
+            pool.destroyPoolable(slot).subscribe(null, t -> pool.bestEffortAllocateOrPend(), pool::bestEffortAllocateOrPend); //TODO manage further errors?
             actual.onError(throwable);
         }
 
@@ -539,10 +538,7 @@ public void onComplete() {
                 pool.recycle(slot);
             }
             else {
-                pool.destroyPoolable(slot).subscribe(); //TODO manage errors?
-
-                //simplified version of what we do in doAcquire, with the caveat that we don't try to create a SubPool
-                pool.bestEffortAllocateOrPend();
+                pool.destroyPoolable(slot).subscribe(null, t -> pool.bestEffortAllocateOrPend(), pool::bestEffortAllocateOrPend); //TODO manage errors?
             }
 
             actual.onComplete();

diff --git a/src/main/java/reactor/pool/SimpleFifoPool.java b/src/main/java/reactor/pool/SimpleFifoPool.java
@@ -87,7 +87,7 @@ public void dispose() {
             }
 
             while (!elements.isEmpty()) {
-                destroyPoolable(elements.poll()).block();
+                destroyPoolable(elements.poll()).subscribe();
             }
         }
     }

diff --git a/src/main/java/reactor/pool/SimpleLifoPool.java b/src/main/java/reactor/pool/SimpleLifoPool.java
@@ -87,7 +87,7 @@ public void dispose() {
             }
 
             while (!elements.isEmpty()) {
-                destroyPoolable(elements.poll()).block();
+                destroyPoolable(elements.poll()).subscribe();
             }
         }
     }

diff --git a/src/main/java/reactor/pool/SimplePool.java b/src/main/java/reactor/pool/SimplePool.java
@@ -136,14 +136,14 @@ final void maybeRecycleAndDrain(QueuePooledRef<POOLABLE> poolSlot) {
             if (!poolConfig.evictionPredicate.test(poolSlot.poolable, poolSlot)) {
                 metricsRecorder.recordRecycled();
                 elements.offer(poolSlot);
+                drain();
             }
             else {
-                destroyPoolable(poolSlot).subscribe(); //TODO manage errors?
+                destroyPoolable(poolSlot).subscribe(null, e -> drain(), this::drain); //TODO manage errors?
             }
-            drain();
         }
         else {
-            destroyPoolable(poolSlot).subscribe(); //TODO manage errors?
+            destroyPoolable(poolSlot).subscribe(null, e -> drain(), this::drain); //TODO manage errors?
         }
     }
 
@@ -196,7 +196,7 @@ else if (pendingCount > 0) {
 
                 //TODO test the idle eviction scenario
                 if (poolConfig.evictionPredicate.test(slot.poolable, slot)) {
-                    destroyPoolable(slot).subscribe();
+                    destroyPoolable(slot).subscribe(null, e -> drain(), this::drain);
                     continue;
                 }
 
@@ -328,8 +328,7 @@ public void onError(Throwable throwable) {
             //TODO should we separate reset errors?
             pool.metricsRecorder.recordResetLatency(pool.metricsRecorder.measureTime(start));
 
-            pool.destroyPoolable(slot).subscribe(); //TODO manage errors?
-            pool.drain();
+            pool.destroyPoolable(slot).subscribe(null, null, pool::drain); //TODO manage errors?
 
             actual.onError(throwable);
         }

diff --git a/src/test/java/reactor/pool/CommonPoolTest.java b/src/test/java/reactor/pool/CommonPoolTest.java
@@ -518,7 +518,7 @@ void smokeTestAsyncLifo(Function<PoolBuilder<PoolableTest>, AbstractPool<Poolabl
 		pool.acquire().subscribe(acquired3::add);
 		pool.acquire().subscribe(acquired3::add);
 
-		if (!latch1.await(1, TimeUnit.SECONDS)) { //wait for creation of max elements
+		if (!latch1.await(15, TimeUnit.SECONDS)) { //wait for creation of max elements
 			fail("not enough elements created initially, missing " + latch1.getCount());
 		}
 		assertThat(acquired1).hasSize(3);
@@ -537,7 +537,7 @@ void smokeTestAsyncLifo(Function<PoolBuilder<PoolableTest>, AbstractPool<Poolabl
 			slot.release().block();
 		}
 
-		if (latch2.await(2, TimeUnit.SECONDS)) { //wait for the re-creation of max elements
+		if (latch2.await(15, TimeUnit.SECONDS)) { //wait for the re-creation of max elements
 
 			assertThat(acquired2).hasSize(3);