From 30c1989baa579eec56170f546ea31fe7156774dc Mon Sep 17 00:00:00 2001 From: Vitaly Takmazov Date: Thu, 20 Sep 2018 15:53:05 +0300 Subject: APNS: using non-blocking pushy API --- .../java/com/juick/components/Notifications.java | 72 ++++++++++++---------- 1 file changed, 41 insertions(+), 31 deletions(-) (limited to 'juick-notifications/src/main/java/com') 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> notification = apns.sendNotification( + new SimpleApnsPushNotification(token, topic, payload)); + notification.addListener((PushNotificationResponseListener) 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> notification = apns.sendNotification( + new SimpleApnsPushNotification(token, topic, payload)); + notification.addListener((PushNotificationResponseListener) future -> { + if (future.isSuccess()) { + processAPNSResponse(token, future.getNow()); + } else { + logger.warn("APNS error ", future.cause()); + } + }); + }); }); } - private void processAPNSResponse(String token, Future> notification) { - try { - final PushNotificationResponse 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 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 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 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) { -- cgit v1.2.3