Repository: jchambers/pushy
Pull request #757 (2020-03-06 00:30:14)

Return CompletableFutures for public-facing operations
As a follow-up to #746, this pull request replaces all public-facing uses of [`io.netty.util.concurrent.Future`](https://netty.io/4.1/api/io/netty/util/concurrent/Future.html) with [`java.util.concurrent.CompletableFuture`](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html). `CompletableFuture` is part of the stock Java SDK and offers all of the affordances—most importantly being able to add listeners—that Netty's `Future` offers.

jchambers: I should also point out that this pull request highlights how much non-IO stuff we're inadvertently doing on IO threads. I'll open a follow-up pull request to add an "insulation" layer.
jchambers: Also, I'm trying to figure out how to describe the action of calling `CompletableFuture#thenApply(...)`. "Add a listener" seems imprecise since `CompletionStage` is its own concept in the Java world, but "add a dependent completion stage" also feels a little obtuse for folks who aren't already familiar with `CompletableFuture`/`CompletionStage`. I'm rolling with "add a follow-up task" for now, but would love suggestions for alternative phrasing!

Patch:
diff --git a/README.md b/README.md
index aab38c76..c9d82a41 100644
--- a/README.md
+++ b/README.md
@@ -92,18 +92,18 @@ final SimpleApnsPushNotification pushNotification;
 }
 ```
 
-The process of sending a push notification is asynchronous; although the process of sending a notification and getting a reply from the server may take some time, the client will return a [`io.netty.util.concurrent.Future`](http://netty.io/4.1/api/io/netty/util/concurrent/Future.html) right away. You can use that `Future` to track the progress and eventual outcome of the sending operation. Note that an `io.netty.util.concurrent.Future` is an extension of the Java [`Future`](http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Future.html) interface that allows callers to add listeners and adds methods for checking the status of the `Future`.
+The process of sending a push notification is asynchronous; although the process of sending a notification and getting a reply from the server may take some time, the client will return a [`CompletableFuture`](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html) right away. You can use that `CompletableFuture` to track the progress and eventual outcome of the sending operation. Note that sending a notification returns a [`PushNotificationFuture`](https://pushy-apns.org/apidocs/0.13/com/eatthepath/pushy/apns/util/concurrent/PushNotificationFuture.html), which is a subclass of `CompletableFuture` that always holds a reference to the notification that was sent.
 
 ```java
 final PushNotificationFuture<SimpleApnsPushNotification, PushNotificationResponse<SimpleApnsPushNotification>>
     sendNotificationFuture = apnsClient.sendNotification(pushNotification);
 ```
 
-The `Future` will complete in one of three circumstances:
+The `CompletableFuture` will complete in one of three circumstances:
 
 1. The gateway accepts the notification and will attempt to deliver it to the destination device.
 2. The gateway rejects the notification; this should be considered a permanent failure, and the notification should not be sent again. Additionally, the APNs gateway may indicate a timestamp at which the destination token became invalid. If that happens, you should stop trying to send *any* notification to that token unless the token has been re-registered since that timestamp.
-3. The `Future` fails with an exception. This should generally be considered a temporary failure, and callers should try to send the notification again when the problem has been resolved.
+3. The `CompletableFuture` fails with an exception. This should generally be considered a temporary failure, and callers should try to send the notification again when the problem has been resolved.
 
 An example:
 
@@ -129,23 +129,18 @@ try {
 }
 ```
 
-Again, it's important to note that the returned `Future` supports listeners; waiting for each individual push notification is inefficient in practice, and most users will be better serverd by adding a listener to the `Future` instead of blocking until it completes. As an example:
+It's important to note that `CompletableFuture` has affordances for scheduling additional tasks to run when an operation is complete. Waiting for each individual push notification is inefficient in practice, and most users will be better served by adding follow-up tasks to the `CompletableFuture` instead of blocking until it completes. As an example:
 
 ```java
-sendNotificationFuture.addListener(future -> {
-    // When using a listener, callers should check for a failure to send a
-    // notification by checking whether the future itself was successful
-    // since an exception will not be thrown.
-    if (future.isSuccess()) {
-        final PushNotificationResponse<SimpleApnsPushNotification> pushNotificationResponse =
-                sendNotificationFuture.getNow();
-
+sendNotificationFuture.whenComplete((response, cause) -> {
+    if (response != null) {
         // Handle the push notification response as before from here.
     } else {
         // Something went wrong when trying to send the notification to the
-        // APNs gateway. We can find the exception that caused the failure
-        // by getting future.cause().
-        future.cause().printStackTrace();
+        // APNs server. Note that this is distinct from a rejection from
+        // the server, and indicates that something went wrong when actually
+        // sending the notification or waiting for a reply.
+        cause.printStackTrace();
     }
 });
 ```
@@ -153,8 +148,7 @@ sendNotificationFuture.addListener(future -> {
 All APNs clients—even those that have never sent a message—may allocate and hold on to system resources, and it's important to release them. APNs clients are intended to be persistent, long-lived resources; you definitely don't need to shut down a client after sending a notification (or even batch of notifications), but you'll want to shut down your client (or clients) when your application is shutting down:
 
 ```java
-final Future<Void> closeFuture = apnsClient.close();
-closeFuture.await();
+final CompletableFuture<Void> closeFuture = apnsClient.close();
 ```
 
 When shutting down, clients will wait for all sent-but-not-acknowledged notifications to receive a reply from the server. Notifications that have been passed to `sendNotification` but not yet sent to the server (i.e. notifications waiting in an internal queue) will fail immediately when disconnecting. Callers should generally make sure that all sent notifications have been acknowledged by the server before shutting down.
@@ -164,7 +158,7 @@ When shutting down, clients will wait for all sent-but-not-acknowledged notifica
 Making the most of your system resources for high-throughput applications always takes some effort. To guide you through the process, we've put together a wiki page covering some [best practices for using Pushy](https://github.com/jchambers/pushy/wiki/Best-practices). All of these points are covered in much more detail on the wiki, but in general, our recommendations are:
 
 - Treat `ApnsClient` instances as long-lived resources
-- Use listeners if you want to track the status of your push notifications
+- Add follow-up tasks to `CompletableFutures` if you want to track the status of your push notifications
 - Use a flow control strategy to avoid enqueueing push notifications faster than the server can respond
 - Choose a number of threads and concurrent connections that balances CPU time and network throughput
 
diff --git a/benchmark/src/main/java/com/eatthepath/pushy/apns/ApnsClientBenchmark.java b/benchmark/src/main/java/com/eatthepath/pushy/apns/ApnsClientBenchmark.java
index 585ee8de..44e6332c 100644
--- a/benchmark/src/main/java/com/eatthepath/pushy/apns/ApnsClientBenchmark.java
+++ b/benchmark/src/main/java/com/eatthepath/pushy/apns/ApnsClientBenchmark.java
@@ -111,7 +111,7 @@ public class ApnsClientBenchmark {
             }
         }
 
-        this.server.start(PORT).await();
+        this.server.start(PORT).get();
     }
 
     @Benchmark
