Repository: relayrides/pushy (java)

Issue:
Flush less frequently
As things stand now, we `flush` after every attempt to send a notification. For users writing small numbers of notifications, this is probably fine since it minimizes latency. For users writing zillions of notifications, it's probably a poor use of system resources, since `flush` is an expensive operation.

I propose:

1. Add a handler to the pipeline that flushes after N bytes have been written or M milliseconds of inactivity have elapsed.
2. Stop flushing after every write.

This will increase latency for single notifications, but may improve throughput for bulk sending operations. We'll definitely want to measure changes in throughput before deciding whether to ship something like this.

Pull request:
Defer flushes (fixes #276)
This change defers calls to `flush` until either:

1. a given number of notifications are waiting to be flushed or
2. the channel has been idle for a given amount of time

This should significantly cut down on the number of calls to `flush`, which should yield significant performance benefits for users sending lots of notifications. These changes are easiest to read with [whitespace-only changes hidden](https://github.com/relayrides/pushy/pull/278/files?w=1).

----

TODO:

- [x] Update docs (9f0ea27)

Patch:
diff --git a/src/main/java/com/relayrides/pushy/apns/ApnsClient.java b/src/main/java/com/relayrides/pushy/apns/ApnsClient.java
index d7738f3f..7edcbe43 100644
--- a/src/main/java/com/relayrides/pushy/apns/ApnsClient.java
+++ b/src/main/java/com/relayrides/pushy/apns/ApnsClient.java
@@ -138,9 +138,12 @@ public class ApnsClient<T extends ApnsPushNotification> {
     private long writeTimeoutMillis = DEFAULT_WRITE_TIMEOUT_MILLIS;
     private Long gracefulShutdownTimeoutMillis;
 
+    private int maxUnflushedNotifications = DEFAULT_MAX_UNFLUSHED_NOTIFICATIONS;
+    private long flushAfterIdleTimeMillis = DEFAULT_WRITE_TIMEOUT_MILLIS;
+
     private volatile ChannelPromise connectionReadyPromise;
     private volatile ChannelPromise reconnectionPromise;
-    private long reconnectDelay = INITIAL_RECONNECT_DELAY;
+    private long reconnectDelaySeconds = INITIAL_RECONNECT_DELAY_SECONDS;
 
     private final Map<T, Promise<PushNotificationResponse<T>>> responsePromises = new IdentityHashMap<>();
 
@@ -152,6 +155,10 @@ public class ApnsClient<T extends ApnsPushNotification> {
      */
     public static final long DEFAULT_WRITE_TIMEOUT_MILLIS = 20_000;
 
+    public static final long DEFAULT_FLUSH_AFTER_IDLE_MILLIS = 50;
+
+    public static final int DEFAULT_MAX_UNFLUSHED_NOTIFICATIONS = 128;
+
     /**
      * The hostname for the production APNs gateway.
      */
@@ -180,9 +187,9 @@ public class ApnsClient<T extends ApnsPushNotification> {
 
     private static final ClientNotConnectedException NOT_CONNECTED_EXCEPTION = new ClientNotConnectedException();
 
-    private static final long INITIAL_RECONNECT_DELAY = 1; // second
-    private static final long MAX_RECONNECT_DELAY = 60; // seconds
-    static final int PING_IDLE_TIME = 60; // seconds
+    private static final long INITIAL_RECONNECT_DELAY_SECONDS = 1; // second
+    private static final long MAX_RECONNECT_DELAY_SECONDS = 60; // seconds
+    static final int PING_IDLE_TIME_MILLIS = 60_000; // milliseconds
 
     private static final Logger log = LoggerFactory.getLogger(ApnsClient.class);
 
@@ -423,6 +430,7 @@ public class ApnsClient<T extends ApnsPushNotification> {
                             final ApnsClientHandler<T> apnsClientHandler = new ApnsClientHandler.ApnsClientHandlerBuilder<T>()
                                     .server(false)
                                     .apnsClient(ApnsClient.this)
+                                    .maxUnflushedNotifications(ApnsClient.this.maxUnflushedNotifications)
                                     .encoderEnforceMaxConcurrentStreams(true)
                                     .build();
 
@@ -432,7 +440,7 @@ public class ApnsClient<T extends ApnsPushNotification> {
                                 }
                             }
 
-                            context.pipeline().addLast(new IdleStateHandler(0, 0, PING_IDLE_TIME));
+                            context.pipeline().addLast(new IdleStateHandler(0, ApnsClient.this.flushAfterIdleTimeMillis, PING_IDLE_TIME_MILLIS, TimeUnit.MILLISECONDS));
                             context.pipeline().addLast(apnsClientHandler);
 
                             // Add this to the end of the queue so any events enqueued by the client handler happen
@@ -556,6 +564,37 @@ public class ApnsClient<T extends ApnsPushNotification> {
         this.writeTimeoutMillis = writeTimeoutMillis;
     }
 
+    /**
+     * <p>Sets the thresholds at which this client will flush notifications enqueued for sending to the APNs server. By
+     * default, notifications will be flushed after {@value ApnsClient#DEFAULT_MAX_UNFLUSHED_NOTIFICATIONS} unflushed
+     * notifications have been enqueued or {@value ApnsClient#DEFAULT_FLUSH_AFTER_IDLE_MILLIS} milliseconds of
+     * inactivity have elapsed.</p>
+     *
+     * <p>Callers may set both thresholds to zero to flush all notifications immediately, which will reduce latency at
+     * the expence of decreasing efficiency and throughput. Changes to the flushing thresholds will take effect on the
+     * next connection attempt.</p>
+     *
+     * @param maxUnflushedNotifications The maximum number of notifications that may be enqueued before the sending
+     * queue is flushed. Must be positive if {@code maxIdleTimeMillis} is also positive or zero if
+     * {@code maxIdleTimeMillis} is also zero. If zero, notifications are always sent immediately.
+     * @param maxIdleTimeMillis The maximum amount of time, in milliseconds, since the last attempt to send a
+     * notification that may elapse before the sending queue is flushed. Must be positive if
+     * {@code maxUnflushedNotifications} is also positive or zero if {@code maxUnflushedNotifications} is also zero. If
+     * zero, notifications are always sent immediately.
+     *
+     * @since 0.7
+     */
+    public void setFlushThresholds(final int maxUnflushedNotifications, final long maxIdleTimeMillis) {
+        if ((maxUnflushedNotifications > 0 && maxIdleTimeMillis > 0) || (maxUnflushedNotifications == 0 && maxIdleTimeMillis == 0)) {
+            synchronized (this.bootstrap) {
+                this.maxUnflushedNotifications = maxUnflushedNotifications;
+                this.flushAfterIdleTimeMillis = maxIdleTimeMillis;
+            }
+        } else {
+            throw new IllegalArgumentException("Notification count and idle time must both be positive or both be zero.");
+        }
+    }
+
     /**
      * Sets the proxy handler factory to be used to construct proxy handlers when establishing a new connection to the
      * APNs gateway. Proxy handlers are added to the beginning of the client's pipeline. A client's proxy handler
@@ -645,7 +684,7 @@ public class ApnsClient<T extends ApnsPushNotification> {
                                 }
 
                                 if (ApnsClient.this.reconnectionPromise != null) {
-                                    log.debug("Disconnected. Next automatic reconnection attempt in {} seconds.", ApnsClient.this.reconnectDelay);
+                                    log.debug("Disconnected. Next automatic reconnection attempt in {} seconds.", ApnsClient.this.reconnectDelaySeconds);
 
                                     future.channel().eventLoop().schedule(new Runnable() {
 
@@ -654,9 +693,9 @@ public class ApnsClient<T extends ApnsPushNotification> {
                                             log.debug("Attempting to reconnect.");
                                             ApnsClient.this.connect(host, port);
                                         }
-                                    }, ApnsClient.this.reconnectDelay, TimeUnit.SECONDS);
+                                    }, ApnsClient.this.reconnectDelaySeconds, TimeUnit.SECONDS);
 
-                                    ApnsClient.this.reconnectDelay = Math.min(ApnsClient.this.reconnectDelay, MAX_RECONNECT_DELAY);
+                                    ApnsClient.this.reconnectDelaySeconds = Math.min(ApnsClient.this.reconnectDelaySeconds, MAX_RECONNECT_DELAY_SECONDS);
                                 }
                             }
 
@@ -691,7 +730,7 @@ public class ApnsClient<T extends ApnsPushNotification> {
                                         log.info("Connected to {}.", future.channel().remoteAddress());
                                     }
 
-                                    ApnsClient.this.reconnectDelay = INITIAL_RECONNECT_DELAY;
+                                    ApnsClient.this.reconnectDelaySeconds = INITIAL_RECONNECT_DELAY_SECONDS;
                                     ApnsClient.this.reconnectionPromise = future.channel().newPromise();
                                 }
 
@@ -778,11 +817,16 @@ public class ApnsClient<T extends ApnsPushNotification> {
      * automatically. Callers may wait for a reconnection attempt to complete by waiting for the {@code Future} returned
      * by the {@link ApnsClient#getReconnectionFuture()} method.</p>
      *
+     * <p>Note that, depending on this client's flushing thresholds, notifications may be enqueued so they may be sent
+     * in bulk to increase efficiency and throughput at the expense of latency.</p>
+     *
      * @param notification the notification to send to the APNs gateway
      *
      * @return a {@code Future} that will complete when the notification has been either accepted or rejected by the
      * APNs gateway
      *
+     * @see ApnsClient#setFlushThresholds(int, long)
+     *
      * @since 0.5
      */
     public Future<PushNotificationResponse<T>> sendNotification(final T notification) {
@@ -815,7 +859,7 @@ public class ApnsClient<T extends ApnsPushNotification> {
                 }
             });
 
-            connectionReadyPromise.channel().writeAndFlush(notification).addListener(new GenericFutureListener<ChannelFuture>() {
+            connectionReadyPromise.channel().write(notification).addListener(new GenericFutureListener<ChannelFuture>() {
 
                 @Override
                 public void operationComplete(final ChannelFuture future) throws Exception {
diff --git a/src/main/java/com/relayrides/pushy/apns/ApnsClientHandler.java b/src/main/java/com/relayrides/pushy/apns/ApnsClientHandler.java
index 1d59451e..162f35a6 100644
--- a/src/main/java/com/relayrides/pushy/apns/ApnsClientHandler.java
+++ b/src/main/java/com/relayrides/pushy/apns/ApnsClientHandler.java
@@ -48,6 +48,7 @@ import io.netty.handler.codec.http2.Http2Exception;
 import io.netty.handler.codec.http2.Http2FrameAdapter;
 import io.netty.handler.codec.http2.Http2Headers;
 import io.netty.handler.codec.http2.Http2Settings;
+import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateEvent;
 import io.netty.handler.timeout.WriteTimeoutException;
 import io.netty.util.AsciiString;
@@ -67,6 +68,9 @@ class ApnsClientHandler<T extends ApnsPushNotification> extends Http2ConnectionH
     private long nextPingId = new Random().nextLong();
     private ScheduledFuture<?> pingTimeoutFuture;
 
+    private final int maxUnflushedNotifications;
+    private int unflushedNotifications = 0;
+
     private static final int PING_TIMEOUT = 30; // seconds
 
     private static final String APNS_PATH_PREFIX = "/3/device/";
@@ -89,6 +93,7 @@ class ApnsClientHandler<T extends ApnsPushNotification> extends Http2ConnectionH
     public static class ApnsClientHandlerBuilder<S extends ApnsPushNotification> extends AbstractHttp2ConnectionHandlerBuilder<ApnsClientHandler<S>, ApnsClientHandlerBuilder<S>> {
 
         private ApnsClient<S> apnsClient;
+        private int maxUnflushedNotifications = 0;
 
         public ApnsClientHandlerBuilder<S> apnsClient(final ApnsClient<S> apnsClient) {
             this.apnsClient = apnsClient;
@@ -99,6 +104,15 @@ class ApnsClientHandler<T extends ApnsPushNotification> extends Http2ConnectionH
             return this.apnsClient;
         }
 
+        public ApnsClientHandlerBuilder<S> maxUnflushedNotifications(final int maxUnflushedNotifications) {
+            this.maxUnflushedNotifications = maxUnflushedNotifications;
+            return this;
+        }
+
+        public int maxUnflushedNotifications() {
+            return this.maxUnflushedNotifications;
+        }
+
         @Override
         public ApnsClientHandlerBuilder<S> server(final boolean isServer) {
             return super.server(isServer);
@@ -111,7 +125,7 @@ class ApnsClientHandler<T extends ApnsPushNotification> extends Http2ConnectionH
 
         @Override
         public ApnsClientHandler<S> build(final Http2ConnectionDecoder decoder, final Http2ConnectionEncoder encoder, final Http2Settings initialSettings) {
-            final ApnsClientHandler<S> handler = new ApnsClientHandler<>(decoder, encoder, initialSettings, this.apnsClient());
+            final ApnsClientHandler<S> handler = new ApnsClientHandler<>(decoder, encoder, initialSettings, this.apnsClient(), this.maxUnflushedNotifications());
             this.frameListener(handler.new ApnsClientHandlerFrameAdapter());
             return handler;
         }
@@ -186,15 +200,16 @@ class ApnsClientHandler<T extends ApnsPushNotification> extends Http2ConnectionH
         }
     }
 
-    protected ApnsClientHandler(final Http2ConnectionDecoder decoder, final Http2ConnectionEncoder encoder, final Http2Settings initialSettings, final ApnsClient<T> apnsClient) {
+    protected ApnsClientHandler(final Http2ConnectionDecoder decoder, final Http2ConnectionEncoder encoder, final Http2Settings initialSettings, final ApnsClient<T> apnsClient, final int maxUnflushedNotifications) {
         super(decoder, encoder, initialSettings);
 
         this.apnsClient = apnsClient;
+        this.maxUnflushedNotifications = maxUnflushedNotifications;
     }
 
     @Override
     @SuppressWarnings("unchecked")
-    public void write(final ChannelHandlerContext context, final Object message, final ChannelPromise writePromise) {
+    public void write(final ChannelHandlerContext context, final Object message, final ChannelPromise writePromise) throws Http2Exception {
         try {
             // We'll catch class cast issues gracefully
             final T pushNotification = (T) message;
@@ -244,6 +259,10 @@ class ApnsClientHandler<T extends ApnsPushNotification> extends Http2ConnectionH
 
             this.nextStreamId += 2;
 
+            if (++this.unflushedNotifications >= this.maxUnflushedNotifications) {
+                this.flush(context);
+            }
+
             if (this.nextStreamId >= STREAM_ID_RESET_THRESHOLD) {
                 // This is very unlikely, but in the event that we run out of stream IDs (the maximum allowed is
                 // 2^31, per https://httpwg.github.io/specs/rfc7540.html#StreamIdentifiers), we need to open a new
@@ -260,37 +279,51 @@ class ApnsClientHandler<T extends ApnsPushNotification> extends Http2ConnectionH
     }
 
     @Override
-    public void userEventTriggered(final ChannelHandlerContext context, final Object event) throws Exception {
-        if (event instanceof IdleStateEvent) {
-
-            assert PING_TIMEOUT < ApnsClient.PING_IDLE_TIME;
-
-            log.trace("Sending ping due to inactivity.");
+    public void flush(final ChannelHandlerContext context) throws Http2Exception {
+        super.flush(context);
 
-            final ByteBuf pingDataBuffer = context.alloc().ioBuffer(8, 8);
-            pingDataBuffer.writeLong(this.nextPingId++);
+        this.unflushedNotifications = 0;
+    }
 
-            this.encoder().writePing(context, false, pingDataBuffer, context.newPromise()).addListener(new GenericFutureListener<ChannelFuture>() {
+    @Override
+    public void userEventTriggered(final ChannelHandlerContext context, final Object event) throws Exception {
+        if (event instanceof IdleStateEvent) {
+            final IdleStateEvent idleStateEvent = (IdleStateEvent) event;
 
-                @Override
-                public void operationComplete(final ChannelFuture future) throws Exception {
-                    if (future.isSuccess()) {
-                        ApnsClientHandler.this.pingTimeoutFuture = future.channel().eventLoop().schedule(new Runnable() {
-
-                            @Override
-                            public void run() {
-                                log.debug("Closing channel due to ping timeout.");
-                                future.channel().close();
-                            }
-                        }, PING_TIMEOUT, TimeUnit.SECONDS);
-                    } else {
-                        log.debug("Failed to write PING frame.", future.cause());
-                        future.channel().close();
-                    }
+            if (IdleState.WRITER_IDLE.equals(idleStateEvent.state())) {
+                if (idleStateEvent.isFirst()) {
+                    this.flush(context);
                 }
-            });
+            } else {
+                assert PING_TIMEOUT < ApnsClient.PING_IDLE_TIME_MILLIS;
+
+                log.trace("Sending ping due to inactivity.");
+
+                final ByteBuf pingDataBuffer = context.alloc().ioBuffer(8, 8);
+                pingDataBuffer.writeLong(this.nextPingId++);
+
+                this.encoder().writePing(context, false, pingDataBuffer, context.newPromise()).addListener(new GenericFutureListener<ChannelFuture>() {
+
+                    @Override
+                    public void operationComplete(final ChannelFuture future) throws Exception {
+                        if (future.isSuccess()) {
+                            ApnsClientHandler.this.pingTimeoutFuture = future.channel().eventLoop().schedule(new Runnable() {
+
+                                @Override
+                                public void run() {
+                                    log.debug("Closing channel due to ping timeout.");
+                                    future.channel().close();
+                                }
+                            }, PING_TIMEOUT, TimeUnit.SECONDS);
+                        } else {
+                            log.debug("Failed to write PING frame.", future.cause());
+                            future.channel().close();
+                        }
+                    }
+                });
 
-            context.flush();
+                this.flush(context);
+            }
         }
 
         super.userEventTriggered(context, event);


Test patch:
diff --git a/src/test/java/com/relayrides/pushy/apns/ApnsClientTest.java b/src/test/java/com/relayrides/pushy/apns/ApnsClientTest.java
index 2c2405f5..81f0355f 100644
--- a/src/test/java/com/relayrides/pushy/apns/ApnsClientTest.java
+++ b/src/test/java/com/relayrides/pushy/apns/ApnsClientTest.java
@@ -299,6 +299,23 @@ public class ApnsClientTest {
         assertTrue(managedGroupClient.disconnect().await().isSuccess());
     }
 
+    @Test
+    public void testSetFlushThresholds() {
+        // We're happy here as long as nothing explodes
+        this.client.setFlushThresholds(0, 0);
+        this.client.setFlushThresholds(1, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testSetFlushThresholdsWithZeroNotificationCount() {
+        this.client.setFlushThresholds(0, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testSetFlushThresholdsWithZeroTimeout() {
+        this.client.setFlushThresholds(1,  0);
+    }
+
     @Test
     public void testRestartApnsClientWithManagedEventLoopGroup() throws Exception {
         final ApnsClient<SimpleApnsPushNotification> managedGroupClient = new ApnsClient<>(
@@ -413,6 +430,23 @@ public class ApnsClientTest {
         assertTrue(response.isAccepted());
     }
 
+    @Test
+    public void testSendNotificationWithImmediateFlush() throws Exception {
+        this.client.disconnect().await();
+        this.client.setFlushThresholds(0, 0);
+        this.client.connect(HOST, PORT).await();
+
+        final String testToken = ApnsClientTest.generateRandomToken();
+
+        this.server.registerToken(DEFAULT_TOPIC, testToken);
+
+        final SimpleApnsPushNotification pushNotification = new SimpleApnsPushNotification(testToken, null, "test-payload");
+        final PushNotificationResponse<SimpleApnsPushNotification> response =
+                this.client.sendNotification(pushNotification).get();
+
+        assertTrue(response.isAccepted());
+    }
+
     @Test
     public void testSendManyNotifications() throws Exception {
         final int notificationCount = 1000;