refactor(messaging): use daemon threads and handle rejected executions

Author: lahirumaramba

src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java

@@ -38,6 +38,7 @@
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.firebase.ErrorCode;
 import com.google.firebase.FirebaseApp;
 import com.google.firebase.FirebaseException;
@@ -61,6 +62,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -111,11 +113,18 @@ private FirebaseMessagingClientImpl(Builder builder) {
   }
 
   private static ExecutorService createDefaultExecutor(ThreadFactory threadFactory) {
+    ThreadFactory baseFactory =
+        threadFactory != null ? threadFactory : Executors.defaultThreadFactory();
+    ThreadFactory factory = new ThreadFactoryBuilder()
+        .setThreadFactory(baseFactory)
+        .setNameFormat("firebase-messaging-topics-%d")
+        .setDaemon(true)
+        .build();
     ThreadPoolExecutor pool = new ThreadPoolExecutor(
         100, 100,
         60L, TimeUnit.SECONDS,
         new LinkedBlockingQueue<Runnable>(),
-        threadFactory != null ? threadFactory : Executors.defaultThreadFactory());
+        factory);
     pool.allowCoreThreadTimeOut(true);
     return pool;
   }
@@ -244,9 +253,14 @@ private TopicManagementResponse sendTopicManagementRequest(
     for (int i = 0; i < registrationTokens.size(); i++) {
       final int index = i;
       final String token = registrationTokens.get(i);
-      futures.add(CompletableFuture.supplyAsync(
-          () -> sendSingleTopicRequest(token, topicName, isSubscribe, index),
-          this.executor));
+      try {
+        futures.add(CompletableFuture.supplyAsync(
+            () -> sendSingleTopicRequest(token, topicName, isSubscribe, index),
+            this.executor));
+      } catch (RejectedExecutionException e) {
+        futures.add(CompletableFuture.completedFuture(
+            TopicResult.error(index, "REJECTED_BY_EXECUTOR")));
+      }
     }
 
     CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java

@@ -53,10 +53,15 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -776,4 +781,82 @@ public void testCustomThreadFactory() {
 
     assertSame(customThreadFactory, clientWithThreadFactory.getThreadFactory());
   }
+
+  @Test
+  public void testDefaultExecutorUsesDaemonThreads() throws Exception {
+    FirebaseMessagingClientImpl clientWithDefaultExecutor =
+        FirebaseMessagingClientImpl.builder()
+            .setProjectId("test-project")
+            .setJsonFactory(ApiClientUtils.getDefaultJsonFactory())
+            .setRequestFactory(new MockHttpTransport().createRequestFactory())
+            .setChildRequestFactory(ApiClientUtils.getDefaultTransport().createRequestFactory())
+            .build();
+
+    final AtomicBoolean isDaemon = new AtomicBoolean();
+    final AtomicReference<String> threadName = new AtomicReference<>();
+    final CountDownLatch latch = new CountDownLatch(1);
+    clientWithDefaultExecutor.getExecutor().execute(() -> {
+      Thread current = Thread.currentThread();
+      isDaemon.set(current.isDaemon());
+      threadName.set(current.getName());
+      latch.countDown();
+    });
+
+    assertTrue(latch.await(5, TimeUnit.SECONDS));
+    assertTrue(isDaemon.get());
+    assertNotNull(threadName.get());
+    assertTrue(threadName.get().startsWith("firebase-messaging-topics-"));
+  }
+
+  @Test
+  public void testTopicManagementRejectedExecution() throws Exception {
+    ExecutorService rejectingExecutor = new AbstractExecutorService() {
+      @Override
+      public void shutdown() {}
+
+      @Override
+      public List<Runnable> shutdownNow() {
+        return ImmutableList.of();
+      }
+
+      @Override
+      public boolean isShutdown() {
+        return false;
+      }
+
+      @Override
+      public boolean isTerminated() {
+        return false;
+      }
+
+      @Override
+      public boolean awaitTermination(long timeout, TimeUnit unit) {
+        return false;
+      }
+
+      @Override
+      public void execute(Runnable command) {
+        throw new RejectedExecutionException("Task rejected");
+      }
+    };
+
+    FirebaseMessagingClientImpl clientWithRejection = FirebaseMessagingClientImpl.builder()
+        .setProjectId("test-project")
+        .setJsonFactory(ApiClientUtils.getDefaultJsonFactory())
+        .setRequestFactory(new MockHttpTransport().createRequestFactory())
+        .setChildRequestFactory(ApiClientUtils.getDefaultTransport().createRequestFactory())
+        .setExecutor(rejectingExecutor)
+        .build();
+
+    TopicManagementResponse result = clientWithRejection.subscribeToTopic(
+        "test-topic", ImmutableList.of("id1", "id2"));
+
+    assertEquals(0, result.getSuccessCount());
+    assertEquals(2, result.getFailureCount());
+    assertEquals(2, result.getErrors().size());
+    assertEquals(0, result.getErrors().get(0).getIndex());
+    assertEquals("rejected-by-executor", result.getErrors().get(0).getReason());
+    assertEquals(1, result.getErrors().get(1).getIndex());
+    assertEquals("rejected-by-executor", result.getErrors().get(1).getReason());
+  }
 }