@@ -123,11 +123,7 @@ public class ApnsClientBenchmark {
         final CountDownLatch countDownLatch = new CountDownLatch(this.pushNotifications.size());
 
         for (final SimpleApnsPushNotification notification : this.pushNotifications) {
-            this.client.sendNotification(notification).addListener(future -> {
-                if (future.isSuccess()) {
-                    countDownLatch.countDown();
-                }
-            });
+            this.client.sendNotification(notification).thenRun(countDownLatch::countDown);
         }
 
         countDownLatch.await();
@@ -136,8 +132,8 @@ public class ApnsClientBenchmark {
 
     @TearDown
     public void tearDown() throws Exception {
-        this.client.close().await();
-        this.server.shutdown().await();
+        this.client.close().get();
+        this.server.shutdown().get();
 
         final Future<?> clientShutdownFuture = this.clientEventLoopGroup.shutdownGracefully();
         final Future<?> serverShutdownFuture = this.serverEventLoopGroup.shutdownGracefully();
diff --git a/pushy/src/main/java/com/eatthepath/pushy/apns/ApnsClient.java b/pushy/src/main/java/com/eatthepath/pushy/apns/ApnsClient.java
index 5ce9d469..c57e3b50 100644
--- a/pushy/src/main/java/com/eatthepath/pushy/apns/ApnsClient.java
+++ b/pushy/src/main/java/com/eatthepath/pushy/apns/ApnsClient.java
@@ -25,7 +25,6 @@ package com.eatthepath.pushy.apns;
 import com.eatthepath.pushy.apns.auth.ApnsSigningKey;
 import com.eatthepath.pushy.apns.proxy.ProxyHandlerFactory;
 import com.eatthepath.pushy.apns.util.concurrent.PushNotificationFuture;
-import com.eatthepath.pushy.apns.util.concurrent.PushNotificationResponseListener;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.EventLoopGroup;
@@ -38,6 +37,7 @@ import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
 import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -66,12 +66,9 @@ import java.util.concurrent.atomic.AtomicLong;
  * as they're constructed.</p>
  *
  * <p>Notifications sent by a client to an APNs server are sent asynchronously. A
- * {@link io.netty.util.concurrent.Future io.netty.util.concurrent.Future} is returned immediately when a notification
- * is sent, but will not complete until the attempt to send the notification has failed, the notification has been
- * accepted by the APNs server, or the notification has been rejected by the APNs server. Please note that the
- * {@code Future} returned is a {@code io.netty.util.concurrent.Future}, which is an extension of the
- * {@link java.util.concurrent.Future java.util.concurrent.Future} interface that allows callers to attach listeners
- * that will be notified when the {@code Future} completes.</p>
+ * {@link CompletableFuture} is returned immediately when a notification is sent, but will not complete until the
+ * attempt to send the notification has failed, the notification has been accepted by the APNs server, or the
+ * notification has been rejected by the APNs server.</p>
  *
  * <p>APNs clients are intended to be long-lived, persistent resources. They are also inherently thread-safe and can be
  * shared across many threads in a complex application. Callers must shut them down via the {@link ApnsClient#close()}
@@ -192,24 +189,20 @@ public class ApnsClient {
      * @return a {@code Future} that will complete when the notification has been either accepted or rejected by the
      * APNs gateway
      *
-     * @see com.eatthepath.pushy.apns.util.concurrent.PushNotificationResponseListener
-     *
      * @since 0.8
      */
     public <T extends ApnsPushNotification> PushNotificationFuture<T, PushNotificationResponse<T>> sendNotification(final T notification) {
-        final PushNotificationFuture<T, PushNotificationResponse<T>> responseFuture;
+        final PushNotificationFuture<T, PushNotificationResponse<T>> responseFuture =
+                new PushNotificationFuture<>(notification);
 
         if (!this.isClosed.get()) {
-            final PushNotificationPromise<T, PushNotificationResponse<T>> responsePromise =
-                    new PushNotificationPromise<>(this.eventLoopGroup.next(), notification);
-
             final long notificationId = this.nextNotificationId.getAndIncrement();
 
             this.channelPool.acquire().addListener((GenericFutureListener<Future<Channel>>) acquireFuture -> {
                 if (acquireFuture.isSuccess()) {
                     final Channel channel = acquireFuture.getNow();
 
-                    channel.writeAndFlush(responsePromise).addListener((GenericFutureListener<ChannelFuture>) future -> {
+                    channel.writeAndFlush(responseFuture).addListener((GenericFutureListener<ChannelFuture>) future -> {
                         if (future.isSuccess()) {
                             ApnsClient.this.metricsListener.handleNotificationSent(ApnsClient.this, notificationId);
                         }
@@ -217,14 +210,12 @@ public class ApnsClient {
 
                     ApnsClient.this.channelPool.release(channel);
                 } else {
-                    responsePromise.tryFailure(acquireFuture.cause());
+                    responseFuture.completeExceptionally(acquireFuture.cause());
                 }
             });
 
-            responsePromise.addListener((PushNotificationResponseListener<T>) future -> {
-                if (future.isSuccess()) {
-                    final PushNotificationResponse<T> response = future.getNow();
-
+            responseFuture.whenComplete((response, cause) -> {
+                if (response != null) {
                     if (response.isAccepted()) {
                         ApnsClient.this.metricsListener.handleNotificationAccepted(ApnsClient.this, notificationId);
                     } else {
@@ -234,15 +225,8 @@ public class ApnsClient {
                     ApnsClient.this.metricsListener.handleWriteFailure(ApnsClient.this, notificationId);
                 }
             });
-
-            responseFuture = responsePromise;
         } else {
-            final PushNotificationPromise<T, PushNotificationResponse<T>> failedPromise =
-                    new PushNotificationPromise<>(GlobalEventExecutor.INSTANCE, notification);
-
-            failedPromise.setFailure(CLIENT_CLOSED_EXCEPTION);
-
-            responseFuture = failedPromise;
+            responseFuture.completeExceptionally(CLIENT_CLOSED_EXCEPTION);
         }
 
         return responseFuture;
@@ -266,28 +250,23 @@ public class ApnsClient {
      *
      * @since 0.11
      */
-    @SuppressWarnings({ "rawtypes", "unchecked" })
-    public Future<Void> close() {
+    public CompletableFuture<Void> close() {
         log.info("Shutting down.");
 
-        final Future<Void> closeFuture;
+        final CompletableFuture<Void> closeFuture;
 
         if (this.isClosed.compareAndSet(false, true)) {
-            // Since we're (maybe) going to clobber the main event loop group, we should have this promise use the global
-            // event executor to notify listeners.
-            final Promise<Void> closePromise = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
+            closeFuture = new CompletableFuture<>();
 
             this.channelPool.close().addListener((GenericFutureListener<Future<Void>>) closePoolFuture -> {
                 if (ApnsClient.this.shouldShutDownEventLoopGroup) {
-                    ApnsClient.this.eventLoopGroup.shutdownGracefully().addListener((GenericFutureListener) future -> closePromise.trySuccess(null));
+                    ApnsClient.this.eventLoopGroup.shutdownGracefully().addListener(future -> closeFuture.complete(null));
                 } else {
-                    closePromise.trySuccess(null);
+                    closeFuture.complete(null);
                 }
             });
-
-            closeFuture = closePromise;
         } else {
-            closeFuture = new SucceededFuture<>(GlobalEventExecutor.INSTANCE, null);
+            closeFuture = CompletableFuture.completedFuture(null);
         }
 
         return closeFuture;
diff --git a/pushy/src/main/java/com/eatthepath/pushy/apns/ApnsClientHandler.java b/pushy/src/main/java/com/eatthepath/pushy/apns/ApnsClientHandler.java
index 1e5eda40..ca34c394 100644
--- a/pushy/src/main/java/com/eatthepath/pushy/apns/ApnsClientHandler.java
+++ b/pushy/src/main/java/com/eatthepath/pushy/apns/ApnsClientHandler.java
@@ -23,6 +23,7 @@
 package com.eatthepath.pushy.apns;
 
 import com.eatthepath.pushy.apns.util.InstantAsTimeSinceEpochTypeAdapter;
+import com.eatthepath.pushy.apns.util.concurrent.PushNotificationFuture;
 import com.eatthepath.uuid.FastUUID;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
@@ -53,11 +54,12 @@ import java.time.Instant;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
 class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameListener, Http2Connection.Listener {
 
-    private final Map<Integer, PushNotificationPromise<?, ?>> unattachedResponsePromisesByStreamId = new IntObjectHashMap<>();
+    private final Map<Integer, PushNotificationFuture<?, ?>> unattachedResponsePromisesByStreamId = new IntObjectHashMap<>();
 
     private final Http2Connection.PropertyKey responseHeadersPropertyKey;
     private final Http2Connection.PropertyKey responsePromisePropertyKey;
@@ -164,17 +166,17 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
 
     @Override
     public void write(final ChannelHandlerContext context, final Object message, final ChannelPromise writePromise) {
-        if (message instanceof PushNotificationPromise) {
-            final PushNotificationPromise<?, ?> pushNotificationPromise = (PushNotificationPromise<?, ?>) message;
+        if (message instanceof PushNotificationFuture) {
+            final PushNotificationFuture<?, ?> pushNotificationFuture = (PushNotificationFuture<?, ?>) message;
 
             writePromise.addListener(future -> {
                 if (!future.isSuccess()) {
                     log.trace("Failed to write push notification.", future.cause());
-                    pushNotificationPromise.tryFailure(future.cause());
+                    pushNotificationFuture.completeExceptionally(future.cause());
                 }
             });
 
-            this.writePushNotification(context, pushNotificationPromise, writePromise);
+            this.writePushNotification(context, pushNotificationFuture, writePromise);
         } else {
             // This should never happen, but in case some foreign debris winds up in the pipeline, just pass it through.
             log.error("Unexpected object in pipeline: {}", message);
@@ -185,13 +187,13 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
     void retryPushNotificationFromStream(final ChannelHandlerContext context, final int streamId) {
         final Http2Stream stream = this.connection().stream(streamId);
 
-        final PushNotificationPromise<?, ?> responsePromise = stream.removeProperty(this.responsePromisePropertyKey);
+        final PushNotificationFuture<?, ?> responseFuture = stream.removeProperty(this.responsePromisePropertyKey);
 
         final ChannelPromise writePromise = context.channel().newPromise();
-        this.writePushNotification(context, responsePromise, writePromise);
+        this.writePushNotification(context, responseFuture, writePromise);
     }
 
-    private void writePushNotification(final ChannelHandlerContext context, final PushNotificationPromise<?, ?> responsePromise, final ChannelPromise writePromise) {
+    private void writePushNotification(final ChannelHandlerContext context, final PushNotificationFuture<?, ?> responsePromise, final ChannelPromise writePromise) {
         if (context.channel().isActive()) {
             final int streamId = this.connection().local().incrementAndGetNextStreamId();
 
@@ -318,15 +320,15 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
 
     private void handleEndOfStream(final ChannelHandlerContext context, final Http2Stream stream, final Http2Headers headers, final ByteBuf data) {
 
-        final PushNotificationPromise<ApnsPushNotification, PushNotificationResponse<ApnsPushNotification>> responsePromise =
+        final PushNotificationFuture<?, PushNotificationResponse<?>> responseFuture =
                 stream.getProperty(this.responsePromisePropertyKey);
 
-        final ApnsPushNotification pushNotification = responsePromise.getPushNotification();
+        final ApnsPushNotification pushNotification = responseFuture.getPushNotification();
 
         final HttpResponseStatus status = HttpResponseStatus.parseLine(headers.status());
 
         if (HttpResponseStatus.OK.equals(status)) {
-            responsePromise.trySuccess(new SimplePushNotificationResponse<>(responsePromise.getPushNotification(),
+            responseFuture.complete(new SimplePushNotificationResponse<>(responseFuture.getPushNotification(),
                     true, getApnsIdFromHeaders(headers), null, null));
         } else {
             if (data != null) {
@@ -339,12 +341,12 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
     }
 
     protected void handleErrorResponse(final ChannelHandlerContext context, final int streamId, final Http2Headers headers, final ApnsPushNotification pushNotification, final ErrorResponse errorResponse) {
-        final PushNotificationPromise<ApnsPushNotification, PushNotificationResponse<ApnsPushNotification>> responsePromise =
+        final PushNotificationFuture<?, PushNotificationResponse<?>> responseFuture =
                 this.connection().stream(streamId).getProperty(this.responsePromisePropertyKey);
 
         final HttpResponseStatus status = HttpResponseStatus.parseLine(headers.status());
 
-        responsePromise.trySuccess(new SimplePushNotificationResponse<>(responsePromise.getPushNotification(),
+        responseFuture.complete(new SimplePushNotificationResponse<>(responseFuture.getPushNotification(),
                 HttpResponseStatus.OK.equals(status), getApnsIdFromHeaders(headers), errorResponse.getReason(),
                 errorResponse.getTimestamp()));
     }
@@ -436,7 +438,7 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
     public void onStreamClosed(final Http2Stream stream) {
         // Always try to fail promises associated with closed streams; most of the time, this should fail silently, but
         // in cases of unexpected closure, it will make sure that nothing gets left hanging.
-        final Promise<PushNotificationResponse<ApnsPushNotification>> responsePromise =
+        final CompletableFuture<PushNotificationResponse<ApnsPushNotification>> responsePromise =
                 stream.getProperty(this.responsePromisePropertyKey);
 
         if (responsePromise != null) {
@@ -450,7 +452,7 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
                 cause = STREAM_CLOSED_BEFORE_REPLY_EXCEPTION;
             }
 
-            responsePromise.tryFailure(cause);
+            responsePromise.completeExceptionally(cause);
         }
     }
 
@@ -487,8 +489,8 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
     public void channelInactive(final ChannelHandlerContext context) throws Exception {
         super.channelInactive(context);
 
-        for (final PushNotificationPromise<?, ?> promise : this.unattachedResponsePromisesByStreamId.values()) {
-            promise.tryFailure(STREAM_CLOSED_BEFORE_REPLY_EXCEPTION);
+        for (final PushNotificationFuture<?, ?> future : this.unattachedResponsePromisesByStreamId.values()) {
+            future.completeExceptionally(STREAM_CLOSED_BEFORE_REPLY_EXCEPTION);
         }
 
         this.unattachedResponsePromisesByStreamId.clear();
diff --git a/pushy/src/main/java/com/eatthepath/pushy/apns/PushNotificationPromise.java b/pushy/src/main/java/com/eatthepath/pushy/apns/PushNotificationPromise.java
deleted file mode 100644
index 1feab73f..00000000
--- a/pushy/src/main/java/com/eatthepath/pushy/apns/PushNotificationPromise.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * Copyright (c) 2020 Jon Chambers
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in
- * all copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
- * THE SOFTWARE.
- */
-
-package com.eatthepath.pushy.apns;
-
-import com.eatthepath.pushy.apns.util.concurrent.PushNotificationFuture;
-import io.netty.util.concurrent.DefaultPromise;
-import io.netty.util.concurrent.EventExecutor;
-
-class PushNotificationPromise<P extends ApnsPushNotification, V> extends DefaultPromise<V> implements PushNotificationFuture<P, V> {
-
-    private final P pushNotification;
-
-    PushNotificationPromise(final EventExecutor eventExecutor, final P pushNotification) {
-        super(eventExecutor);
-
-        this.pushNotification = pushNotification;
-    }
-
-    @Override
-    public P getPushNotification() {
-        return this.pushNotification;
-    }
-}
diff --git a/pushy/src/main/java/com/eatthepath/pushy/apns/server/BaseHttp2Server.java b/pushy/src/main/java/com/eatthepath/pushy/apns/server/BaseHttp2Server.java
index d63eaeec..8170e09f 100644
--- a/pushy/src/main/java/com/eatthepath/pushy/apns/server/BaseHttp2Server.java
+++ b/pushy/src/main/java/com/eatthepath/pushy/apns/server/BaseHttp2Server.java
@@ -36,6 +36,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.net.ssl.SSLSession;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 abstract class BaseHttp2Server {
@@ -113,11 +114,21 @@ abstract class BaseHttp2Server {
      * @return a {@code Future} that will succeed when the server has bound to the given port and is ready to accept
      * traffic
      */
-    public Future<Void> start(final int port) {
+    public CompletableFuture<Void> start(final int port) {
         final ChannelFuture channelFuture = this.bootstrap.bind(port);
         this.allChannels.add(channelFuture.channel());
 
-        return channelFuture;
+        final CompletableFuture<Void> startFuture = new CompletableFuture<>();
+
+        channelFuture.addListener((future -> {
+            if (future.isSuccess()) {
+                startFuture.complete(null);
+            } else {
+                startFuture.completeExceptionally(future.cause());
+            }
+        }));
+
+        return startFuture;
     }
 
     /**
@@ -132,35 +143,39 @@ abstract class BaseHttp2Server {
      * @return a {@code Future} that will succeed once the server has finished unbinding from its port and, if the
      * server was managing its own event loop group, its event loop group has shut down
      */
-    public Future<Void> shutdown() {
+    public CompletableFuture<Void> shutdown() {
+        final CompletableFuture<Void> shutdownFuture = new CompletableFuture<>();
         final Future<Void> channelCloseFuture = this.allChannels.close();
 
-        final Future<Void> disconnectFuture;
-
         if (this.shouldShutDownEventLoopGroup) {
             // Wait for the channel to close before we try to shut down the event loop group
             channelCloseFuture.addListener(future ->
                     BaseHttp2Server.this.bootstrap.config().group().shutdownGracefully());
 
-            // Since the termination future for the event loop group is a Future<?> instead of a Future<Void>,
-            // we'll need to create our own promise and then notify it when the termination future completes.
-            disconnectFuture = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
-
-            this.bootstrap.config().group().terminationFuture().addListener(future ->
-                    ((Promise<Void>) disconnectFuture).trySuccess(null));
+            this.bootstrap.config().group().terminationFuture().addListener(future -> {
+                if (future.isSuccess()) {
+                    shutdownFuture.complete(null);
+                } else {
+                    shutdownFuture.completeExceptionally(future.cause());
+                }
+            });
         } else {
-            // We're done once we've closed all the channels, so we can return the closure future directly.
-            disconnectFuture = channelCloseFuture;
+            // We're done once we've closed all the channels.
+            channelCloseFuture.addListener(future -> {
+                if (future.isSuccess()) {
+                    shutdownFuture.complete(null);
+                } else {
+                    shutdownFuture.completeExceptionally(future.cause());
+                }
+            });
         }
 
-        disconnectFuture.addListener((GenericFutureListener<Future<Void>>) future -> {
+        return shutdownFuture.thenRun(() -> {
             if (BaseHttp2Server.this.sslContext instanceof ReferenceCounted) {
                 if (BaseHttp2Server.this.hasReleasedSslContext.compareAndSet(false, true)) {
                     ((ReferenceCounted) BaseHttp2Server.this.sslContext).release();
                 }
             }
         });
-
-        return disconnectFuture;
     }
 }
diff --git a/pushy/src/main/java/com/eatthepath/pushy/apns/util/concurrent/PushNotificationFuture.java b/pushy/src/main/java/com/eatthepath/pushy/apns/util/concurrent/PushNotificationFuture.java
index d1ddb19b..43eaf9c3 100644
--- a/pushy/src/main/java/com/eatthepath/pushy/apns/util/concurrent/PushNotificationFuture.java
+++ b/pushy/src/main/java/com/eatthepath/pushy/apns/util/concurrent/PushNotificationFuture.java
@@ -23,22 +23,33 @@
 package com.eatthepath.pushy.apns.util.concurrent;
 
 import com.eatthepath.pushy.apns.ApnsPushNotification;
-import io.netty.util.concurrent.Future;
+
+import java.util.concurrent.CompletableFuture;
 
 /**
- * A push notification future represents the result an operation on a push notification.
+ * A push notification future represents the result an asynchronous operation on a {@link ApnsPushNotification}.
  *
  * @param <P> the type of push notification sent
  * @param <V> the type of value returned by the operation
  *
  * @author <a href="https://github.com/jchambers">Jon Chambers</a>
  */
-public interface PushNotificationFuture<P extends ApnsPushNotification, V> extends Future<V> {
+public class PushNotificationFuture<P extends ApnsPushNotification, V> extends CompletableFuture<V> {
+
+    private final P pushNotification;
+
+    public PushNotificationFuture(final P pushNotification) {
+        super();
+
+        this.pushNotification = pushNotification;
+    }
 
     /**
      * Returns the push notification to which the operation represented by this future applies.
      *
      * @return the push notification to which the operation represented by this future applies
      */
-    P getPushNotification();
+    public P getPushNotification() {
+        return this.pushNotification;
+    }
 }
diff --git a/pushy/src/main/java/com/eatthepath/pushy/apns/util/concurrent/PushNotificationResponseListener.java b/pushy/src/main/java/com/eatthepath/pushy/apns/util/concurrent/PushNotificationResponseListener.java
deleted file mode 100644
index b5c8d475..00000000
--- a/pushy/src/main/java/com/eatthepath/pushy/apns/util/concurrent/PushNotificationResponseListener.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Copyright (c) 2020 Jon Chambers
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in
- * all copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
- * THE SOFTWARE.
- */
-
-package com.eatthepath.pushy.apns.util.concurrent;
-
-import com.eatthepath.pushy.apns.ApnsPushNotification;
-import com.eatthepath.pushy.apns.PushNotificationResponse;
-import io.netty.util.concurrent.GenericFutureListener;
-
-/**
- * A type-specific convenience interface for listening for the result of an attempt to send a push notification.
- *
- * @param <T> the type of push notification sent
- *
- * @see com.eatthepath.pushy.apns.ApnsClient#sendNotification(ApnsPushNotification)
- */
-public interface PushNotificationResponseListener<T extends ApnsPushNotification>
-        extends GenericFutureListener<PushNotificationFuture<T, PushNotificationResponse<T>>> {
-}


Test patch:
diff --git a/pushy/src/test/java/com/eatthepath/pushy/apns/ApnsClientBuilderTest.java b/pushy/src/test/java/com/eatthepath/pushy/apns/ApnsClientBuilderTest.java
index 11c6de1a..da606a97 100644
--- a/pushy/src/test/java/com/eatthepath/pushy/apns/ApnsClientBuilderTest.java
+++ b/pushy/src/test/java/com/eatthepath/pushy/apns/ApnsClientBuilderTest.java
@@ -65,7 +65,7 @@ public class ApnsClientBuilderTest {
                 .setClientCredentials(new File(this.getClass().getResource(SINGLE_TOPIC_CLIENT_KEYSTORE_FILENAME).toURI()), KEYSTORE_PASSWORD)
                 .build();
 
-        client.close().await();
+        client.close().get();
     }
 
     @Test
@@ -78,7 +78,7 @@ public class ApnsClientBuilderTest {
                     .setClientCredentials(p12InputStream, KEYSTORE_PASSWORD)
                     .build();
 
-            client.close().await();
+            client.close().get();
         }
     }
 
@@ -103,7 +103,7 @@ public class ApnsClientBuilderTest {
                     .setClientCredentials((X509Certificate) privateKeyEntry.getCertificate(), privateKeyEntry.getPrivateKey(), KEYSTORE_PASSWORD)
                     .build();
 
-            client.close().await();
+            client.close().get();
         }
     }
 
@@ -121,7 +121,7 @@ public class ApnsClientBuilderTest {
                     .setClientCredentials((X509Certificate) privateKeyEntry.getCertificate(), privateKeyEntry.getPrivateKey(), null)
                     .build();
 
-            client.close().await();
+            client.close().get();
         }
     }
 
@@ -137,7 +137,7 @@ public class ApnsClientBuilderTest {
                     .setSigningKey(signingKey)
                     .build();
 
-            client.close().await();
+            client.close().get();
         }
     }
 
diff --git a/pushy/src/test/java/com/eatthepath/pushy/apns/ApnsClientTest.java b/pushy/src/test/java/com/eatthepath/pushy/apns/ApnsClientTest.java
index 2a453e13..2c30f541 100644
--- a/pushy/src/test/java/com/eatthepath/pushy/apns/ApnsClientTest.java
+++ b/pushy/src/test/java/com/eatthepath/pushy/apns/ApnsClientTest.java
@@ -24,7 +24,7 @@ package com.eatthepath.pushy.apns;
 
 import com.eatthepath.pushy.apns.server.*;
 import com.eatthepath.pushy.apns.util.SimpleApnsPushNotification;
-import io.netty.util.concurrent.Future;
+import com.eatthepath.pushy.apns.util.concurrent.PushNotificationFuture;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -36,7 +36,9 @@ import java.io.InputStream;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Stream;
@@ -208,7 +210,7 @@ public class ApnsClientTest extends AbstractClientServerTest {
                 .setSigningKey(this.signingKey)
                 .build();
 
-        assertTrue(managedGroupClient.close().await().isSuccess());
+        assertDoesNotThrow(() -> managedGroupClient.close().get());
     }
 
     @ParameterizedTest
@@ -235,18 +237,16 @@ public class ApnsClientTest extends AbstractClientServerTest {
         final MockApnsServer server = this.buildServer(new AcceptAllPushNotificationHandlerFactory());
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
-            final Future<PushNotificationResponse<SimpleApnsPushNotification>> sendFuture =
-                    cautiousClient.sendNotification(
-                            new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD)).await();
+            final PushNotificationFuture<SimpleApnsPushNotification, PushNotificationResponse<SimpleApnsPushNotification>> sendFuture =
+                    cautiousClient.sendNotification(new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD));
 
-            assertFalse(sendFuture.isSuccess(), "Clients must not connect to untrusted servers.");
+            Throwable cause = assertThrows(ExecutionException.class, sendFuture::get,
+                    "Clients must not connect to untrusted servers.");
 
             boolean hasSSLHandshakeException = false;
             {
-                Throwable cause = sendFuture.cause();
-
                 while (!hasSSLHandshakeException && cause != null) {
                     hasSSLHandshakeException = (cause instanceof SSLHandshakeException);
                     cause = cause.getCause();
@@ -256,8 +256,8 @@ public class ApnsClientTest extends AbstractClientServerTest {
             assertTrue(hasSSLHandshakeException,
                     "Clients should refuse to connect to untrusted servers due to an SSL handshake failure.");
         } finally {
-            cautiousClient.close().await();
-            server.shutdown().await();
+            cautiousClient.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -268,13 +268,13 @@ public class ApnsClientTest extends AbstractClientServerTest {
                 this.buildTokenAuthenticationClient() : this.buildTlsAuthenticationClient();
 
         try {
-            assertTrue(client.close().await().isSuccess(),
+            assertDoesNotThrow(() -> client.close().get(),
                     "Client should close successfully under normal circumstances.");
 
-            assertTrue(client.close().await().isSuccess(),
+            assertDoesNotThrow(() -> client.close().get(),
                     "Client should report successful closure on subsequent calls to close().");
         } finally {
-            client.close().await();
+            client.close().get();
         }
     }
 
@@ -286,20 +286,17 @@ public class ApnsClientTest extends AbstractClientServerTest {
                 this.buildTokenAuthenticationClient() : this.buildTlsAuthenticationClient();
 
         try {
-            server.start(PORT).await();
-            client.close().await();
+            server.start(PORT).get();
+            client.close().get();
 
             final SimpleApnsPushNotification pushNotification =
                     new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
 
-            final Future<PushNotificationResponse<SimpleApnsPushNotification>> sendFuture =
-                    client.sendNotification(pushNotification).await();
-
-            assertFalse(sendFuture.isSuccess(),
+            assertThrows(ExecutionException.class, () -> client.sendNotification(pushNotification).get(),
                     "Once a client has closed, attempts to send push notifications should fail.");
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -315,7 +312,7 @@ public class ApnsClientTest extends AbstractClientServerTest {
                 this.buildTokenAuthenticationClient() : this.buildTlsAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             final SimpleApnsPushNotification pushNotification = new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
 
@@ -327,8 +324,8 @@ public class ApnsClientTest extends AbstractClientServerTest {
 
             assertNotNull(response.getApnsId());
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -343,7 +340,7 @@ public class ApnsClientTest extends AbstractClientServerTest {
         final ApnsClient client = this.buildTokenAuthenticationClient(metricsListener);
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             final SimpleApnsPushNotification pushNotification =
                     new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
@@ -366,8 +363,8 @@ public class ApnsClientTest extends AbstractClientServerTest {
             assertTrue(metricsListener.getRejectedNotifications().isEmpty(),
                     "Notifications with expired authentication tokens should not count as rejections.");
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -385,28 +382,25 @@ public class ApnsClientTest extends AbstractClientServerTest {
             pushNotifications.add(new SimpleApnsPushNotification(token, TOPIC, payload));
         }
 
-        final List<Future<PushNotificationResponse<SimpleApnsPushNotification>>> futures = new ArrayList<>();
-
         final MockApnsServer server = this.buildServer(new AcceptAllPushNotificationHandlerFactory());
         final ApnsClient client = useTokenAuthentication ?
                 this.buildTokenAuthenticationClient() : this.buildTlsAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
+
+            final List<CompletableFuture<PushNotificationResponse<SimpleApnsPushNotification>>> futures =
+                    new ArrayList<>();
 
             for (final SimpleApnsPushNotification pushNotification : pushNotifications) {
                 futures.add(client.sendNotification(pushNotification));
             }
 
-            for (final Future<PushNotificationResponse<SimpleApnsPushNotification>> future : futures) {
-                future.await();
-
-                assertTrue(future.isSuccess(),
-                        "Send future should have succeeded, but failed with: " + future.cause());
-            }
+            //noinspection ZeroLengthArrayAllocation
+            assertDoesNotThrow(() -> CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get());
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -431,20 +425,16 @@ public class ApnsClientTest extends AbstractClientServerTest {
         final CountDownLatch countDownLatch = new CountDownLatch(notificationCount);
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             for (final SimpleApnsPushNotification pushNotification : pushNotifications) {
-                client.sendNotification(pushNotification).addListener(future -> {
-                    if (future.isSuccess()) {
-                        countDownLatch.countDown();
-                    }
-                });
+                client.sendNotification(pushNotification).thenRun(countDownLatch::countDown);
             }
 
             countDownLatch.await();
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -464,20 +454,18 @@ public class ApnsClientTest extends AbstractClientServerTest {
                 this.buildTokenAuthenticationClient() : this.buildTlsAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             for (int i = 0; i < notificationCount; i++) {
-                client.sendNotification(pushNotification).addListener(future -> {
-                    // All we're concerned with here is that the client told us SOMETHING about what happened to the
-                    // notification
-                    countDownLatch.countDown();
-                });
+                // All we're concerned with here is that the client told us SOMETHING about what happened to the
+                // notification
+                client.sendNotification(pushNotification).whenComplete((response, cause) -> countDownLatch.countDown());
             }
 
             countDownLatch.await();
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -498,7 +486,7 @@ public class ApnsClientTest extends AbstractClientServerTest {
                 this.buildTokenAuthenticationClient() : this.buildTlsAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             final PushNotificationResponse<SimpleApnsPushNotification> response =
                     client.sendNotification(pushNotification).get();
@@ -507,8 +495,8 @@ public class ApnsClientTest extends AbstractClientServerTest {
             assertEquals("Unregistered", response.getRejectionReason());
             assertEquals(expiration, response.getTokenInvalidationTimestamp());
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -522,12 +510,12 @@ public class ApnsClientTest extends AbstractClientServerTest {
                 this.buildTokenAuthenticationClient(metricsListener) : this.buildTlsAuthenticationClient(metricsListener);
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             final SimpleApnsPushNotification pushNotification =
                     new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
 
-            client.sendNotification(pushNotification).await();
+            client.sendNotification(pushNotification).get();
 
             metricsListener.waitForNonZeroAcceptedNotifications();
 
@@ -539,8 +527,8 @@ public class ApnsClientTest extends AbstractClientServerTest {
             assertEquals(0, metricsListener.getConnectionsRemoved().get());
             assertEquals(0, metricsListener.getFailedConnectionAttempts().get());
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -558,12 +546,12 @@ public class ApnsClientTest extends AbstractClientServerTest {
                 this.buildTokenAuthenticationClient(metricsListener) : this.buildTlsAuthenticationClient(metricsListener);
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             final SimpleApnsPushNotification pushNotification =
                     new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
 
-            client.sendNotification(pushNotification).await();
+            client.sendNotification(pushNotification).get();
 
             metricsListener.waitForNonZeroRejectedNotifications();
 
@@ -571,8 +559,8 @@ public class ApnsClientTest extends AbstractClientServerTest {
             assertEquals(metricsListener.getSentNotifications(), metricsListener.getRejectedNotifications());
             assertTrue(metricsListener.getAcceptedNotifications().isEmpty());
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -588,7 +576,8 @@ public class ApnsClientTest extends AbstractClientServerTest {
             final SimpleApnsPushNotification pushNotification =
                     new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
 
-            client.sendNotification(pushNotification).await();
+            // This should fail because there's no server running
+            assertThrows(ExecutionException.class, () -> client.sendNotification(pushNotification).get());
 
             metricsListener.waitForNonZeroFailedConnections();
             metricsListener.waitForNonZeroWriteFailures();
@@ -599,7 +588,7 @@ public class ApnsClientTest extends AbstractClientServerTest {
 
             assertEquals(1, metricsListener.getWriteFailures().size());
         } finally {
-            client.close().await();
+            client.close().get();
         }
     }
 
@@ -616,14 +605,14 @@ public class ApnsClientTest extends AbstractClientServerTest {
             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 =
+                final CompletableFuture<PushNotificationResponse<SimpleApnsPushNotification>> sendFuture =
                         client.sendNotification(pushNotification);
 
-                assertTrue(sendFuture.await(4, TimeUnit.SECONDS));
-                assertFalse(sendFuture.isSuccess());
+                // This should fail because there's no server running
+                assertThrows(ExecutionException.class, () -> sendFuture.get(4, TimeUnit.SECONDS));
             }
         } finally {
-            client.close().await();
+            client.close().get();
         }
     }
 
@@ -642,15 +631,14 @@ public class ApnsClientTest extends AbstractClientServerTest {
             final CountDownLatch countDownLatch = new CountDownLatch(notificationCount);
 
             for (int i = 0; i < notificationCount; i++) {
-                client.sendNotification(pushNotification).addListener(
-                        future -> countDownLatch.countDown());
+                client.sendNotification(pushNotification).whenComplete((response, cause) -> 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));
         } finally {
-            client.close().await();
+            client.close().get();
         }
     }
 
@@ -668,7 +656,7 @@ public class ApnsClientTest extends AbstractClientServerTest {
         final ApnsClient client = this.buildTokenAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             final SimpleApnsPushNotification pushNotification =
                     new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD, null, DeliveryPriority.IMMEDIATE, pushType);
@@ -683,8 +671,8 @@ public class ApnsClientTest extends AbstractClientServerTest {
 
             assertEquals(pushType, parsingServerHandler.acceptedNotifications.get(0).getPushType());
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
diff --git a/pushy/src/test/java/com/eatthepath/pushy/apns/ExampleApp.java b/pushy/src/test/java/com/eatthepath/pushy/apns/ExampleApp.java
index 8a6781f0..4c9af092 100644
--- a/pushy/src/test/java/com/eatthepath/pushy/apns/ExampleApp.java
+++ b/pushy/src/test/java/com/eatthepath/pushy/apns/ExampleApp.java
@@ -28,10 +28,10 @@ import com.eatthepath.pushy.apns.util.ApnsPayloadBuilder;
 import com.eatthepath.pushy.apns.util.SimpleApnsPushNotification;
 import com.eatthepath.pushy.apns.util.TokenUtil;
 import com.eatthepath.pushy.apns.util.concurrent.PushNotificationFuture;
-import io.netty.util.concurrent.Future;
 
 import java.io.File;
 import java.net.InetSocketAddress;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
 /**
@@ -122,27 +122,19 @@ public class ExampleApp {
         // To send push notifications efficiently, it's best to attach listeners
         // to "send push notification" futures so we don't have to wait for the
         // server to reply before we start sending the next notification.
-        sendNotificationFuture.addListener(future -> {
-            // When using a listener, callers should check for a failure to send a
-            // notification by checking whether the future itself was successful
-            // since an exception will not be thrown.
-            if (future.isSuccess()) {
-                final PushNotificationResponse<SimpleApnsPushNotification> pushNotificationResponse =
-                        sendNotificationFuture.getNow();
-
+        sendNotificationFuture.whenComplete((response, cause) -> {
+            if (response != null) {
                 // Handle the push notification response as before from here.
             } else {
                 // Something went wrong when trying to send the notification to the
-                // APNs gateway. We can find the exception that caused the failure
-                // by getting future.cause().
-                future.cause().printStackTrace();
+                // APNs gateway.
+                cause.printStackTrace();
             }
         });
 
         // Finally, when we're done sending notifications (i.e. when our
         // application is shutting down), we should close all APNs clients
         // that may be in play.
-        final Future<Void> closeFuture = apnsClient.close();
-        closeFuture.await();
+        final CompletableFuture<Void> closeFuture = apnsClient.close();
     }
 }
diff --git a/pushy/src/test/java/com/eatthepath/pushy/apns/server/MockApnsServerTest.java b/pushy/src/test/java/com/eatthepath/pushy/apns/server/MockApnsServerTest.java
index b9efe7f0..a2d132d1 100644
--- a/pushy/src/test/java/com/eatthepath/pushy/apns/server/MockApnsServerTest.java
+++ b/pushy/src/test/java/com/eatthepath/pushy/apns/server/MockApnsServerTest.java
@@ -25,7 +25,6 @@ package com.eatthepath.pushy.apns.server;
 import com.eatthepath.pushy.apns.*;
 import com.eatthepath.pushy.apns.util.SimpleApnsPushNotification;
 import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.util.concurrent.Future;
 import org.junit.jupiter.api.Test;
 
 import java.time.Instant;
@@ -93,15 +92,15 @@ public class MockApnsServerTest extends AbstractClientServerTest {
     void testStartAndShutdown() throws Exception {
         final MockApnsServer server = this.buildServer(new AcceptAllPushNotificationHandlerFactory());
 
-        assertTrue(server.start(PORT).await().isSuccess());
-        assertTrue(server.shutdown().await().isSuccess());
+        assertDoesNotThrow(() -> server.start(PORT).get());
+        assertDoesNotThrow(() -> server.shutdown().get());
     }
 
     @Test
     void testShutdownBeforeStart() throws Exception {
         final MockApnsServer server = this.buildServer(new AcceptAllPushNotificationHandlerFactory());
 
-        assertTrue(server.shutdown().await().isSuccess());
+        assertDoesNotThrow(() -> server.shutdown().get());
     }
 
     @Test
@@ -116,8 +115,8 @@ public class MockApnsServerTest extends AbstractClientServerTest {
                     .setEventLoopGroup(eventLoopGroup)
                     .build();
 
-            assertTrue(providedGroupServer.start(PORT).await().isSuccess());
-            assertTrue(providedGroupServer.shutdown().await().isSuccess());
+            assertDoesNotThrow(() -> providedGroupServer.start(PORT).get());
+            assertDoesNotThrow(() -> providedGroupServer.shutdown().get());
 
             assertFalse(eventLoopGroup.isShutdown());
         } finally {
@@ -146,11 +145,11 @@ public class MockApnsServerTest extends AbstractClientServerTest {
                     .setEventLoopGroup(eventLoopGroup)
                     .build();
 
-            assertTrue(providedGroupServer.start(PORT).await().isSuccess());
-            assertTrue(providedGroupServer.shutdown().await().isSuccess());
+            assertDoesNotThrow(() -> providedGroupServer.start(PORT).get());
+            assertDoesNotThrow(() -> providedGroupServer.shutdown().get());
 
-            assertTrue(providedGroupServer.start(PORT).await().isSuccess());
-            assertTrue(providedGroupServer.shutdown().await().isSuccess());
+            assertDoesNotThrow(() -> providedGroupServer.start(PORT).get());
+            assertDoesNotThrow(() -> providedGroupServer.shutdown().get());
         } finally {
             eventLoopGroup.shutdownGracefully().await();
         }
@@ -164,7 +163,7 @@ public class MockApnsServerTest extends AbstractClientServerTest {
         final ApnsClient client = this.buildTokenAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             final SimpleApnsPushNotification pushNotification = new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
 
@@ -179,8 +178,8 @@ public class MockApnsServerTest extends AbstractClientServerTest {
             assertEquals(pushNotification.getTopic(), listener.mostRecentPushNotification.getTopic());
             assertEquals(pushNotification.getPayload(), listener.mostRecentPushNotification.getPayload());
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -195,7 +194,7 @@ public class MockApnsServerTest extends AbstractClientServerTest {
         final ApnsClient client = this.buildTokenAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             final SimpleApnsPushNotification pushNotification = new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
 
@@ -209,8 +208,8 @@ public class MockApnsServerTest extends AbstractClientServerTest {
             assertEquals(RejectionReason.BAD_DEVICE_TOKEN, listener.mostRecentRejectionReason);
             assertNull(listener.mostRecentDeviceTokenExpiration);
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -226,7 +225,7 @@ public class MockApnsServerTest extends AbstractClientServerTest {
         final ApnsClient client = this.buildTokenAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             final SimpleApnsPushNotification pushNotification = new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
 
@@ -240,8 +239,8 @@ public class MockApnsServerTest extends AbstractClientServerTest {
             assertEquals(RejectionReason.UNREGISTERED, listener.mostRecentRejectionReason);
             assertEquals(expiration, listener.mostRecentDeviceTokenExpiration);
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -256,21 +255,21 @@ public class MockApnsServerTest extends AbstractClientServerTest {
         final ApnsClient client = this.buildTokenAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             final SimpleApnsPushNotification pushNotification = new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
 
-            final Future<PushNotificationResponse<SimpleApnsPushNotification>> sendFuture =
-                    client.sendNotification(pushNotification).await();
+            final PushNotificationResponse<SimpleApnsPushNotification> response =
+                    client.sendNotification(pushNotification).get();
 
-            assertTrue(sendFuture.isSuccess());
+            assertNotNull(response);
 
             listener.waitForNonZeroRejectedNotifications();
 
             assertEquals(RejectionReason.INTERNAL_SERVER_ERROR, listener.mostRecentRejectionReason);
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -280,7 +279,7 @@ public class MockApnsServerTest extends AbstractClientServerTest {
         final ApnsClient client = this.buildTokenAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             {
                 final SimpleApnsPushNotification pushNotification = new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
@@ -306,8 +305,8 @@ public class MockApnsServerTest extends AbstractClientServerTest {
                 assertEquals(pushNotificationWithApnsId.getApnsId(), response.getApnsId());
             }
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 
@@ -320,7 +319,7 @@ public class MockApnsServerTest extends AbstractClientServerTest {
         final ApnsClient client = this.buildTokenAuthenticationClient();
 
         try {
-            server.start(PORT).await();
+            server.start(PORT).get();
 
             {
                 final SimpleApnsPushNotification pushNotification = new SimpleApnsPushNotification(DEVICE_TOKEN, TOPIC, PAYLOAD);
@@ -346,8 +345,8 @@ public class MockApnsServerTest extends AbstractClientServerTest {
                 assertEquals(pushNotificationWithApnsId.getApnsId(), response.getApnsId());
             }
         } finally {
-            client.close().await();
-            server.shutdown().await();
+            client.close().get();
+            server.shutdown().get();
         }
     }
 }