From 0cab00f6e6742ebf599d9927f9b0023f793f13dc Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Tue, 15 Sep 2026 18:20:13 -0400 Subject: [PATCH 1/5] 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 db871372389f8aa7b35464ca545321d40a446d3c Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Thu, 17 Sep 2026 13:43:13 -0400 Subject: [PATCH 2/5] 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 5511d970e49cf3a4b841dea16bf92ea4d48e59c0 Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Thu, 17 Sep 2026 14:06:55 -0400 Subject: [PATCH 3/5] 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()); + } } From 0d686e12383bc3d25c12055c2799b3f580338af2 Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Thu, 24 Sep 2026 12:20:34 -0400 Subject: [PATCH 4/5] refactor(messaging): use app thread pool and sendEachAsync pattern for topic management - Delegate topic subscriptions asynchronously per token using CallableOperation.callAsync(app) in FirebaseMessaging, matching sendEachAsync pattern and respecting developer-configured ThreadManager - Remove internal ExecutorService and ThreadFactory configuration from FirebaseMessagingClientImpl - Simplify extractReason to rely on MessagingErrorHandler populating getMessagingErrorCode() without reparsing raw JSON - Map HTTP 503 to UNAVAILABLE and HTTP 504 to DEADLINE_EXCEEDED in extractReason - Normalize fcmHost during construction by stripping trailing slashes - Update unit tests in FirebaseMessagingTest and FirebaseMessagingClientImplTest --- .../firebase/messaging/FirebaseMessaging.java | 146 ++++++++++-- .../messaging/FirebaseMessagingClient.java | 18 +- .../FirebaseMessagingClientImpl.java | 209 ++---------------- .../FirebaseMessagingClientImplTest.java | 190 ++++------------ .../messaging/FirebaseMessagingTest.java | 139 ++++++++---- 5 files changed, 295 insertions(+), 407 deletions(-) diff --git a/src/main/java/com/google/firebase/messaging/FirebaseMessaging.java b/src/main/java/com/google/firebase/messaging/FirebaseMessaging.java index cc04f65f2..b38e8d827 100644 --- a/src/main/java/com/google/firebase/messaging/FirebaseMessaging.java +++ b/src/main/java/com/google/firebase/messaging/FirebaseMessaging.java @@ -537,7 +537,11 @@ FirebaseMessagingClient getMessagingClient() { */ public TopicManagementResponse subscribeToTopic(@NonNull List registrationTokens, @NonNull String topic) throws FirebaseMessagingException { - return subscribeOp(registrationTokens, topic).call(); + try { + return subscribeToTopicAsync(registrationTokens, topic).get(); + } catch (InterruptedException | ExecutionException e) { + throw new FirebaseMessagingException(ErrorCode.CANCELLED, SERVICE_ID); + } } /** @@ -550,20 +554,7 @@ public TopicManagementResponse subscribeToTopic(@NonNull List registrati */ public ApiFuture subscribeToTopicAsync( @NonNull List registrationTokens, @NonNull String topic) { - return subscribeOp(registrationTokens, topic).callAsync(app); - } - - private CallableOperation subscribeOp( - 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); - } - }; + return manageTopicAsync(registrationTokens, topic, true); } /** @@ -612,7 +603,11 @@ protected TopicManagementResponse execute() throws FirebaseMessagingException { */ public TopicManagementResponse unsubscribeFromTopic(@NonNull List registrationTokens, @NonNull String topic) throws FirebaseMessagingException { - return unsubscribeOp(registrationTokens, topic).call(); + try { + return unsubscribeFromTopicAsync(registrationTokens, topic).get(); + } catch (InterruptedException | ExecutionException e) { + throw new FirebaseMessagingException(ErrorCode.CANCELLED, SERVICE_ID); + } } /** @@ -626,22 +621,129 @@ public TopicManagementResponse unsubscribeFromTopic(@NonNull List regist */ public ApiFuture unsubscribeFromTopicAsync( @NonNull List registrationTokens, @NonNull String topic) { - return unsubscribeOp(registrationTokens, topic).callAsync(app); + return manageTopicAsync(registrationTokens, topic, false); } - private CallableOperation unsubscribeOp( - final List registrationTokens, final String topic) { + private ApiFuture manageTopicAsync( + final List registrationTokens, final String topic, final boolean isSubscribe) { checkRegistrationTokens(registrationTokens); checkTopic(topic); + final String cleanTopic = topic.startsWith("/topics/") + ? topic.substring("/topics/".length()) : topic; + final List immutableTokens = ImmutableList.copyOf(registrationTokens); + + List> futures = new ArrayList<>(immutableTokens.size()); + for (int i = 0; i < immutableTokens.size(); i++) { + futures.add( + manageTopicSingleOp(immutableTokens.get(i), cleanTopic, isSubscribe, i) + .callAsync(app)); + } + + ApiFuture> resultsFuture = ApiFutures.allAsList(futures); + return ApiFutures.transform( + resultsFuture, + (results) -> { + int successCount = 0; + List errors = new ArrayList<>(); + for (TopicResult result : results) { + if (result.isSuccess()) { + successCount++; + } else { + errors.add(new TopicManagementResponse.Error( + result.getIndex(), result.getReason())); + } + } + return new TopicManagementResponse(successCount, errors); + }, + MoreExecutors.directExecutor()); + } + + private CallableOperation manageTopicSingleOp( + final String token, final String topic, final boolean isSubscribe, final int index) { final FirebaseMessagingClient messagingClient = getMessagingClient(); - return new CallableOperation() { + return new CallableOperation() { @Override - protected TopicManagementResponse execute() throws FirebaseMessagingException { - return messagingClient.unsubscribeFromTopic(topic, registrationTokens); + protected TopicResult execute() { + try { + if (isSubscribe) { + messagingClient.subscribeToTopic(topic, token); + } else { + messagingClient.unsubscribeFromTopic(topic, token); + } + return TopicResult.success(index); + } catch (FirebaseMessagingException e) { + return TopicResult.error(index, extractReason(e)); + } catch (Exception e) { + return TopicResult.error(index, "UNKNOWN_ERROR"); + } } }; } + @VisibleForTesting + static String extractReason(FirebaseMessagingException e) { + if (e.getMessagingErrorCode() != null) { + return e.getMessagingErrorCode().name(); + } + 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 "UNAVAILABLE"; + case 504: + 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; + } + } + /** * Unsubscribes a list of registration tokens from a topic using the legacy Instance ID API. * diff --git a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClient.java b/src/main/java/com/google/firebase/messaging/FirebaseMessagingClient.java index d24a57ce0..a4e436a61 100644 --- a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClient.java +++ b/src/main/java/com/google/firebase/messaging/FirebaseMessagingClient.java @@ -30,24 +30,22 @@ interface FirebaseMessagingClient { BatchResponse sendAll(List messages, boolean dryRun) throws FirebaseMessagingException; /** - * Subscribes a list of registration tokens to a topic via the FCM v1 API. + * Subscribes a registration token 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}. + * @param registrationToken A registration token. * @throws FirebaseMessagingException If an error occurs. */ - TopicManagementResponse subscribeToTopic( - String topic, List registrationTokens) throws FirebaseMessagingException; + void subscribeToTopic( + String topic, String registrationToken) throws FirebaseMessagingException; /** - * Unsubscribes a list of registration tokens from a topic via the FCM v1 API. + * Unsubscribes a registration token 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}. + * @param registrationToken A registration token. * @throws FirebaseMessagingException If an error occurs. */ - TopicManagementResponse unsubscribeFromTopic( - String topic, List registrationTokens) throws FirebaseMessagingException; + void unsubscribeFromTopic( + String topic, String registrationToken) 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 fd1c6f808..9b38ac769 100644 --- a/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java +++ b/src/main/java/com/google/firebase/messaging/FirebaseMessagingClientImpl.java @@ -38,7 +38,6 @@ 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; @@ -55,17 +54,8 @@ 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.LinkedBlockingQueue; -import java.util.concurrent.RejectedExecutionException; -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. @@ -90,13 +80,15 @@ 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.projectId = builder.projectId; - this.fcmHost = Strings.isNullOrEmpty(builder.fcmHost) ? DEFAULT_FCM_HOST : builder.fcmHost; + String host = Strings.isNullOrEmpty(builder.fcmHost) ? DEFAULT_FCM_HOST : builder.fcmHost; + while (host.endsWith("/")) { + host = host.substring(0, host.length() - 1); + } + this.fcmHost = host; this.fcmSendUrl = String.format(FCM_URL, this.fcmHost, builder.projectId); this.requestFactory = checkNotNull(builder.requestFactory); this.childRequestFactory = checkNotNull(builder.childRequestFactory); @@ -106,27 +98,6 @@ private FirebaseMessagingClientImpl(Builder builder) { this.httpClient = new ErrorHandlingHttpClient<>(requestFactory, jsonFactory, errorHandler) .setInterceptor(responseInterceptor); this.batchClient = new MessagingBatchClient(requestFactory.getTransport(), jsonFactory); - this.executor = builder.executor != null - ? builder.executor - : createDefaultExecutor(builder.threadFactory); - this.threadFactory = builder.threadFactory; - } - - 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(), - factory); - pool.allowCoreThreadTimeOut(true); - return pool; } @VisibleForTesting @@ -149,16 +120,6 @@ 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); } @@ -233,54 +194,22 @@ public void initialize(HttpRequest request) throws IOException { } @Override - public TopicManagementResponse subscribeToTopic( - String topic, List registrationTokens) throws FirebaseMessagingException { - return sendTopicManagementRequest(topic, registrationTokens, true); + public void subscribeToTopic( + String topic, String registrationToken) throws FirebaseMessagingException { + sendSingleTopicRequest(registrationToken, topic, true); } @Override - public TopicManagementResponse unsubscribeFromTopic( - String topic, List registrationTokens) throws FirebaseMessagingException { - return sendTopicManagementRequest(topic, registrationTokens, false); + public void unsubscribeFromTopic( + String topic, String registrationToken) throws FirebaseMessagingException { + sendSingleTopicRequest(registrationToken, topic, false); } - private TopicManagementResponse sendTopicManagementRequest( - String topic, List registrationTokens, boolean isSubscribe) { - 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); - 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(); - - 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( - String token, String topicName, boolean isSubscribe, int index) { + private void sendSingleTopicRequest( + String token, String topic, boolean isSubscribe) throws FirebaseMessagingException { try { + String topicName = topic.startsWith("/topics/") + ? topic.substring("/topics/".length()) : topic; String encodedToken = URLEncoder.encode(token, StandardCharsets.UTF_8.name()); String encodedTopic = URLEncoder.encode(topicName, StandardCharsets.UTF_8.name()); HttpRequestInfo requestInfo; @@ -299,15 +228,13 @@ private TopicResult sendSingleTopicRequest( } httpClient.send(requestInfo); - return TopicResult.success(index); } catch (FirebaseMessagingException e) { if (isSubscribe && isAlreadyExists(e)) { - return TopicResult.success(index); + return; } - String reason = extractReason(e); - return TopicResult.error(index, reason); - } catch (Exception e) { - return TopicResult.error(index, "UNKNOWN_ERROR"); + throw e; + } catch (IOException e) { + throw errorHandler.handleIOException(e); } } @@ -321,103 +248,17 @@ private boolean isAlreadyExists(FirebaseMessagingException e) { 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(); } @@ -433,8 +274,6 @@ static final class Builder { private HttpRequestFactory childRequestFactory; private JsonFactory jsonFactory; private HttpResponseInterceptor responseInterceptor; - private ExecutorService executor; - private ThreadFactory threadFactory; private Builder() { } @@ -468,16 +307,6 @@ 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/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java b/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java index 7a178ce8f..8052beb36 100644 --- a/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java +++ b/src/test/java/com/google/firebase/messaging/FirebaseMessagingClientImplTest.java @@ -53,15 +53,7 @@ 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; @@ -663,16 +655,13 @@ private static Map> buildTestMessages() { @Test public void testSubscribeToTopic() throws Exception { response.setContent("{}"); - TopicManagementResponse result = client.subscribeToTopic( - "test-topic", ImmutableList.of("id1")); + client.subscribeToTopic("test-topic", "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", + "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")); @@ -682,44 +671,35 @@ public void testSubscribeToTopic() throws Exception { @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()); + client.subscribeToTopic("test-topic", "id1"); } @Test public void testUnsubscribeFromTopic() throws Exception { response.setContent("{}"); - TopicManagementResponse result = client.unsubscribeFromTopic( - "test-topic", ImmutableList.of("id1")); + client.unsubscribeFromTopic("test-topic", "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", + "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 { + public void testUnsubscribeFromTopic404() { 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()); + try { + client.unsubscribeFromTopic("test-topic", "id1"); + fail("No error thrown"); + } catch (FirebaseMessagingException e) { + assertEquals(ErrorCode.NOT_FOUND, e.getErrorCode()); + } } @Test - public void testTopicManagementFcmErrorDetails() throws Exception { + public void testTopicManagementFcmErrorDetails() { response.setStatusCode(404).setContent("{\n" + " \"error\": {\n" + " \"status\": \"NOT_FOUND\",\n" @@ -731,132 +711,48 @@ public void testTopicManagementFcmErrorDetails() throws Exception { + " ]\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()); + try { + client.subscribeToTopic("test-topic", "id1"); + fail("No error thrown"); + } catch (FirebaseMessagingException e) { + assertEquals(MessagingErrorCode.UNREGISTERED, e.getMessagingErrorCode()); + } } @Test - public void testTopicManagement500Error() throws Exception { + public void testTopicManagement500Error() { 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()); - } - - @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(); + client.subscribeToTopic("test-topic", "id1"); + fail("No error thrown"); + } catch (FirebaseMessagingException e) { + assertEquals(ErrorCode.INTERNAL, e.getErrorCode()); } } @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) + public void testFcmHostTrailingSlash() throws Exception { + TestResponseInterceptor testInterceptor = new TestResponseInterceptor(); + MockHttpTransport transport = new MockHttpTransport.Builder() + .setLowLevelHttpResponse(new MockLowLevelHttpResponse().setContent("{}")) .build(); - - 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() + FirebaseMessagingClientImpl clientWithSlash = FirebaseMessagingClientImpl.builder() .setProjectId("test-project") + .setFcmHost("https://custom.fcm.host///") .setJsonFactory(ApiClientUtils.getDefaultJsonFactory()) - .setRequestFactory(new MockHttpTransport().createRequestFactory()) + .setRequestFactory(transport.createRequestFactory()) .setChildRequestFactory(ApiClientUtils.getDefaultTransport().createRequestFactory()) - .setExecutor(rejectingExecutor) + .setResponseInterceptor(testInterceptor) .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()); + clientWithSlash.subscribeToTopic("test-topic", "id1"); + HttpRequest request = testInterceptor.getLastRequest(); + assertEquals( + "https://custom.fcm.host/v1/projects/test-project/registrations/id1" + + "/topicSubscriptions?topic_name=test-topic", + request.getUrl().toString()); + assertEquals( + "https://custom.fcm.host/v1/projects/test-project/messages:send", + clientWithSlash.getFcmSendUrl()); } } diff --git a/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java b/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java index 6444dbc99..e572d4258 100644 --- a/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java +++ b/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java @@ -26,6 +26,8 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import com.google.api.client.http.HttpHeaders; +import com.google.api.client.http.HttpResponseException; import com.google.api.client.json.GenericJson; import com.google.common.base.Supplier; import com.google.common.base.Suppliers; @@ -33,16 +35,18 @@ import com.google.common.collect.ImmutableMap; import com.google.firebase.ErrorCode; import com.google.firebase.FirebaseApp; +import com.google.firebase.FirebaseException; import com.google.firebase.FirebaseOptions; +import com.google.firebase.IncomingHttpResponse; +import com.google.firebase.OutgoingHttpRequest; import com.google.firebase.TestOnlyImplFirebaseTrampolines; import com.google.firebase.auth.MockGoogleCredentials; import com.google.firebase.internal.Nullable; - +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ExecutionException; - import org.junit.After; import org.junit.Test; @@ -572,21 +576,26 @@ public void testSubscribeToTopic() throws FirebaseMessagingException { TopicManagementResponse got = messaging.subscribeToTopic( ImmutableList.of("id1", "id2"), "test-topic"); - assertSame(TOPIC_MGT_RESPONSE, got); + assertEquals(2, got.getSuccessCount()); + assertEquals(0, got.getFailureCount()); + assertTrue(got.getErrors().isEmpty()); assertEquals("test-topic", client.lastTopic); assertEquals(ImmutableList.of("id1", "id2"), client.lastBatch); } @Test - public void testSubscribeToTopicFailure() { + public void testSubscribeToTopicFailure() throws FirebaseMessagingException { MockFirebaseMessagingClient client = MockFirebaseMessagingClient.fromException(TEST_EXCEPTION); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); - try { - messaging.subscribeToTopic(ImmutableList.of("id1", "id2"), "test-topic"); - } catch (FirebaseMessagingException e) { - assertSame(TEST_EXCEPTION, e); - } + TopicManagementResponse got = messaging.subscribeToTopic( + ImmutableList.of("id1", "id2"), "test-topic"); + + assertEquals(0, got.getSuccessCount()); + assertEquals(2, got.getFailureCount()); + assertEquals(2, got.getErrors().size()); + assertEquals(0, got.getErrors().get(0).getIndex()); + assertEquals(1, got.getErrors().get(1).getIndex()); } @Test @@ -598,19 +607,24 @@ public void testSubscribeToTopicAsync() throws Exception { TopicManagementResponse got = messaging.subscribeToTopicAsync( ImmutableList.of("id1", "id2"), "test-topic").get(); - assertSame(TOPIC_MGT_RESPONSE, got); + assertEquals(2, got.getSuccessCount()); + assertEquals(0, got.getFailureCount()); + assertTrue(got.getErrors().isEmpty()); + assertEquals("test-topic", client.lastTopic); + assertEquals(ImmutableList.of("id1", "id2"), client.lastBatch); } @Test - public void testSubscribeToTopicAsyncFailure() throws InterruptedException { + public void testSubscribeToTopicAsyncFailure() throws Exception { MockFirebaseMessagingClient client = MockFirebaseMessagingClient.fromException(TEST_EXCEPTION); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); - try { - messaging.subscribeToTopicAsync(ImmutableList.of("id1", "id2"), "test-topic").get(); - } catch (ExecutionException e) { - assertSame(TEST_EXCEPTION, e.getCause()); - } + TopicManagementResponse got = messaging.subscribeToTopicAsync( + ImmutableList.of("id1", "id2"), "test-topic").get(); + + assertEquals(0, got.getSuccessCount()); + assertEquals(2, got.getFailureCount()); + assertEquals(2, got.getErrors().size()); } @Test @@ -664,21 +678,26 @@ public void testUnsubscribeFromTopic() throws FirebaseMessagingException { TopicManagementResponse got = messaging.unsubscribeFromTopic( ImmutableList.of("id1", "id2"), "test-topic"); - assertSame(TOPIC_MGT_RESPONSE, got); + assertEquals(2, got.getSuccessCount()); + assertEquals(0, got.getFailureCount()); + assertTrue(got.getErrors().isEmpty()); assertEquals("test-topic", client.lastTopic); assertEquals(ImmutableList.of("id1", "id2"), client.lastBatch); } @Test - public void testUnsubscribeFromTopicFailure() { + public void testUnsubscribeFromTopicFailure() throws FirebaseMessagingException { MockFirebaseMessagingClient client = MockFirebaseMessagingClient.fromException(TEST_EXCEPTION); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); - try { - messaging.unsubscribeFromTopic(ImmutableList.of("id1", "id2"), "test-topic"); - } catch (FirebaseMessagingException e) { - assertSame(TEST_EXCEPTION, e); - } + TopicManagementResponse got = messaging.unsubscribeFromTopic( + ImmutableList.of("id1", "id2"), "test-topic"); + + assertEquals(0, got.getSuccessCount()); + assertEquals(2, got.getFailureCount()); + assertEquals(2, got.getErrors().size()); + assertEquals(0, got.getErrors().get(0).getIndex()); + assertEquals(1, got.getErrors().get(1).getIndex()); } @Test @@ -690,19 +709,59 @@ public void testUnsubscribeFromTopicAsync() throws Exception { TopicManagementResponse got = messaging.unsubscribeFromTopicAsync( ImmutableList.of("id1", "id2"), "test-topic").get(); - assertSame(TOPIC_MGT_RESPONSE, got); + assertEquals(2, got.getSuccessCount()); + assertEquals(0, got.getFailureCount()); + assertTrue(got.getErrors().isEmpty()); + assertEquals("test-topic", client.lastTopic); + assertEquals(ImmutableList.of("id1", "id2"), client.lastBatch); } @Test - public void testUnsubscribeFromTopicAsyncFailure() throws InterruptedException { + public void testUnsubscribeFromTopicAsyncFailure() throws Exception { MockFirebaseMessagingClient client = MockFirebaseMessagingClient.fromException(TEST_EXCEPTION); FirebaseMessaging messaging = getMessagingForTopicManagement(Suppliers.ofInstance(client)); - try { - messaging.unsubscribeFromTopicAsync(ImmutableList.of("id1", "id2"), "test-topic").get(); - } catch (ExecutionException e) { - assertSame(TEST_EXCEPTION, e.getCause()); - } + TopicManagementResponse got = messaging.unsubscribeFromTopicAsync( + ImmutableList.of("id1", "id2"), "test-topic").get(); + + assertEquals(0, got.getSuccessCount()); + assertEquals(2, got.getFailureCount()); + assertEquals(2, got.getErrors().size()); + } + + @Test + public void testExtractReason() { + FirebaseMessagingException messagingError = + FirebaseMessagingException.withMessagingErrorCode( + new FirebaseException(ErrorCode.INVALID_ARGUMENT, "bad arg", null), + MessagingErrorCode.UNREGISTERED); + assertEquals("UNREGISTERED", FirebaseMessaging.extractReason(messagingError)); + + FirebaseMessagingException platformError = + new FirebaseMessagingException(ErrorCode.PERMISSION_DENIED, "permission denied"); + assertEquals("PERMISSION_DENIED", FirebaseMessaging.extractReason(platformError)); + + IncomingHttpResponse resp503 = new IncomingHttpResponse( + new HttpResponseException.Builder(503, "Unavailable", new HttpHeaders()).build(), + new OutgoingHttpRequest("GET", "https://example.com")); + FirebaseMessagingException unavailableError = + FirebaseMessagingException.withMessagingErrorCode( + new FirebaseException(ErrorCode.UNKNOWN, "unavailable", null, resp503), + null); + assertEquals("UNAVAILABLE", FirebaseMessaging.extractReason(unavailableError)); + + IncomingHttpResponse resp504 = new IncomingHttpResponse( + new HttpResponseException.Builder(504, "Gateway Timeout", new HttpHeaders()).build(), + new OutgoingHttpRequest("GET", "https://example.com")); + FirebaseMessagingException timeoutError = + FirebaseMessagingException.withMessagingErrorCode( + new FirebaseException(ErrorCode.UNKNOWN, "timeout", null, resp504), + null); + assertEquals("DEADLINE_EXCEEDED", FirebaseMessaging.extractReason(timeoutError)); + + FirebaseMessagingException unknownError = + new FirebaseMessagingException(ErrorCode.UNKNOWN, "something unknown"); + assertEquals("UNKNOWN_ERROR", FirebaseMessaging.extractReason(unknownError)); } @Test @@ -834,25 +893,29 @@ public BatchResponse sendAll( } @Override - public TopicManagementResponse subscribeToTopic( - String topic, List registrationTokens) throws FirebaseMessagingException { + public void subscribeToTopic( + String topic, String registrationToken) throws FirebaseMessagingException { this.lastTopic = topic; - this.lastBatch = registrationTokens; + if (this.lastBatch == null) { + this.lastBatch = new ArrayList<>(); + } + this.lastBatch.add(registrationToken); if (exception != null) { throw exception; } - return topicManagementResponse; } @Override - public TopicManagementResponse unsubscribeFromTopic( - String topic, List registrationTokens) throws FirebaseMessagingException { + public void unsubscribeFromTopic( + String topic, String registrationToken) throws FirebaseMessagingException { this.lastTopic = topic; - this.lastBatch = registrationTokens; + if (this.lastBatch == null) { + this.lastBatch = new ArrayList<>(); + } + this.lastBatch.add(registrationToken); if (exception != null) { throw exception; } - return topicManagementResponse; } } From 77b0a38bc9f5e34194a3cddcfb08fe62613221e5 Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Thu, 24 Sep 2026 12:40:14 -0400 Subject: [PATCH 5/5] test(messaging): ensure thread-safe and order-independent topic batch assertions --- .../firebase/messaging/FirebaseMessagingTest.java | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java b/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java index e572d4258..02e05fdd6 100644 --- a/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java +++ b/src/test/java/com/google/firebase/messaging/FirebaseMessagingTest.java @@ -33,6 +33,7 @@ import com.google.common.base.Suppliers; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; import com.google.firebase.ErrorCode; import com.google.firebase.FirebaseApp; import com.google.firebase.FirebaseException; @@ -580,7 +581,7 @@ public void testSubscribeToTopic() throws FirebaseMessagingException { assertEquals(0, got.getFailureCount()); assertTrue(got.getErrors().isEmpty()); assertEquals("test-topic", client.lastTopic); - assertEquals(ImmutableList.of("id1", "id2"), client.lastBatch); + assertEquals(ImmutableSet.of("id1", "id2"), ImmutableSet.copyOf(client.lastBatch)); } @Test @@ -611,7 +612,7 @@ public void testSubscribeToTopicAsync() throws Exception { assertEquals(0, got.getFailureCount()); assertTrue(got.getErrors().isEmpty()); assertEquals("test-topic", client.lastTopic); - assertEquals(ImmutableList.of("id1", "id2"), client.lastBatch); + assertEquals(ImmutableSet.of("id1", "id2"), ImmutableSet.copyOf(client.lastBatch)); } @Test @@ -682,7 +683,7 @@ public void testUnsubscribeFromTopic() throws FirebaseMessagingException { assertEquals(0, got.getFailureCount()); assertTrue(got.getErrors().isEmpty()); assertEquals("test-topic", client.lastTopic); - assertEquals(ImmutableList.of("id1", "id2"), client.lastBatch); + assertEquals(ImmutableSet.of("id1", "id2"), ImmutableSet.copyOf(client.lastBatch)); } @Test @@ -713,7 +714,7 @@ public void testUnsubscribeFromTopicAsync() throws Exception { assertEquals(0, got.getFailureCount()); assertTrue(got.getErrors().isEmpty()); assertEquals("test-topic", client.lastTopic); - assertEquals(ImmutableList.of("id1", "id2"), client.lastBatch); + assertEquals(ImmutableSet.of("id1", "id2"), ImmutableSet.copyOf(client.lastBatch)); } @Test @@ -893,7 +894,7 @@ public BatchResponse sendAll( } @Override - public void subscribeToTopic( + public synchronized void subscribeToTopic( String topic, String registrationToken) throws FirebaseMessagingException { this.lastTopic = topic; if (this.lastBatch == null) { @@ -906,7 +907,7 @@ public void subscribeToTopic( } @Override - public void unsubscribeFromTopic( + public synchronized void unsubscribeFromTopic( String topic, String registrationToken) throws FirebaseMessagingException { this.lastTopic = topic; if (this.lastBatch == null) {