From 96d3a69479818578c2ad1ba1942df58133a003bf Mon Sep 17 00:00:00 2001
From: Jon Chambers <jon@signal.org>
Date: Thu, 22 Oct 2020 18:10:09 -0400
Subject: [PATCH] Use container-managed executors for APN/GCM senders.

---
 .../textsecuregcm/WhisperServerService.java   |  6 +++--
 .../textsecuregcm/push/APNSender.java         |  7 +++---
 .../textsecuregcm/push/GCMSender.java         | 23 ++++---------------
 .../textsecuregcm/push/MessageSender.java     |  2 --
 .../tests/push/GCMSenderTest.java             |  6 ++---
 5 files changed, 15 insertions(+), 29 deletions(-)

diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java b/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java
index e54e4bb0..81132788 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java
@@ -299,6 +299,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
 
     ScheduledExecutorService recurringJobExecutor                 = environment.lifecycle().scheduledExecutorService(name(getClass(), "recurringJob-%d")).threads(2).build();
     ExecutorService          keyspaceNotificationDispatchExecutor = environment.lifecycle().executorService(name(getClass(), "keyspaceNotification-%d")).maxThreads(16).workQueue(keyspaceNotificationDispatchQueue).build();
+    ExecutorService          apnSenderExecutor                    = environment.lifecycle().executorService(name(getClass(), "apnSender-%d")).maxThreads(1).minThreads(1).build();
+    ExecutorService          gcmSenderExecutor                    = environment.lifecycle().executorService(name(getClass(), "gcmSender-%d")).maxThreads(1).minThreads(1).build();
 
     ClientPresenceManager      clientPresenceManager      = new ClientPresenceManager(messagesCacheCluster, recurringJobExecutor, keyspaceNotificationDispatchExecutor);
     DirectoryManager           directory                  = new DirectoryManager(directoryClient);
@@ -316,8 +318,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     DeadLetterHandler          deadLetterHandler          = new DeadLetterHandler(accountsManager, messagesManager);
     DispatchManager            dispatchManager            = new DispatchManager(pubSubClientFactory, Optional.of(deadLetterHandler));
     PubSubManager              pubSubManager              = new PubSubManager(pubsubClient, dispatchManager);
