From 51005bd0724a30d6975be7f6881d8faba906cb73 Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Tue, 15 Sep 2026 18:20:13 -0400 Subject: [PATCH 1/3] feat(messaging): topic subscription changes --- .../firebase/messaging/FirebaseMessaging.java | 73 ++++++ .../messaging/FirebaseMessagingClient.java | 21 ++ .../FirebaseMessagingClientImpl.java | 220 +++++++++++++++++- .../messaging/TopicManagementResponse.java | 7 +- .../FirebaseMessagingClientImplTest.java | 90 +++++++ .../messaging/FirebaseMessagingTest.java | 125 +++++++++- 6 files changed, 521 insertions(+), 15 deletions(-) diff --git a/src/main/java/com/google/firebase/messaging/FirebaseMessaging.java b/src/main/java/com/google/firebase/messaging/FirebaseMessaging.java index 0e9831588..cc04f65f2 100644 --- a/src/main/java/com/google/firebase/messaging/FirebaseMessaging.java +++ b/src/main/java/com/google/firebase/messaging/FirebaseMessaging.java @@ -557,6 +557,42 @@ private CallableOperation s final List registrationTokens, final String topic) { checkRegistrationTokens(registrationTokens); checkTopic(topic); + final FirebaseMessagingClient messagingClient = getMessagingClient(); + return new CallableOperation() { + @Override + protected TopicManagementResponse execute() throws FirebaseMessagingException { + return messagingClient.subscribeToTopic(topic, registrationTokens); + } + }; + } + + /** + * Subscribes a list of registration tokens to a topic using the legacy Instance ID API. + * + * @deprecated Use {@link #subscribeToTopic(List, String)} instead. + */ + @Deprecated + public TopicManagementResponse subscribeToTopicLegacy(@NonNull List registrationTokens, + @NonNull String topic) throws FirebaseMessagingException { + return subscribeLegacyOp(registrationTokens, topic).call(); + } + + /** + * Similar to {@link #subscribeToTopicLegacy(List, String)} but performs the operation + * asynchronously. + * + * @deprecated Use {@link #subscribeToTopicAsync(List, String)} instead. + */ + @Deprecated + public ApiFuture subscribeToTopicLegacyAsync( + @NonNull List registrationTokens, @NonNull String topic) { + return subscribeLegacyOp(registrationTokens, topic).callAsync(app); + } + + private CallableOperation subscribeLegacyOp( + final List registrationTokens, final String topic) { + checkRegistrationTokens(registrationTokens); + checkTopic(topic); final InstanceIdClient instanceIdClient = getInstanceIdClient(); return new CallableOperation() { @Override @@ -597,6 +633,43 @@ private CallableOperation u final List registrationTokens, final String topic) { checkRegistrationTokens(registrationTokens); checkTopic(topic); + final FirebaseMessagingClient messagingClient = getMessagingClient(); + return new CallableOperation() { + @Override + protected TopicManagementResponse execute() throws FirebaseMessagingException { + return messagingClient.unsubscribeFromTopic(topic, registrationTokens); + } + }; + } + + /** + * Unsubscribes a list of registration tokens from a topic using the legacy Instance ID API. + * + * @deprecated Use {@link #unsubscribeFromTopic(List, String)} instead. + */ + @Deprecated + public TopicManagementResponse unsubscribeFromTopicLegacy( + @NonNull List registrationTokens, + @NonNull String topic) throws FirebaseMessagingException { + return unsubscribeLegacyOp(registrationTokens, topic).call(); + } + + /** + * Similar to {@link #unsubscribeFromTopicLegacy(List, String)} but performs the operation + * asynchronously. + * + * @deprecated Use {@link #unsubscribeFromTopicAsync(List, String)} instead. + */ + @Deprecated + public ApiFuture unsubscribeFromTopicLegacyAsync( + @NonNull List registrationTokens, @NonNull String topic) { + return unsubscribeLegacyOp(registrationTokens, topic).callAsync(app); + } + + private CallableOperation + unsubscribeLegacyOp(final List registrationTokens, final String topic) { + checkRegistrationTokens(registrationTokens); + checkTopic(topic); final InstanceIdClient instanceIdClient = getInstanceIdClient(); return new CallableOperation() { @Override diff --git a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClient.java b/src/main/java/com/google/firebase/messaging/FirebaseMessagingClient.java index da049565d..d24a57ce0 100644 --- a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClient.java +++ b/src/main/java/com/google/firebase/messaging/FirebaseMessagingClient.java @@ -29,4 +29,25 @@ interface FirebaseMessagingClient { */ BatchResponse sendAll(List messages, boolean dryRun) throws FirebaseMessagingException; + /** + * Subscribes a list of registration tokens to a topic via the FCM v1 API. + * + * @param topic Name of the topic. + * @param registrationTokens A list of registration tokens. + * @return A {@link TopicManagementResponse}. + * @throws FirebaseMessagingException If an error occurs. + */ + TopicManagementResponse subscribeToTopic( + String topic, List registrationTokens) throws FirebaseMessagingException; + + /** + * Unsubscribes a list of registration tokens from a topic via the FCM v1 API. + * + * @param topic Name of the topic. + * @param registrationTokens A list of registration tokens. + * @return A {@link TopicManagementResponse}. + * @throws FirebaseMessagingException If an error occurs. + */ + TopicManagementResponse unsubscribeFromTopic( + String topic, List registrationTokens) throws FirebaseMessagingException; } diff --git a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java b/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java index 6049b4f4d..d59a2b60b 100644 --- a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java +++ b/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java @@ -52,21 +52,31 @@ import com.google.firebase.messaging.internal.MessagingServiceErrorResponse; import com.google.firebase.messaging.internal.MessagingServiceResponse; import java.io.IOException; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; /** * A helper class for interacting with Firebase Cloud Messaging service. */ final class FirebaseMessagingClientImpl implements FirebaseMessagingClient { - private static final String FCM_URL = "https://fcm.googleapis.com/v1/projects/%s/messages:send"; + private static final String DEFAULT_FCM_HOST = "https://fcm.googleapis.com"; + private static final String FCM_URL = "%s/v1/projects/%s/messages:send"; private static final Map COMMON_HEADERS = ImmutableMap.of( "X-GOOG-API-FORMAT-VERSION", "2", "X-Firebase-Client", "fire-admin-java/" + SdkUtils.getVersion()); + private final String projectId; + private final String fcmHost; private final String fcmSendUrl; private final HttpRequestFactory requestFactory; private final HttpRequestFactory childRequestFactory; @@ -75,10 +85,14 @@ final class FirebaseMessagingClientImpl implements FirebaseMessagingClient { private final MessagingErrorHandler errorHandler; private final ErrorHandlingHttpClient httpClient; private final MessagingBatchClient batchClient; + private final ExecutorService executor; + private final ThreadFactory threadFactory; private FirebaseMessagingClientImpl(Builder builder) { checkArgument(!Strings.isNullOrEmpty(builder.projectId)); - this.fcmSendUrl = String.format(FCM_URL, builder.projectId); + this.projectId = builder.projectId; + this.fcmHost = Strings.isNullOrEmpty(builder.fcmHost) ? DEFAULT_FCM_HOST : builder.fcmHost; + this.fcmSendUrl = String.format(FCM_URL, this.fcmHost, builder.projectId); this.requestFactory = checkNotNull(builder.requestFactory); this.childRequestFactory = checkNotNull(builder.childRequestFactory); this.jsonFactory = checkNotNull(builder.jsonFactory); @@ -87,6 +101,8 @@ private FirebaseMessagingClientImpl(Builder builder) { this.httpClient = new ErrorHandlingHttpClient<>(requestFactory, jsonFactory, errorHandler) .setInterceptor(responseInterceptor); this.batchClient = new MessagingBatchClient(requestFactory.getTransport(), jsonFactory); + this.executor = builder.executor; + this.threadFactory = builder.threadFactory; } @VisibleForTesting @@ -182,17 +198,199 @@ public void initialize(HttpRequest request) throws IOException { }; } + @Override + public TopicManagementResponse subscribeToTopic( + String topic, List registrationTokens) throws FirebaseMessagingException { + return sendTopicManagementRequest(topic, registrationTokens, true); + } + + @Override + public TopicManagementResponse unsubscribeFromTopic( + String topic, List registrationTokens) throws FirebaseMessagingException { + return sendTopicManagementRequest(topic, registrationTokens, false); + } + + private TopicManagementResponse sendTopicManagementRequest( + String topic, List registrationTokens, boolean isSubscribe) { + String topicName = topic.startsWith("/topics/") ? topic.substring("/topics/".length()) : topic; + + ExecutorService pool = this.executor != null + ? this.executor + : (this.threadFactory != null + ? Executors.newFixedThreadPool( + Math.min(registrationTokens.size(), 100), this.threadFactory) + : Executors.newFixedThreadPool(Math.min(registrationTokens.size(), 100))); + boolean shouldShutdown = (this.executor == null); + + try { + List> futures = new ArrayList<>(); + 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), pool)); + } + + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); + + int successCount = 0; + List errors = new ArrayList<>(); + for (CompletableFuture future : futures) { + TopicResult result = future.join(); + if (result.isSuccess()) { + successCount++; + } else { + errors.add(new TopicManagementResponse.Error(result.getIndex(), result.getReason())); + } + } + return new TopicManagementResponse(successCount, errors); + } finally { + if (shouldShutdown) { + pool.shutdown(); + } + } + } + + private TopicResult sendSingleTopicRequest( + String token, String topicName, boolean isSubscribe, int index) { + try { + String encodedToken = URLEncoder.encode(token, StandardCharsets.UTF_8.name()); + String encodedTopic = URLEncoder.encode(topicName, StandardCharsets.UTF_8.name()); + HttpRequestInfo requestInfo; + if (isSubscribe) { + String url = String.format( + "%s/v1/projects/%s/registrations/%s/topicSubscriptions?topic_name=%s", + fcmHost, projectId, encodedToken, encodedTopic); + requestInfo = HttpRequestInfo.buildJsonPostRequest(url, ImmutableMap.of()) + .addAllHeaders(COMMON_HEADERS); + } else { + String url = String.format( + "%s/v1/projects/%s/registrations/%s/topicSubscriptions/%s?allow_missing=true", + fcmHost, projectId, encodedToken, encodedTopic); + requestInfo = HttpRequestInfo.buildDeleteRequest(url) + .addAllHeaders(COMMON_HEADERS); + } + + httpClient.send(requestInfo); + return TopicResult.success(index); + } catch (FirebaseMessagingException e) { + if (isSubscribe && isAlreadyExists(e)) { + return TopicResult.success(index); + } + String reason = extractReason(e); + return TopicResult.error(index, reason); + } catch (Exception e) { + return TopicResult.error(index, "UNKNOWN_ERROR"); + } + } + + private boolean isAlreadyExists(FirebaseMessagingException e) { + if (e.getHttpResponse() != null && e.getHttpResponse().getStatusCode() == 409) { + return true; + } + if (e.getErrorCode() == ErrorCode.ALREADY_EXISTS || e.getErrorCode() == ErrorCode.CONFLICT) { + return true; + } + return false; + } + + private String extractReason(FirebaseMessagingException e) { + if (e.getMessagingErrorCode() != null) { + return e.getMessagingErrorCode().name(); + } + if (e.getHttpResponse() != null && !Strings.isNullOrEmpty(e.getHttpResponse().getContent())) { + try { + MessagingServiceErrorResponse parsed = jsonFactory.createJsonParser( + e.getHttpResponse().getContent()) + .parseAndClose(MessagingServiceErrorResponse.class); + if (parsed.getMessagingErrorCode() != null) { + return parsed.getMessagingErrorCode().name(); + } + if (!Strings.isNullOrEmpty(parsed.getStatus())) { + return parsed.getStatus(); + } + if (!Strings.isNullOrEmpty(parsed.getErrorMessage())) { + return parsed.getErrorMessage(); + } + } catch (Exception ignore) { + // Ignore JSON parsing errors + } + } + if (e.getErrorCode() != null && e.getErrorCode() != ErrorCode.UNKNOWN) { + return e.getErrorCode().name(); + } + if (e.getHttpResponse() != null) { + switch (e.getHttpResponse().getStatusCode()) { + case 400: + return "INVALID_ARGUMENT"; + case 401: + case 403: + return "PERMISSION_DENIED"; + case 404: + return "NOT_FOUND"; + case 429: + return "RESOURCE_EXHAUSTED"; + case 500: + return "INTERNAL"; + case 503: + return "DEADLINE_EXCEEDED"; + default: + return "UNKNOWN_ERROR"; + } + } + return "UNKNOWN_ERROR"; + } + + private static class TopicResult { + private final int index; + private final boolean success; + private final String reason; + + private TopicResult(int index, boolean success, String reason) { + this.index = index; + this.success = success; + this.reason = reason; + } + + static TopicResult success(int index) { + return new TopicResult(index, true, null); + } + + static TopicResult error(int index, String reason) { + return new TopicResult(index, false, reason); + } + + int getIndex() { + return index; + } + + boolean isSuccess() { + return success; + } + + String getReason() { + return reason; + } + } + static FirebaseMessagingClientImpl fromApp(FirebaseApp app) { String projectId = ImplFirebaseTrampolines.getProjectId(app); checkArgument(!Strings.isNullOrEmpty(projectId), "Project ID is required to access messaging service. Use a service account credential or " + "set the project ID explicitly via FirebaseOptions. Alternatively you can also " + "set the project ID via the GOOGLE_CLOUD_PROJECT environment variable."); + ThreadFactory threadFactory = null; + try { + threadFactory = ImplFirebaseTrampolines.getThreadFactory(app); + } catch (Exception ignored) { + // Ignored + } return FirebaseMessagingClientImpl.builder() .setProjectId(projectId) .setRequestFactory(ApiClientUtils.newAuthorizedRequestFactory(app)) .setChildRequestFactory(ApiClientUtils.newUnauthorizedRequestFactory(app)) .setJsonFactory(app.getOptions().getJsonFactory()) + .setThreadFactory(threadFactory) .build(); } @@ -203,10 +401,13 @@ static Builder builder() { static final class Builder { private String projectId; + private String fcmHost = DEFAULT_FCM_HOST; private HttpRequestFactory requestFactory; private HttpRequestFactory childRequestFactory; private JsonFactory jsonFactory; private HttpResponseInterceptor responseInterceptor; + private ExecutorService executor; + private ThreadFactory threadFactory; private Builder() { } @@ -215,6 +416,11 @@ Builder setProjectId(String projectId) { return this; } + Builder setFcmHost(String fcmHost) { + this.fcmHost = fcmHost; + return this; + } + Builder setRequestFactory(HttpRequestFactory requestFactory) { this.requestFactory = requestFactory; return this; @@ -235,6 +441,16 @@ Builder setResponseInterceptor(HttpResponseInterceptor responseInterceptor) { return this; } + Builder setExecutor(ExecutorService executor) { + this.executor = executor; + return this; + } + + Builder setThreadFactory(ThreadFactory threadFactory) { + this.threadFactory = threadFactory; + return this; + } + FirebaseMessagingClientImpl build() { return new FirebaseMessagingClientImpl(this); } diff --git a/src/main/java/com/google/firebase/messaging/TopicManagementResponse.java b/src/main/java/com/google/firebase/messaging/TopicManagementResponse.java index f02590f74..28664efce 100644 --- a/src/main/java/com/google/firebase/messaging/TopicManagementResponse.java +++ b/src/main/java/com/google/firebase/messaging/TopicManagementResponse.java @@ -61,6 +61,11 @@ public class TopicManagementResponse { this.errors = errors.build(); } + TopicManagementResponse(int successCount, List errors) { + this.successCount = successCount; + this.errors = ImmutableList.copyOf(errors); + } + /** * Gets the number of registration tokens that were successfully subscribed or unsubscribed. * @@ -97,7 +102,7 @@ public static class Error { private final int index; private final String reason; - private Error(int index, String reason) { + Error(int index, String reason) { this.index = index; if (reason == null || reason.trim().isEmpty()) { this.reason = UNKNOWN_ERROR; diff --git a/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java b/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java index 03bfc4327..17b255e7c 100644 --- a/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java +++ b/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java @@ -651,4 +651,94 @@ private static Map> buildTestMessages() { return builder.build(); } + + @Test + public void testSubscribeToTopic() throws Exception { + response.setContent("{}"); + TopicManagementResponse result = client.subscribeToTopic( + "test-topic", ImmutableList.of("id1")); + + assertEquals(1, result.getSuccessCount()); + assertEquals(0, result.getFailureCount()); + assertEquals(0, result.getErrors().size()); + HttpRequest request = interceptor.getLastRequest(); + assertEquals("POST", request.getRequestMethod()); + assertEquals( + "https://fcm.googleapis.com/v1/projects/test-project/registrations/id1/topicSubscriptions?topic_name=test-topic", + request.getUrl().toString()); + HttpHeaders headers = request.getHeaders(); + assertEquals("2", headers.get("X-GOOG-API-FORMAT-VERSION")); + assertEquals("fire-admin-java/" + SdkUtils.getVersion(), headers.get("X-Firebase-Client")); + } + + @Test + public void testSubscribeToTopic409() throws Exception { + response.setStatusCode(409).setContent("{\"error\": {\"status\": \"ALREADY_EXISTS\"}}"); + TopicManagementResponse result = client.subscribeToTopic( + "test-topic", ImmutableList.of("id1")); + + assertEquals(1, result.getSuccessCount()); + assertEquals(0, result.getFailureCount()); + } + + @Test + public void testUnsubscribeFromTopic() throws Exception { + response.setContent("{}"); + TopicManagementResponse result = client.unsubscribeFromTopic( + "test-topic", ImmutableList.of("id1")); + + assertEquals(1, result.getSuccessCount()); + assertEquals(0, result.getFailureCount()); + assertEquals(0, result.getErrors().size()); + HttpRequest request = interceptor.getLastRequest(); + assertEquals("DELETE", request.getRequestMethod()); + assertEquals( + "https://fcm.googleapis.com/v1/projects/test-project/registrations/id1/topicSubscriptions/test-topic?allow_missing=true", + request.getUrl().toString()); + } + + @Test + public void testUnsubscribeFromTopic404() throws Exception { + response.setStatusCode(404).setContent("{\"error\": {\"status\": \"NOT_FOUND\"}}"); + TopicManagementResponse result = client.unsubscribeFromTopic( + "test-topic", ImmutableList.of("id1")); + + assertEquals(0, result.getSuccessCount()); + assertEquals(1, result.getFailureCount()); + assertEquals(1, result.getErrors().size()); + assertEquals(0, result.getErrors().get(0).getIndex()); + assertEquals("registration-token-not-registered", result.getErrors().get(0).getReason()); + } + + @Test + public void testTopicManagementFcmErrorDetails() throws Exception { + response.setStatusCode(404).setContent("{\n" + + " \"error\": {\n" + + " \"status\": \"NOT_FOUND\",\n" + + " \"details\": [\n" + + " {\n" + + " \"@type\": \"type.googleapis.com/google.firebase.fcm.v1.FcmError\",\n" + + " \"errorCode\": \"UNREGISTERED\"\n" + + " }\n" + + " ]\n" + + " }\n" + + "}"); + TopicManagementResponse result = client.subscribeToTopic( + "test-topic", ImmutableList.of("id1")); + + assertEquals(0, result.getSuccessCount()); + assertEquals(1, result.getFailureCount()); + assertEquals("unregistered", result.getErrors().get(0).getReason()); + } + + @Test + public void testTopicManagement500Error() throws Exception { + response.setStatusCode(500).setContent("{}"); + TopicManagementResponse result = client.subscribeToTopic( + "test-topic", ImmutableList.of("id1")); + + assertEquals(0, result.getSuccessCount()); + assertEquals(1, result.getFailureCount()); + assertEquals("internal-error", result.getErrors().get(0).getReason()); + } } diff --git a/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java b/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java index 42a499b75..6444dbc99 100644 --- a/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java +++ b/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java @@ -547,9 +547,9 @@ public void testSendEachForMulticastAsyncFailure() throws Exception { @Test public void testInvalidSubscribe() throws FirebaseMessagingException { - MockInstanceIdClient client = MockInstanceIdClient.fromResponse(null); + MockFirebaseMessagingClient client = MockFirebaseMessagingClient.fromResponse(null); FirebaseMessaging messaging = getMessagingForTopicManagement( - Suppliers.ofInstance(client)); + Suppliers.ofInstance(client)); for (TopicMgtArgs args : INVALID_TOPIC_MGT_ARGS) { try { @@ -565,18 +565,21 @@ public void testInvalidSubscribe() throws FirebaseMessagingException { @Test public void testSubscribeToTopic() throws FirebaseMessagingException { - MockInstanceIdClient client = MockInstanceIdClient.fromResponse(TOPIC_MGT_RESPONSE); + MockFirebaseMessagingClient client = + MockFirebaseMessagingClient.fromResponse(TOPIC_MGT_RESPONSE); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); TopicManagementResponse got = messaging.subscribeToTopic( ImmutableList.of("id1", "id2"), "test-topic"); assertSame(TOPIC_MGT_RESPONSE, got); + assertEquals("test-topic", client.lastTopic); + assertEquals(ImmutableList.of("id1", "id2"), client.lastBatch); } @Test public void testSubscribeToTopicFailure() { - MockInstanceIdClient client = MockInstanceIdClient.fromException(TEST_EXCEPTION); + MockFirebaseMessagingClient client = MockFirebaseMessagingClient.fromException(TEST_EXCEPTION); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); try { @@ -588,7 +591,8 @@ public void testSubscribeToTopicFailure() { @Test public void testSubscribeToTopicAsync() throws Exception { - MockInstanceIdClient client = MockInstanceIdClient.fromResponse(TOPIC_MGT_RESPONSE); + MockFirebaseMessagingClient client = + MockFirebaseMessagingClient.fromResponse(TOPIC_MGT_RESPONSE); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); TopicManagementResponse got = messaging.subscribeToTopicAsync( @@ -599,7 +603,7 @@ public void testSubscribeToTopicAsync() throws Exception { @Test public void testSubscribeToTopicAsyncFailure() throws InterruptedException { - MockInstanceIdClient client = MockInstanceIdClient.fromException(TEST_EXCEPTION); + MockFirebaseMessagingClient client = MockFirebaseMessagingClient.fromException(TEST_EXCEPTION); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); try { @@ -609,11 +613,35 @@ public void testSubscribeToTopicAsyncFailure() throws InterruptedException { } } + @Test + public void testSubscribeToTopicLegacy() throws FirebaseMessagingException { + MockInstanceIdClient client = MockInstanceIdClient.fromResponse(TOPIC_MGT_RESPONSE); + FirebaseMessaging messaging = + getMessagingForLegacyTopicManagement(Suppliers.ofInstance(client)); + + TopicManagementResponse got = messaging.subscribeToTopicLegacy( + ImmutableList.of("id1", "id2"), "test-topic"); + + assertSame(TOPIC_MGT_RESPONSE, got); + } + + @Test + public void testSubscribeToTopicLegacyAsync() throws Exception { + MockInstanceIdClient client = MockInstanceIdClient.fromResponse(TOPIC_MGT_RESPONSE); + FirebaseMessaging messaging = + getMessagingForLegacyTopicManagement(Suppliers.ofInstance(client)); + + TopicManagementResponse got = messaging.subscribeToTopicLegacyAsync( + ImmutableList.of("id1", "id2"), "test-topic").get(); + + assertSame(TOPIC_MGT_RESPONSE, got); + } + @Test public void testInvalidUnsubscribe() throws FirebaseMessagingException { - MockInstanceIdClient client = MockInstanceIdClient.fromResponse(null); + MockFirebaseMessagingClient client = MockFirebaseMessagingClient.fromResponse(null); FirebaseMessaging messaging = getMessagingForTopicManagement( - Suppliers.ofInstance(client)); + Suppliers.ofInstance(client)); for (TopicMgtArgs args : INVALID_TOPIC_MGT_ARGS) { try { @@ -629,18 +657,21 @@ public void testInvalidUnsubscribe() throws FirebaseMessagingException { @Test public void testUnsubscribeFromTopic() throws FirebaseMessagingException { - MockInstanceIdClient client = MockInstanceIdClient.fromResponse(TOPIC_MGT_RESPONSE); + MockFirebaseMessagingClient client = + MockFirebaseMessagingClient.fromResponse(TOPIC_MGT_RESPONSE); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); TopicManagementResponse got = messaging.unsubscribeFromTopic( ImmutableList.of("id1", "id2"), "test-topic"); assertSame(TOPIC_MGT_RESPONSE, got); + assertEquals("test-topic", client.lastTopic); + assertEquals(ImmutableList.of("id1", "id2"), client.lastBatch); } @Test public void testUnsubscribeFromTopicFailure() { - MockInstanceIdClient client = MockInstanceIdClient.fromException(TEST_EXCEPTION); + MockFirebaseMessagingClient client = MockFirebaseMessagingClient.fromException(TEST_EXCEPTION); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); try { @@ -652,7 +683,8 @@ public void testUnsubscribeFromTopicFailure() { @Test public void testUnsubscribeFromTopicAsync() throws Exception { - MockInstanceIdClient client = MockInstanceIdClient.fromResponse(TOPIC_MGT_RESPONSE); + MockFirebaseMessagingClient client = + MockFirebaseMessagingClient.fromResponse(TOPIC_MGT_RESPONSE); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); TopicManagementResponse got = messaging.unsubscribeFromTopicAsync( @@ -663,7 +695,7 @@ public void testUnsubscribeFromTopicAsync() throws Exception { @Test public void testUnsubscribeFromTopicAsyncFailure() throws InterruptedException { - MockInstanceIdClient client = MockInstanceIdClient.fromException(TEST_EXCEPTION); + MockFirebaseMessagingClient client = MockFirebaseMessagingClient.fromException(TEST_EXCEPTION); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); try { @@ -673,6 +705,30 @@ public void testUnsubscribeFromTopicAsyncFailure() throws InterruptedException { } } + @Test + public void testUnsubscribeFromTopicLegacy() throws FirebaseMessagingException { + MockInstanceIdClient client = MockInstanceIdClient.fromResponse(TOPIC_MGT_RESPONSE); + FirebaseMessaging messaging = + getMessagingForLegacyTopicManagement(Suppliers.ofInstance(client)); + + TopicManagementResponse got = messaging.unsubscribeFromTopicLegacy( + ImmutableList.of("id1", "id2"), "test-topic"); + + assertSame(TOPIC_MGT_RESPONSE, got); + } + + @Test + public void testUnsubscribeFromTopicLegacyAsync() throws Exception { + MockInstanceIdClient client = MockInstanceIdClient.fromResponse(TOPIC_MGT_RESPONSE); + FirebaseMessaging messaging = + getMessagingForLegacyTopicManagement(Suppliers.ofInstance(client)); + + TopicManagementResponse got = messaging.unsubscribeFromTopicLegacyAsync( + ImmutableList.of("id1", "id2"), "test-topic").get(); + + assertSame(TOPIC_MGT_RESPONSE, got); + } + private FirebaseMessaging getMessagingForSend( Supplier supplier) { FirebaseApp app = FirebaseApp.initializeApp(TEST_OPTIONS); @@ -684,6 +740,16 @@ private FirebaseMessaging getMessagingForSend( } private FirebaseMessaging getMessagingForTopicManagement( + Supplier supplier) { + FirebaseApp app = FirebaseApp.initializeApp(TEST_OPTIONS); + return FirebaseMessaging.builder() + .setFirebaseApp(app) + .setMessagingClient(supplier) + .setInstanceIdClient(Suppliers.ofInstance(null)) + .build(); + } + + private FirebaseMessaging getMessagingForLegacyTopicManagement( Supplier supplier) { FirebaseApp app = FirebaseApp.initializeApp(TEST_OPTIONS); return FirebaseMessaging.builder() @@ -697,11 +763,14 @@ private static class MockFirebaseMessagingClient implements FirebaseMessagingCli private String messageId; private BatchResponse batchResponse; + private TopicManagementResponse topicManagementResponse; private FirebaseMessagingException exception; private Message lastMessage; private boolean isLastDryRun; private ImmutableMap messageMap; + private String lastTopic; + private List lastBatch; private MockFirebaseMessagingClient( String messageId, BatchResponse batchResponse, FirebaseMessagingException exception) { @@ -710,6 +779,12 @@ private MockFirebaseMessagingClient( this.exception = exception; } + private MockFirebaseMessagingClient( + TopicManagementResponse topicManagementResponse, FirebaseMessagingException exception) { + this.topicManagementResponse = topicManagementResponse; + this.exception = exception; + } + private MockFirebaseMessagingClient( Map messageMap, FirebaseMessagingException exception) { this.messageMap = ImmutableMap.copyOf(messageMap); @@ -720,6 +795,10 @@ static MockFirebaseMessagingClient fromMessageId(String messageId) { return new MockFirebaseMessagingClient(messageId, null, null); } + static MockFirebaseMessagingClient fromResponse(TopicManagementResponse response) { + return new MockFirebaseMessagingClient(response, null); + } + static MockFirebaseMessagingClient fromMessageMap(Map messageMap) { return new MockFirebaseMessagingClient(messageMap, null); } @@ -753,6 +832,28 @@ public BatchResponse sendAll( List messages, boolean dryRun) throws FirebaseMessagingException { return batchResponse; } + + @Override + public TopicManagementResponse subscribeToTopic( + String topic, List registrationTokens) throws FirebaseMessagingException { + this.lastTopic = topic; + this.lastBatch = registrationTokens; + if (exception != null) { + throw exception; + } + return topicManagementResponse; + } + + @Override + public TopicManagementResponse unsubscribeFromTopic( + String topic, List registrationTokens) throws FirebaseMessagingException { + this.lastTopic = topic; + this.lastBatch = registrationTokens; + if (exception != null) { + throw exception; + } + return topicManagementResponse; + } } private static class MockInstanceIdClient implements InstanceIdClient { From 64fd97d1d8e161db3b21e98850bb8dee1bd8703d Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Thu, 17 Sep 2026 13:43:13 -0400 Subject: [PATCH 2/3] refactor(messaging): reuse shared ExecutorService across topic management requests --- .../FirebaseMessagingClientImpl.java | 85 +++++++++++-------- .../FirebaseMessagingClientImplTest.java | 35 ++++++++ 2 files changed, 84 insertions(+), 36 deletions(-) diff --git a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java b/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java index d59a2b60b..e0c09dce9 100644 --- a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java +++ b/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java @@ -60,7 +60,10 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; /** * A helper class for interacting with Firebase Cloud Messaging service. @@ -101,10 +104,22 @@ private FirebaseMessagingClientImpl(Builder builder) { this.httpClient = new ErrorHandlingHttpClient<>(requestFactory, jsonFactory, errorHandler) .setInterceptor(responseInterceptor); this.batchClient = new MessagingBatchClient(requestFactory.getTransport(), jsonFactory); - this.executor = builder.executor; + this.executor = builder.executor != null + ? builder.executor + : createDefaultExecutor(builder.threadFactory); this.threadFactory = builder.threadFactory; } + private static ExecutorService createDefaultExecutor(ThreadFactory threadFactory) { + ThreadPoolExecutor pool = new ThreadPoolExecutor( + 100, 100, + 60L, TimeUnit.SECONDS, + new LinkedBlockingQueue(), + threadFactory != null ? threadFactory : Executors.defaultThreadFactory()); + pool.allowCoreThreadTimeOut(true); + return pool; + } + @VisibleForTesting String getFcmSendUrl() { return fcmSendUrl; @@ -125,6 +140,16 @@ JsonFactory getJsonFactory() { return jsonFactory; } + @VisibleForTesting + ExecutorService getExecutor() { + return executor; + } + + @VisibleForTesting + ThreadFactory getThreadFactory() { + return threadFactory; + } + public String send(Message message, boolean dryRun) throws FirebaseMessagingException { return sendSingleRequest(message, dryRun); } @@ -212,43 +237,31 @@ public TopicManagementResponse unsubscribeFromTopic( private TopicManagementResponse sendTopicManagementRequest( String topic, List registrationTokens, boolean isSubscribe) { - String topicName = topic.startsWith("/topics/") ? topic.substring("/topics/".length()) : topic; - - ExecutorService pool = this.executor != null - ? this.executor - : (this.threadFactory != null - ? Executors.newFixedThreadPool( - Math.min(registrationTokens.size(), 100), this.threadFactory) - : Executors.newFixedThreadPool(Math.min(registrationTokens.size(), 100))); - boolean shouldShutdown = (this.executor == null); - - try { - List> futures = new ArrayList<>(); - 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), pool)); - } - - CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); - - int successCount = 0; - List errors = new ArrayList<>(); - for (CompletableFuture future : futures) { - TopicResult result = future.join(); - if (result.isSuccess()) { - successCount++; - } else { - errors.add(new TopicManagementResponse.Error(result.getIndex(), result.getReason())); - } - } - return new TopicManagementResponse(successCount, errors); - } finally { - if (shouldShutdown) { - pool.shutdown(); + String topicName = topic.startsWith("/topics/") + ? topic.substring("/topics/".length()) : topic; + + List> futures = new ArrayList<>(); + 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)); + } + + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); + + int successCount = 0; + List errors = new ArrayList<>(); + for (CompletableFuture future : futures) { + TopicResult result = future.join(); + if (result.isSuccess()) { + successCount++; + } else { + errors.add(new TopicManagementResponse.Error(result.getIndex(), result.getReason())); } } + return new TopicManagementResponse(successCount, errors); } private TopicResult sendSingleTopicRequest( diff --git a/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java b/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java index 17b255e7c..3217e26e2 100644 --- a/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java +++ b/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java @@ -53,6 +53,9 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Test; @@ -741,4 +744,36 @@ public void testTopicManagement500Error() throws Exception { assertEquals(1, result.getFailureCount()); assertEquals("internal-error", result.getErrors().get(0).getReason()); } + + @Test + public void testCustomExecutorService() { + ExecutorService customExecutor = Executors.newSingleThreadExecutor(); + try { + FirebaseMessagingClientImpl clientWithExecutor = FirebaseMessagingClientImpl.builder() + .setProjectId("test-project") + .setJsonFactory(ApiClientUtils.getDefaultJsonFactory()) + .setRequestFactory(new MockHttpTransport().createRequestFactory()) + .setChildRequestFactory(ApiClientUtils.getDefaultTransport().createRequestFactory()) + .setExecutor(customExecutor) + .build(); + + assertSame(customExecutor, clientWithExecutor.getExecutor()); + } finally { + customExecutor.shutdown(); + } + } + + @Test + public void testCustomThreadFactory() { + ThreadFactory customThreadFactory = Executors.defaultThreadFactory(); + FirebaseMessagingClientImpl clientWithThreadFactory = FirebaseMessagingClientImpl.builder() + .setProjectId("test-project") + .setJsonFactory(ApiClientUtils.getDefaultJsonFactory()) + .setRequestFactory(new MockHttpTransport().createRequestFactory()) + .setChildRequestFactory(ApiClientUtils.getDefaultTransport().createRequestFactory()) + .setThreadFactory(customThreadFactory) + .build(); + + assertSame(customThreadFactory, clientWithThreadFactory.getThreadFactory()); + } } From d9277ebd52fc091ff8b721c00e92f4b91b177d4b Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Thu, 17 Sep 2026 14:06:55 -0400 Subject: [PATCH 3/3] refactor(messaging): use daemon threads and handle rejected executions --- .../FirebaseMessagingClientImpl.java | 22 ++++- .../FirebaseMessagingClientImplTest.java | 83 +++++++++++++++++++ 2 files changed, 101 insertions(+), 4 deletions(-) diff --git a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java b/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java index e0c09dce9..fd1c6f808 100644 --- a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java +++ b/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(), - 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(); diff --git a/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java b/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java index 3217e26e2..7a178ce8f 100644 --- a/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java +++ b/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 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 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()); + } }