Make testClosePendingCreateChannelFutureDuringPoolClosure less time-based.

Author: jchambers

pushy/src/main/java/com/eatthepath/pushy/apns/ApnsChannelPool.java

@@ -235,7 +235,8 @@ private void discardChannel(final Channel channel) {
      * @return a {@code Future} that will be completed when all resources held by this pool have been released
      */
     public Future<Void> close() {
-        final Promise<Void> closeDonePromise = new DefaultPromise<>(this.executor);
+        final Promise<Void> closePromise = new DefaultPromise<>(this.executor);
+
         this.allChannels.close().addListener(allCloseFuture -> {
             ApnsChannelPool.this.isClosed = true;
 
@@ -257,9 +258,10 @@ public Future<Void> close() {
                     acquisitionPromise.tryFailure(POOL_CLOSED_EXCEPTION);
                 }
 
-                closeDonePromise.setSuccess(null);
+                closePromise.setSuccess(null);
             });
         });
-        return closeDonePromise;
+
+        return closePromise;
     }
 }

pushy/src/test/java/com/eatthepath/pushy/apns/ApnsChannelPoolTest.java

@@ -33,6 +33,8 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -240,15 +242,13 @@ void testPendingAcquisitionsDuringPoolClosure() throws Exception {
 
     @SuppressWarnings("AnonymousInnerClassMayBeStatic")
     @Test
-    void testPendingCreateChannelFutureDuringPoolClosure() throws Exception {
-        final Promise<Void> createSuccess = new DefaultPromise<>(EVENT_EXECUTOR);
+    void testClosePendingCreateChannelFutureDuringPoolClosure() throws Exception {
+        final List<Promise<Channel>> createPromises = new ArrayList<>();
+
         final PooledObjectFactory<Channel> factory = new PooledObjectFactory<Channel>() {
             @Override
             public Future<Channel> create(final Promise<Channel> promise) {
-                EVENT_EXECUTOR.schedule(() -> {
-                    promise.trySuccess(new TestChannel(true));
-                    createSuccess.setSuccess(null);
-                }, 1, TimeUnit.SECONDS);
+                createPromises.add(promise);
                 return promise;
             }
 
@@ -261,8 +261,22 @@ public Future<Void> destroy(final Channel channel, final Promise<Void> promise)
 
         final ApnsChannelPool pool = new ApnsChannelPool(factory, 1, EVENT_EXECUTOR, this.metricsListener);
 
-        pool.acquire();
-        pool.close().await();
-        assertTrue(createSuccess.isSuccess());
+        final Future<Channel> acquireNewChannelFuture = pool.acquire();
+        final Future<Channel> acquireReturnedChannelFuture = pool.acquire();
+
+        final Future<Void> closeFuture = pool.close();
+
+        EVENT_EXECUTOR.submit(() -> {
+            final TestChannel channel = new TestChannel(true);
+            createPromises.forEach(channelPromise -> channelPromise.trySuccess(channel));
+        });
+
+        closeFuture.await();
+
+        assertTrue(acquireNewChannelFuture.await().isSuccess(),
+                "Futures waiting for new connections at pool closure should succeed.");
+
+        assertFalse(acquireReturnedChannelFuture.await().isSuccess(),
+                "Futures waiting for existing connections at pool closure should fail.");
     }
 }