-    APNSender                  apnSender                  = new APNSender(accountsManager, config.getApnConfiguration());
-    GCMSender                  gcmSender                  = new GCMSender(accountsManager, config.getGcmConfiguration().getApiKey());
+    APNSender                  apnSender                  = new APNSender(apnSenderExecutor, accountsManager, config.getApnConfiguration());
+    GCMSender                  gcmSender                  = new GCMSender(gcmSenderExecutor, accountsManager, config.getGcmConfiguration().getApiKey());
     RateLimiters               rateLimiters               = new RateLimiters(config.getLimitsConfiguration(), cacheCluster);
     ProvisioningManager        provisioningManager        = new ProvisioningManager(pubSubManager);
 
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/push/APNSender.java b/service/src/main/java/org/whispersystems/textsecuregcm/push/APNSender.java
index 7ca01605..1886d584 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/push/APNSender.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/push/APNSender.java
@@ -54,17 +54,18 @@ public class APNSender implements Managed {
   private static final Meter unregisteredEventStale  = metricRegistry.meter(name(APNSender.class, "unregistered_event_stale"));
   private static final Meter unregisteredEventFresh  = metricRegistry.meter(name(APNSender.class, "unregistered_event_fresh"));
 
-  private ExecutorService    executor;
   private ApnFallbackManager fallbackManager;
 
+  private final ExecutorService    executor;
   private final AccountsManager    accountsManager;
   private final String             bundleId;
   private final boolean            sandbox;
   private final RetryingApnsClient apnsClient;
 
-  public APNSender(AccountsManager accountsManager, ApnConfiguration configuration)
+  public APNSender(ExecutorService executor, AccountsManager accountsManager, ApnConfiguration configuration)
       throws IOException, NoSuchAlgorithmException, InvalidKeyException
   {
+    this.executor        = executor;
     this.accountsManager = accountsManager;
     this.bundleId        = configuration.getBundleId();
     this.sandbox         = configuration.isSandboxEnabled();
@@ -120,12 +121,10 @@ public class APNSender implements Managed {
 
   @Override
   public void start() {
-    this.executor = Executors.newSingleThreadExecutor();
   }
 
   @Override
   public void stop() {
-    this.executor.shutdown();
     this.apnsClient.disconnect();
   }
 
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/push/GCMSender.java b/service/src/main/java/org/whispersystems/textsecuregcm/push/GCMSender.java
index 94fa723f..2ac2bc8c 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/push/GCMSender.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/push/GCMSender.java
@@ -22,13 +22,11 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
 import static com.codahale.metrics.MetricRegistry.name;
-import io.dropwizard.lifecycle.Managed;
 
-public class GCMSender implements Managed {
+public class GCMSender {
 
   private final Logger logger = LoggerFactory.getLogger(GCMSender.class);
 
@@ -46,17 +44,16 @@ public class GCMSender implements Managed {
 
   private final AccountsManager   accountsManager;
   private final Sender            signalSender;
-  private       ExecutorService   executor;
+  private final ExecutorService   executor;
 
-  public GCMSender(AccountsManager accountsManager, String signalKey) {
-    this.accountsManager = accountsManager;
-    this.signalSender    = new Sender(signalKey, SystemMapper.getMapper(), 6);
+  public GCMSender(ExecutorService executor, AccountsManager accountsManager, String signalKey) {
+    this(executor, accountsManager, new Sender(signalKey, SystemMapper.getMapper(), 6));
 
     CircuitBreakerUtil.registerMetrics(metricRegistry, signalSender.getRetry(), Sender.class);
   }
 
   @VisibleForTesting
-  public GCMSender(AccountsManager accountsManager, Sender sender, ExecutorService executor) {
+  public GCMSender(ExecutorService executor, AccountsManager accountsManager, Sender sender) {
     this.accountsManager = accountsManager;
     this.signalSender    = sender;
     this.executor        = executor;
@@ -99,16 +96,6 @@ public class GCMSender implements Managed {
     });
   }
 
-  @Override
-  public void start() {
-    executor = Executors.newSingleThreadExecutor();
-  }
-
-  @Override
-  public void stop() {
-    this.executor.shutdown();
-  }
-
   private void handleBadRegistration(GcmMessage message) {
     Optional<Account> account = getAccountForEvent(message);
 
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/push/MessageSender.java b/service/src/main/java/org/whispersystems/textsecuregcm/push/MessageSender.java
index c138ac5e..72f892f0 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/push/MessageSender.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/push/MessageSender.java
@@ -191,7 +191,6 @@ public class MessageSender implements Managed {
   @Override
   public void start() {
     apnSender.start();
-    gcmSender.start();
   }
 
   @Override
@@ -200,6 +199,5 @@ public class MessageSender implements Managed {
     executor.awaitTermination(5, TimeUnit.MINUTES);
 
     apnSender.stop();
-    gcmSender.stop();
   }
 }
diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/tests/push/GCMSenderTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/tests/push/GCMSenderTest.java
index ae137587..a81c5b6a 100644
--- a/service/src/test/java/org/whispersystems/textsecuregcm/tests/push/GCMSenderTest.java
+++ b/service/src/test/java/org/whispersystems/textsecuregcm/tests/push/GCMSenderTest.java
@@ -33,7 +33,7 @@ public class GCMSenderTest {
     when(successResult.isSuccess()).thenReturn(true);
 
     GcmMessage message = new GcmMessage("foo", "+12223334444", 1, GcmMessage.Type.NOTIFICATION, Optional.empty());
-    GCMSender gcmSender = new GCMSender(accountsManager, sender, executorService);
+    GCMSender gcmSender = new GCMSender(executorService, accountsManager, sender);
 
     CompletableFuture<Result> successFuture = CompletableFuture.completedFuture(successResult);
 
@@ -67,7 +67,7 @@ public class GCMSenderTest {
     when(invalidResult.isSuccess()).thenReturn(true);
 
     GcmMessage message = new GcmMessage(gcmId, destinationNumber, 1, GcmMessage.Type.NOTIFICATION, Optional.empty());
-    GCMSender gcmSender = new GCMSender(accountsManager, sender, executorService);
+    GCMSender gcmSender = new GCMSender(executorService, accountsManager, sender);
 
     CompletableFuture<Result> invalidFuture = CompletableFuture.completedFuture(invalidResult);
 
@@ -106,7 +106,7 @@ public class GCMSenderTest {
     when(canonicalResult.getCanonicalRegistrationId()).thenReturn(canonicalId);
 
     GcmMessage message = new GcmMessage(gcmId, destinationNumber, 1, GcmMessage.Type.NOTIFICATION, Optional.empty());
-    GCMSender gcmSender = new GCMSender(accountsManager, sender, executorService);
+    GCMSender gcmSender = new GCMSender(executorService, accountsManager, sender);
 
     CompletableFuture<Result> invalidFuture = CompletableFuture.completedFuture(canonicalResult);