Repository: relayrides/pushy (java)

Issue:
Messages hangs after connection failure
When ApnsChannelFactory.create fails, for instance during a network outage, the caller only gets feedback on the listener of the first message and some random (assume it's timing base) number of subsequent messages. Most of the messages seem to be queued up an only processed at some later point when ApnsChannelFactory.create succeeds. Closing the ApnsClient seem to trigger failure callbacks on all of the outstanding messages.

I would expect to get a failure callback on the listeners of all messages sent during a network outage in a timely manner. The fact that they get queued up and don't seem to be processed until connections are reestablished creates three potential issues. 

- The queue size can get very large triggering OOM problems
- If reestablishing connection takes a long time messages can be out of date when they are delivered and make no sense to clients
- The lack of timely feedback on the failure by calling the listener makes it hard to catch and log failures in an appropriate manner

Is the behavior described in the first paragraph intended behavior or is this a bug?

Below is a partial test showing how to recreate the issue.

```java
@Test
public void sendMessagesSimple() throws InterruptedException, ExecutionException {

    SimpleApnsPushNotification pushNotification = createNotification(pushAddress, payload);

    for (int i = 0; i < 4; i++) {
        PushNotificationFuture<SimpleApnsPushNotification, PushNotificationResponse<SimpleApnsPushNotification>>
                pushNotificationFuture = apnsClient.sendNotification(pushNotification);

        pushNotificationFuture.addListener(
                (PushNotificationResponseListener<SimpleApnsPushNotification>) future -> {
                    System.out.println("Callback " + future);
                });
        Thread.sleep(1);
    }

    Thread.sleep(30000);
    System.out.println("\nWaited 30 sec. Shutting down\n");
    apnsClient.close();
}
```

This creates the following output when setting the APNS server name to an invalid server.

```
653  [main] INFO  com.turo.pushy.apns.SslUtil - Native SSL provider is available and supports ALPN; will use native provider. 
Callback handleConnectionCreationFailed
Callback handleWriteFailure
Callback PushNotificationPromise@7aa13b53(failure: io.netty.channel.ConnectTimeoutException: connection timed out: test.encap.no/85.221.67.15:9080)

Waited 30 sec. Shutting down

31051 [main] INFO  com.turo.pushy.apns.ApnsClient - Shutting down. 
Callback handleWriteFailure
Callback PushNotificationPromise@12a6f922(failure: java.lang.IllegalStateException: Channel pool has closed and no more channels may be acquired.)
Callback handleWriteFailure
Callback PushNotificationPromise@61ae3386(failure: java.lang.IllegalStateException: Channel pool has closed and no more channels may be acquired.)
Callback handleWriteFailure
Callback PushNotificationPromise@dbea44f(failure: java.lang.IllegalStateException: Channel pool has closed and no more channels may be acquired.)
```

Pull request description:
Process pending acquisition attempts when a connection attempt fails
This fixes #583, a bug where the `ApnsChannelPool` would get stuck and stop processing acquisition attempts if a connection attempt failed. After this change, the pool will move on to the next pending acquisition attempt when a connection attempt fails.

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 8ed698d3..a7e760fc 100644
--- a/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java
+++ b/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java
@@ -150,6 +150,7 @@ class ApnsChannelPool {
                 if (channelFromIdlePool.isActive()) {
                     acquirePromise.trySuccess(channelFromIdlePool);
                 } else {
+                    // The channel from the idle pool isn't usable; discard it and create a new one instead
                     this.discardChannel(channelFromIdlePool);
                     this.acquireWithinEventExecutor(acquirePromise);
                 }
@@ -176,6 +177,11 @@ class ApnsChannelPool {
                                 ApnsChannelPool.this.metricsListener.handleConnectionCreationFailed();
 
                                 acquirePromise.tryFailure(future.cause());
+
+                                // If we failed to open a connection, this is the end of the line for this acquisition
+                                // attempt, and callers won't be able to release the channel (since they didn't get one
+                                // in the first place). Move on to the next acquisition attempt if one is present.
+                                ApnsChannelPool.this.handleNextAcquisition();
                             }
                         }
                     });
