Repository: relayrides/pushy (java)

Issue:
Notifications sent while connection pool is closing may not be resolved
desc:when I close the client, I only received one response but have send ten messages.

In class ApnsChannelPool, I think these codes below has something wrong (**_please see 'added by me '  and 'modified by me'_**):

```java
class ApnsChannelPool {

    private final PooledObjectFactory<Channel> channelFactory;
    private final OrderedEventExecutor executor;
    private final int capacity;

    private final ApnsChannelPoolMetricsListener metricsListener;

    private final ChannelGroup allChannels;
    private final Queue<Channel> idleChannels = new ArrayDeque<>();

    private final Set<Future<Channel>> pendingCreateChannelFutures = new HashSet<>();
    private final Queue<Promise<Channel>> pendingAcquisitionPromises = new ArrayDeque<>();

    private static final Logger log = LoggerFactory.getLogger(ApnsChannelPool.class);
    
    // added by me
    private volatile boolean closed = false;
   // added by me

    private void acquireWithinEventExecutor(final Promise<Channel> acquirePromise) {
        assert this.executor.inEventLoop();

        final Channel channelFromIdlePool = ApnsChannelPool.this.idleChannels.poll();

        if (channelFromIdlePool != null) {
            if (channelFromIdlePool.isActive()) {
                acquirePromise.trySuccess(channelFromIdlePool);
            } else {
                this.discardChannel(channelFromIdlePool);
                this.acquireWithinEventExecutor(acquirePromise);
            }
        } else {
            // We don't have any connections ready to go; create a new one if possible.
            if (this.allChannels.size() + this.pendingCreateChannelFutures.size() < this.capacity) {
                final Future<Channel> createChannelFuture = this.channelFactory.create(executor.<Channel>newPromise());
                this.pendingCreateChannelFutures.add(createChannelFuture);

                createChannelFuture.addListener(new GenericFutureListener<Future<Channel>>() {

                    @Override
                    public void operationComplete(final Future<Channel> future) throws Exception {
                        ApnsChannelPool.this.pendingCreateChannelFutures.remove(createChannelFuture);

                        if (future.isSuccess()) {
                            final Channel channel = future.getNow();

                            ApnsChannelPool.this.allChannels.add(channel);
                            ApnsChannelPool.this.metricsListener.handleConnectionAdded();

                            acquirePromise.trySuccess(channel);
                        } else {
                            ApnsChannelPool.this.metricsListener.handleConnectionCreationFailed();

                            acquirePromise.tryFailure(future.cause());
                            
                        }
                    }
                });
            } else {
            	// modified by me
            	if(closed){
            		acquirePromise.tryFailure(new IllegalStateException("pool has been closed"));
            	}else{
            		pendingAcquisitionPromises.add(acquirePromise);
            	}
               // added by me
            }
        }
    }

    public Future<Void> close() {
    	// added by me
    	closed = true;
    	Promise<Channel> pro;
    	while(null != (pro = pendingAcquisitionPromises.poll())){
    		pro.tryFailure(new IllegalStateException("pool has been closed"));
    	}
     // added by me
        return this.allChannels.close();
    }
```

Pull request description:
Gracefully handle channel acquisition attempts after channel pools have closed
This fixes #524 by (as @Tielancy suggested) explicitly keeping track of a connection pool's closure state and failing acquisition attempts when and after a pool closes. The changes are easiest to read with [whitespace-only changes hidden](https://github.com/relayrides/pushy/pull/527/files?w=1).

