feat(fcm): Migrate topic management to FCM v1 API

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

@@ -537,7 +537,11 @@ FirebaseMessagingClient getMessagingClient() {
    */
   public TopicManagementResponse subscribeToTopic(@NonNull List<String> 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,10 +554,33 @@ public TopicManagementResponse subscribeToTopic(@NonNull List<String> registrati
    */
   public ApiFuture<TopicManagementResponse> subscribeToTopicAsync(
       @NonNull List<String> registrationTokens, @NonNull String topic) {
-    return subscribeOp(registrationTokens, topic).callAsync(app);
+    return manageTopicAsync(registrationTokens, topic, true);
+  }
+
+  /**
+   * 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<String> 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<TopicManagementResponse> subscribeToTopicLegacyAsync(
+      @NonNull List<String> registrationTokens, @NonNull String topic) {
+    return subscribeLegacyOp(registrationTokens, topic).callAsync(app);
   }
 
-  private CallableOperation<TopicManagementResponse, FirebaseMessagingException> subscribeOp(
+  private CallableOperation<TopicManagementResponse, FirebaseMessagingException> subscribeLegacyOp(
       final List<String> registrationTokens, final String topic) {
     checkRegistrationTokens(registrationTokens);
     checkTopic(topic);
@@ -576,7 +603,11 @@ protected TopicManagementResponse execute() throws FirebaseMessagingException {
    */
   public TopicManagementResponse unsubscribeFromTopic(@NonNull List<String> 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);
+    }
   }
 
   /**
@@ -590,11 +621,155 @@ public TopicManagementResponse unsubscribeFromTopic(@NonNull List<String> regist
    */
   public ApiFuture<TopicManagementResponse> unsubscribeFromTopicAsync(
       @NonNull List<String> registrationTokens, @NonNull String topic) {
-    return unsubscribeOp(registrationTokens, topic).callAsync(app);
+    return manageTopicAsync(registrationTokens, topic, false);
   }
 
-  private CallableOperation<TopicManagementResponse, FirebaseMessagingException> unsubscribeOp(
-      final List<String> registrationTokens, final String topic) {
+  private ApiFuture<TopicManagementResponse> manageTopicAsync(
+      final List<String> registrationTokens, final String topic, final boolean isSubscribe) {
+    checkRegistrationTokens(registrationTokens);
+    checkTopic(topic);
+    final String cleanTopic = topic.startsWith("/topics/")
+        ? topic.substring("/topics/".length()) : topic;
+    final List<String> immutableTokens = ImmutableList.copyOf(registrationTokens);
+
+    List<ApiFuture<TopicResult>> 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<List<TopicResult>> resultsFuture = ApiFutures.allAsList(futures);
+    return ApiFutures.transform(
+        resultsFuture,
+        (results) -> {
+          int successCount = 0;
+          List<TopicManagementResponse.Error> 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<TopicResult, FirebaseMessagingException> manageTopicSingleOp(
+      final String token, final String topic, final boolean isSubscribe, final int index) {
+    final FirebaseMessagingClient messagingClient = getMessagingClient();
+    return new CallableOperation<TopicResult, FirebaseMessagingException>() {
+      @Override
+      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.
+   *
+   * @deprecated Use {@link #unsubscribeFromTopic(List, String)} instead.
+   */
+  @Deprecated
+  public TopicManagementResponse unsubscribeFromTopicLegacy(
+      @NonNull List<String> 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<TopicManagementResponse> unsubscribeFromTopicLegacyAsync(
+      @NonNull List<String> registrationTokens, @NonNull String topic) {
+    return unsubscribeLegacyOp(registrationTokens, topic).callAsync(app);
+  }
+
+  private CallableOperation<TopicManagementResponse, FirebaseMessagingException>
+      unsubscribeLegacyOp(final List<String> registrationTokens, final String topic) {
     checkRegistrationTokens(registrationTokens);
     checkTopic(topic);
     final InstanceIdClient instanceIdClient = getInstanceIdClient();

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

@@ -29,4 +29,23 @@ interface FirebaseMessagingClient {
    */
   BatchResponse sendAll(List<Message> messages, boolean dryRun) throws FirebaseMessagingException;
 
+  /**
+   * Subscribes a registration token to a topic via the FCM v1 API.
+   *
+   * @param topic Name of the topic.
+   * @param registrationToken A registration token.
+   * @throws FirebaseMessagingException If an error occurs.
+   */
+  void subscribeToTopic(
+      String topic, String registrationToken) throws FirebaseMessagingException;
+
+  /**
+   * Unsubscribes a registration token from a topic via the FCM v1 API.
+   *
+   * @param topic Name of the topic.
+   * @param registrationToken A registration token.
+   * @throws FirebaseMessagingException If an error occurs.
+   */
+  void unsubscribeFromTopic(
+      String topic, String registrationToken) throws FirebaseMessagingException;
 }

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

@@ -52,6 +52,8 @@
 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.List;
 import java.util.Map;
 
@@ -60,13 +62,16 @@
  */
 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<String, String> 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;
@@ -78,7 +83,13 @@ final class FirebaseMessagingClientImpl implements FirebaseMessagingClient {
 
   private FirebaseMessagingClientImpl(Builder builder) {
     checkArgument(!Strings.isNullOrEmpty(builder.projectId));
-    this.fcmSendUrl = String.format(FCM_URL, builder.projectId);
+    this.projectId = builder.projectId;
+    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);
     this.jsonFactory = checkNotNull(builder.jsonFactory);
@@ -182,6 +193,61 @@ public void initialize(HttpRequest request) throws IOException {
     };
   }
 
+  @Override
+  public void subscribeToTopic(
+      String topic, String registrationToken) throws FirebaseMessagingException {
+    sendSingleTopicRequest(registrationToken, topic, true);
+  }
+
+  @Override
+  public void unsubscribeFromTopic(
+      String topic, String registrationToken) throws FirebaseMessagingException {
+    sendSingleTopicRequest(registrationToken, topic, false);
+  }
+
+  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;
+      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);
+    } catch (FirebaseMessagingException e) {
+      if (isSubscribe && isAlreadyExists(e)) {
+        return;
+      }
+      throw e;
+    } catch (IOException e) {
+      throw errorHandler.handleIOException(e);
+    }
+  }
+
+  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;
+  }
+
   static FirebaseMessagingClientImpl fromApp(FirebaseApp app) {
     String projectId = ImplFirebaseTrampolines.getProjectId(app);
     checkArgument(!Strings.isNullOrEmpty(projectId),
@@ -203,6 +269,7 @@ 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;
@@ -215,6 +282,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;

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

@@ -61,6 +61,11 @@ public class TopicManagementResponse {
     this.errors = errors.build();
   }
 
+  TopicManagementResponse(int successCount, List<Error> 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;