@@ -210,6 +216,11 @@ class ApnsChannelPool {
         assert this.executor.inEventLoop();
 
         this.idleChannels.add(channel);
+        this.handleNextAcquisition();
+    }
+
+    private void handleNextAcquisition() {
+        assert this.executor.inEventLoop();
 
         if (!this.pendingAcquisitionPromises.isEmpty()) {
             this.acquireWithinEventExecutor(this.pendingAcquisitionPromises.poll());


Test patch:
diff --git a/pushy/src/test/java/com/turo/pushy/apns/ApnsClientTest.java b/pushy/src/test/java/com/turo/pushy/apns/ApnsClientTest.java
index 885b7a72..f7a255b8 100644
--- a/pushy/src/test/java/com/turo/pushy/apns/ApnsClientTest.java
+++ b/pushy/src/test/java/com/turo/pushy/apns/ApnsClientTest.java
@@ -24,6 +24,8 @@ package com.turo.pushy.apns;
 
 import com.turo.pushy.apns.server.*;
 import com.turo.pushy.apns.util.SimpleApnsPushNotification;
+import com.turo.pushy.apns.util.concurrent.PushNotificationFuture;
+import com.turo.pushy.apns.util.concurrent.PushNotificationResponseListener;
 import io.netty.buffer.ByteBuf;
 import io.netty.handler.codec.http2.Http2Headers;
 import io.netty.util.concurrent.Future;
@@ -40,6 +42,7 @@ import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.Assert.*;
@@ -636,4 +639,52 @@ public class ApnsClientTest extends AbstractClientServerTest {
 
         assertEquals(1, metricsListener.getWriteFailures().size());
     }
+
+    @Test
+    @Parameters({"true", "false"})
+    public void testRepeatedlySendNotificationAfterConnectionFailure(final boolean useTokenAuthentication) throws Exception {
+        final ApnsClient client = useTokenAuthentication ?
+                this.buildTokenAuthenticationClient() : this.buildTlsAuthenticationClient();
+
+        final SimpleApnsPushNotification pushNotification =
+                new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
+
+        for (int i = 0; i < 3; i++) {
+            // We should see delays of roughly 0, 1, and 2 seconds; 4 seconds per notification is excessive, but better
+            // to play it safe with a timed assertion.
+            final Future<PushNotificationResponse<SimpleApnsPushNotification>> sendFuture =
+                    client.sendNotification(pushNotification);
+
+            assertTrue(sendFuture.await(4, TimeUnit.SECONDS));
+            assertFalse(sendFuture.isSuccess());
+        }
+    }
+
+    @Test
+    @Parameters({"true", "false"})
+    public void testRepeatedlySendNotificationAfterConnectionFailureWithListeners(final boolean useTokenAuthentication) throws Exception {
+        final ApnsClient client = useTokenAuthentication ?
+                this.buildTokenAuthenticationClient() : this.buildTlsAuthenticationClient();
+
+        final SimpleApnsPushNotification pushNotification =
+                new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
+
+        final int notificationCount = 3;
+
+        final CountDownLatch countDownLatch = new CountDownLatch(notificationCount);
+
+        for (int i = 0; i < notificationCount; i++) {
+            client.sendNotification(pushNotification).addListener(new PushNotificationResponseListener<SimpleApnsPushNotification>() {
+
+                @Override
+                public void operationComplete(final PushNotificationFuture<SimpleApnsPushNotification, PushNotificationResponse<SimpleApnsPushNotification>> simpleApnsPushNotificationPushNotificationResponsePushNotificationFuture) throws Exception {
+                    countDownLatch.countDown();
+                }
+            });
+        }
+
+        // We should see delays of roughly 0, 1, and 2 seconds (for a total of 3 seconds); waiting 6 seconds in total
+        // is overkill, but it's best to leave significant margin on timed assertions.
+        assertTrue(countDownLatch.await(6, TimeUnit.SECONDS));
+    }
 }