Patch:
diff --git a/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java b/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java
index 491b4b8c..8ed698d3 100644
--- a/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java
+++ b/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java
@@ -63,6 +63,11 @@ class ApnsChannelPool {
     private final Set<Future<Channel>> pendingCreateChannelFutures = new HashSet<>();
     private final Queue<Promise<Channel>> pendingAcquisitionPromises = new ArrayDeque<>();
 
+    private boolean isClosed = false;
+
+    private static final Exception POOL_CLOSED_EXCEPTION =
+            new IllegalStateException("Channel pool has closed and no more channels may be acquired.");
+
     private static final Logger log = LoggerFactory.getLogger(ApnsChannelPool.class);
 
     private static class NoopChannelPoolMetricsListener implements ApnsChannelPoolMetricsListener {
@@ -89,7 +94,7 @@ class ApnsChannelPool {
      * @param executor the executor on which listeners for acquisition/release promises will be called
      * @param metricsListener an optional listener for metrics describing the performance and behavior of the pool
      */
-    public ApnsChannelPool(final PooledObjectFactory<Channel> channelFactory, final int capacity, final OrderedEventExecutor executor, final ApnsChannelPoolMetricsListener metricsListener) {
+    ApnsChannelPool(final PooledObjectFactory<Channel> channelFactory, final int capacity, final OrderedEventExecutor executor, final ApnsChannelPoolMetricsListener metricsListener) {
         this.channelFactory = channelFactory;
         this.capacity = capacity;
         this.executor = executor;
@@ -111,7 +116,7 @@ class ApnsChannelPool {
      *
      * @see ApnsChannelPool#release(Channel)
      */
-    public Future<Channel> acquire() {
+    Future<Channel> acquire() {
         final Promise<Channel> acquirePromise = new DefaultPromise<>(this.executor);
 
         if (this.executor.inEventLoop()) {
@@ -138,44 +143,48 @@ class ApnsChannelPool {
     private void acquireWithinEventExecutor(final Promise<Channel> acquirePromise) {
         assert this.executor.inEventLoop();
 
-        final Channel channelFromIdlePool = ApnsChannelPool.this.idleChannels.poll();
+        if (!this.isClosed) {
+            final Channel channelFromIdlePool = ApnsChannelPool.this.idleChannels.poll();
 
-        if (channelFromIdlePool != null) {
-            if (channelFromIdlePool.isActive()) {
-                acquirePromise.trySuccess(channelFromIdlePool);
+            if (channelFromIdlePool != null) {
+                if (channelFromIdlePool.isActive()) {
+                    acquirePromise.trySuccess(channelFromIdlePool);
+                } else {
+                    this.discardChannel(channelFromIdlePool);
+                    this.acquireWithinEventExecutor(acquirePromise);
+                }
             } else {
-                this.discardChannel(channelFromIdlePool);
-                this.acquireWithinEventExecutor(acquirePromise);
-            }
-        } else {
-            // We don't have any connections ready to go; create a new one if possible.
-            if (this.allChannels.size() + this.pendingCreateChannelFutures.size() < this.capacity) {
-                final Future<Channel> createChannelFuture = this.channelFactory.create(executor.<Channel>newPromise());
-                this.pendingCreateChannelFutures.add(createChannelFuture);
+                // We don't have any connections ready to go; create a new one if possible.
+                if (this.allChannels.size() + this.pendingCreateChannelFutures.size() < this.capacity) {
+                    final Future<Channel> createChannelFuture = this.channelFactory.create(executor.<Channel>newPromise());
+                    this.pendingCreateChannelFutures.add(createChannelFuture);
 
-                createChannelFuture.addListener(new GenericFutureListener<Future<Channel>>() {
+                    createChannelFuture.addListener(new GenericFutureListener<Future<Channel>>() {
 
-                    @Override
-                    public void operationComplete(final Future<Channel> future) throws Exception {
-                        ApnsChannelPool.this.pendingCreateChannelFutures.remove(createChannelFuture);
+                        @Override
+                        public void operationComplete(final Future<Channel> future) throws Exception {
+                            ApnsChannelPool.this.pendingCreateChannelFutures.remove(createChannelFuture);
 
-                        if (future.isSuccess()) {
-                            final Channel channel = future.getNow();
+                            if (future.isSuccess()) {
+                                final Channel channel = future.getNow();
 
-                            ApnsChannelPool.this.allChannels.add(channel);
-                            ApnsChannelPool.this.metricsListener.handleConnectionAdded();
+                                ApnsChannelPool.this.allChannels.add(channel);
+                                ApnsChannelPool.this.metricsListener.handleConnectionAdded();
 
-                            acquirePromise.trySuccess(channel);
-                        } else {
-                            ApnsChannelPool.this.metricsListener.handleConnectionCreationFailed();
+                                acquirePromise.trySuccess(channel);
+                            } else {
+                                ApnsChannelPool.this.metricsListener.handleConnectionCreationFailed();
 
-                            acquirePromise.tryFailure(future.cause());
+                                acquirePromise.tryFailure(future.cause());
+                            }
                         }
-                    }
-                });
-            } else {
-                pendingAcquisitionPromises.add(acquirePromise);
+                    });
+                } else {
+                    pendingAcquisitionPromises.add(acquirePromise);
+                }
             }
+        } else {
+            acquirePromise.tryFailure(POOL_CLOSED_EXCEPTION);
         }
     }
 
@@ -184,7 +193,7 @@ class ApnsChannelPool {
      *
      * @param channel the channel to return to the pool
      */
-    public void release(final Channel channel) {
+    void release(final Channel channel) {
         if (this.executor.inEventLoop()) {
             this.releaseWithinEventExecutor(channel);
         } else {
@@ -234,9 +243,15 @@ class ApnsChannelPool {
         return this.allChannels.close().addListener(new GenericFutureListener<Future<Void>>() {
             @Override
             public void operationComplete(final Future<Void> future) throws Exception {
+                ApnsChannelPool.this.isClosed = true;
+
                 if (ApnsChannelPool.this.channelFactory instanceof Closeable) {
                     ((Closeable) ApnsChannelPool.this.channelFactory).close();
                 }
+
+                for (final Promise<Channel> acquisitionPromise : ApnsChannelPool.this.pendingAcquisitionPromises) {
+                    acquisitionPromise.tryFailure(POOL_CLOSED_EXCEPTION);
+                }
             }
         });
     }


Test patch:
diff --git a/pushy/src/test/java/com/turo/pushy/apns/ApnsChannelPoolTest.java b/pushy/src/test/java/com/turo/pushy/apns/ApnsChannelPoolTest.java
index 53bc208e..395c380c 100644
--- a/pushy/src/test/java/com/turo/pushy/apns/ApnsChannelPoolTest.java
+++ b/pushy/src/test/java/com/turo/pushy/apns/ApnsChannelPoolTest.java
@@ -192,4 +192,26 @@ public class ApnsChannelPoolTest {
         assertEquals(1, this.metricsListener.getConnectionsRemoved());
         assertEquals(0, this.metricsListener.getConnectionsFailed());
     }
+
+    @Test
+    public void testAcquireFromClosedPool() throws Exception {
+        this.pool.close().await();
+
+        final Future<Channel> acquireFuture = this.pool.acquire().await();
+        assertFalse(acquireFuture.isSuccess());
+    }
+
+    @Test
+    public void testPendingAcquisitionsDuringPoolClosure() throws Exception {
+        final Future<Channel> firstFuture = this.pool.acquire().await();
+
+        assertTrue(firstFuture.isSuccess());
+
+        final Future<Channel> pendingFuture = this.pool.acquire();
+
+        this.pool.close().await();
+
+        assertTrue(pendingFuture.await(1000));
+        assertFalse(pendingFuture.isSuccess());
+    }
 }