APNS: using non-blocking pushy API

1 files changed, 41 insertions, 31 deletions

diff --git a/juick-notifications/src/main/java/com/juick/components/Notifications.java b/juick-notifications/src/main/java/com/juick/components/Notifications.java
index a42f23cb..ec6ffe9c 100644
--- a/juick-notifications/src/main/java/com/juick/components/Notifications.java
+++ b/juick-notifications/src/main/java/com/juick/components/Notifications.java
@@ -28,6 +28,9 @@ import com.turo.pushy.apns.ApnsClient;
 import com.turo.pushy.apns.PushNotificationResponse;
 import com.turo.pushy.apns.util.ApnsPayloadBuilder;
 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.util.concurrent.Future;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.text.StringEscapeUtils;
 import org.slf4j.Logger;
@@ -55,8 +58,6 @@ import javax.annotation.PreDestroy;
 import javax.inject.Inject;
 import java.io.IOException;
 import java.util.*;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
@@ -193,9 +194,17 @@ public class Notifications extends TextWebSocketHandler implements NotificationC
             apnsPayloadBuilder.setBadgeNumber(user.getUnreadCount());
             String payload = apnsPayloadBuilder.buildWithDefaultMaximumLength();
             user.getTokens().stream().filter(t -> t.getType().equals("apns"))
-                    .map(ExternalToken::getToken).forEach(token ->
-                    processAPNSResponse(token, apns.sendNotification(
-                            new SimpleApnsPushNotification(token, topic, payload))));
+                    .map(ExternalToken::getToken).forEach(token -> {
+                Future<PushNotificationResponse<SimpleApnsPushNotification>> notification = apns.sendNotification(
+                        new SimpleApnsPushNotification(token, topic, payload));
+                notification.addListener((PushNotificationResponseListener<SimpleApnsPushNotification>) future -> {
+                    if (future.isSuccess()) {
+                        processAPNSResponse(token, future.getNow());
+                    } else {
+                        logger.warn("APNS error ", future.cause());
+                    }
+                });
+            });
         });
     }
     private void serviceMessageReceived(@Nonnull com.juick.Message jmsg) {
@@ -210,37 +219,38 @@ public class Notifications extends TextWebSocketHandler implements NotificationC
             apnsPayloadBuilder.setBadgeNumber(user.getUnreadCount());
             String payload = apnsPayloadBuilder.buildWithDefaultMaximumLength();
             user.getTokens().stream().filter(t -> t.getType().equals("apns"))
-                    .map(ExternalToken::getToken).forEach(token ->
-                    processAPNSResponse(token, apns.sendNotification(
-                            new SimpleApnsPushNotification(token, topic, payload))));
+                    .map(ExternalToken::getToken).forEach(token -> {
+                Future<PushNotificationResponse<SimpleApnsPushNotification>> notification = apns.sendNotification(
+                        new SimpleApnsPushNotification(token, topic, payload));
+                notification.addListener((PushNotificationResponseListener<SimpleApnsPushNotification>) future -> {
+                    if (future.isSuccess()) {
+                        processAPNSResponse(token, future.getNow());
+                    } else {
+                        logger.warn("APNS error ", future.cause());
+                    }
+                });
+            });
         });
     }
 
-    private void processAPNSResponse(String token, Future<PushNotificationResponse<SimpleApnsPushNotification>> notification) {
-        try {
-            final PushNotificationResponse<SimpleApnsPushNotification> pushNotificationResponse
-                    = notification.get();
-            if (pushNotificationResponse.isAccepted()) {
-                logger.info("APNS accepted: {}", token);
-            } else {
-                String reason = pushNotificationResponse.getRejectionReason();
-                logger.info("APNS rejected: {}", reason);
-                if (reason.equals("BadDeviceToken")) {
-                    invalidAPNSTokens.add(token);
-                }
+    private void processAPNSResponse(String token, PushNotificationResponse<SimpleApnsPushNotification> pushNotificationResponse) {
+        if (pushNotificationResponse.isAccepted()) {
+            logger.info("APNS accepted: {}", token);
+        } else {
+            String reason = pushNotificationResponse.getRejectionReason();
+            logger.info("APNS rejected: {}", reason);
+            if (reason.equals("BadDeviceToken")) {
+                invalidAPNSTokens.add(token);
             }
-            Optional<Date> invalidationDate = Optional.ofNullable(
-                    pushNotificationResponse.getTokenInvalidationTimestamp());
-            invalidationDate.ifPresent(date -> {
-                if (date.before(new Date())) {
-                    logger.info("Token invalidated: {}", token);
-                    invalidAPNSTokens.add(token);
-                }
-            });
-
-        } catch (final ExecutionException | InterruptedException ex) {
-            logger.info("APNS exception", ex);
         }
+        Optional<Date> invalidationDate = Optional.ofNullable(
+                pushNotificationResponse.getTokenInvalidationTimestamp());
+        invalidationDate.ifPresent(date -> {
+            if (date.before(new Date())) {
+                logger.info("Token invalidated: {}", token);
+                invalidAPNSTokens.add(token);
+            }
+        });
     }
 
     public void addInvalidGCMToken(String token) {