From 4af44f866dba7e03663300e08c4cb27383d0ec41 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 5 Nov 2026 13:12:57 +0000 Subject: [PATCH] Add FCM topics REST API Wraps the /push/fcmTopics endpoints (ably/realtime#8809), which let FCM fan a single send out to every Android device subscribed to a topic. - push.admin.fcmTopics: subscribe/unsubscribe a device, bulk subscribe by clientId or channels, unsubscribe by clientId, list a device's topics, and publish to a topic (returns the FCM message ID). - push.fcmTopics (Android): subscribe/unsubscribe/list for the local device, authenticated with its identity token. Argument errors are reported as AblyException through the request, and the shared "device not activated" check moves from PushChannel to Push. Co-Authored-By: Claude Opus 5.5 --- .../java/io/ably/pubsub/push/FcmTopic.java | 77 +++++ .../java/io/ably/pubsub/push/FcmTopics.java | 65 ++++ .../main/java/io/ably/pubsub/push/Push.java | 48 +++ .../java/io/ably/pubsub/push/PushChannel.java | 12 +- .../io/ably/pubsub/push/AdminFcmTopic.java | 287 ++++++++++++++++++ .../io/ably/pubsub/push/AdminFcmTopics.java | 217 +++++++++++++ .../java/io/ably/pubsub/push/PushBase.java | 9 +- .../io/ably/pubsub/push/AdminFcmTopicsTest.kt | 176 +++++++++++ 8 files changed, 878 insertions(+), 13 deletions(-) create mode 100644 core-android/src/main/java/io/ably/pubsub/push/FcmTopic.java create mode 100644 core-android/src/main/java/io/ably/pubsub/push/FcmTopics.java create mode 100644 lib/src/main/java/io/ably/pubsub/push/AdminFcmTopic.java create mode 100644 lib/src/main/java/io/ably/pubsub/push/AdminFcmTopics.java create mode 100644 lib/src/test/kotlin/io/ably/pubsub/push/AdminFcmTopicsTest.kt diff --git a/core-android/src/main/java/io/ably/pubsub/push/FcmTopic.java b/core-android/src/main/java/io/ably/pubsub/push/FcmTopic.java new file mode 100644 index 000000000..acc7539fb --- /dev/null +++ b/core-android/src/main/java/io/ably/pubsub/push/FcmTopic.java @@ -0,0 +1,77 @@ +package io.ably.pubsub.push; + +import io.ably.pubsub.http.HttpRequester; +import io.ably.pubsub.http.PubSubBase; +import io.ably.pubsub.realtime.CompletionListener; +import io.ably.pubsub.types.AblyException; + +/** + * A single FCM topic that this device can subscribe to or unsubscribe from. + * Obtained from {@link FcmTopics#get(String)}. + */ +public class FcmTopic { + /** + * The FCM topic name. + */ + public final String name; + + private final PubSubBase rest; + private final Push push; + + FcmTopic(PubSubBase rest, Push push, String name) { + this.rest = rest; + this.push = push; + this.name = name; + } + + /** + * Subscribes this device to the topic. Subscribing when already subscribed succeeds. + *

+ * If FCM rejects the device's registration token, Ably marks the device registration failed + * and the request fails; the device must then activate again. + * + * @throws AblyException + */ + public void subscribe() throws AblyException { + subscribeImpl().sync(); + } + + /** + * Asynchronously subscribes this device to the topic. See {@link #subscribe()}. + * + * @param listener A listener to be notified of success or failure. + *

+ * This listener is invoked on a background thread. + */ + public void subscribeAsync(CompletionListener listener) { + subscribeImpl().async(new CompletionListener.ToCallback<>(listener)); + } + + /** + * Unsubscribes this device from the topic. Unsubscribing when not subscribed succeeds. + * + * @throws AblyException + */ + public void unsubscribe() throws AblyException { + unsubscribeImpl().sync(); + } + + /** + * Asynchronously unsubscribes this device from the topic. See {@link #unsubscribe()}. + * + * @param listener A listener to be notified of success or failure. + *

+ * This listener is invoked on a background thread. + */ + public void unsubscribeAsync(CompletionListener listener) { + unsubscribeImpl().async(new CompletionListener.ToCallback<>(listener)); + } + + private HttpRequester.Request subscribeImpl() { + return push.localDeviceRequest((deviceId, headers) -> AdminFcmTopics.subscribeDeviceImpl(rest, headers, name, deviceId)); + } + + private HttpRequester.Request unsubscribeImpl() { + return push.localDeviceRequest((deviceId, headers) -> AdminFcmTopics.unsubscribeDeviceImpl(rest, headers, name, deviceId)); + } +} diff --git a/core-android/src/main/java/io/ably/pubsub/push/FcmTopics.java b/core-android/src/main/java/io/ably/pubsub/push/FcmTopics.java new file mode 100644 index 000000000..00dc439f0 --- /dev/null +++ b/core-android/src/main/java/io/ably/pubsub/push/FcmTopics.java @@ -0,0 +1,65 @@ +package io.ably.pubsub.push; + +import io.ably.pubsub.http.HttpRequester; +import io.ably.pubsub.http.PubSubBase; +import io.ably.pubsub.types.AblyException; +import io.ably.pubsub.types.Callback; + +/** + * Manages this device's FCM topic subscriptions. + *

+ * FCM topics let FCM fan a single send out to every device subscribed to the topic. A topic exists + * once its first device subscribes; there is no create or delete call. Topic names must match + * {@code [a-zA-Z0-9-_.~%]{1,900}}. FCM allows at most 2,000 topic subscriptions per device and never + * removes them automatically, so unsubscribe when the device no longer needs a topic. + *

+ * The device must be activated for push with FCM ({@link Push#activate()} must have completed) before + * using these methods. Requests authenticate as the device, so no push-admin capability is needed. + */ +public class FcmTopics { + private final PubSubBase rest; + private final Push push; + + FcmTopics(PubSubBase rest, Push push) { + this.rest = rest; + this.push = push; + } + + /** + * Returns an {@link FcmTopic} for subscribing this device to, or unsubscribing it from, the named topic. + * An empty topic name is reported as an {@link AblyException} by the operations on the returned object. + * + * @param topic The FCM topic name. + * @return An {@link FcmTopic} object. + */ + public FcmTopic get(String topic) { + return new FcmTopic(rest, push, topic); + } + + /** + * Retrieves the names of the FCM topics this device is subscribed to, + * including any it subscribed to other than through Ably. + * + * @return The topic names. + * @throws AblyException + */ + public String[] list() throws AblyException { + return listImpl().sync(); + } + + /** + * Asynchronously retrieves the names of the FCM topics this device is subscribed to, + * including any it subscribed to other than through Ably. + * + * @param callback A callback returning the topic names. + *

+ * This callback is invoked on a background thread. + */ + public void listAsync(Callback callback) { + listImpl().async(callback); + } + + private HttpRequester.Request listImpl() { + return push.localDeviceRequest((deviceId, headers) -> AdminFcmTopics.listImpl(rest, headers, deviceId)); + } +} diff --git a/core-android/src/main/java/io/ably/pubsub/push/Push.java b/core-android/src/main/java/io/ably/pubsub/push/Push.java index 05be1ee29..2ce66e5ab 100644 --- a/core-android/src/main/java/io/ably/pubsub/push/Push.java +++ b/core-android/src/main/java/io/ably/pubsub/push/Push.java @@ -3,6 +3,7 @@ import android.content.Context; import android.content.SharedPreferences; import android.preference.PreferenceManager; +import io.ably.pubsub.http.HttpRequester; import io.ably.pubsub.http.HttpUtils; import io.ably.pubsub.http.PubSubBase; import io.ably.pubsub.http.PubSubHttpClient; @@ -18,8 +19,14 @@ */ public class Push extends PushBase { + /** + * An {@link FcmTopics} object, for managing this device's FCM topic subscriptions. + */ + public final FcmTopics fcmTopics; + public Push(PubSubBase rest) { super(rest); + this.fcmTopics = new FcmTopics(rest, this); } /** @@ -122,6 +129,47 @@ public LocalDevice getLocalDevice() throws AblyException { return getActivationContext().getLocalDevice(); } + /** + * Returns the local device, failing if push activation has not completed, since the device has + * no ID or identity token to authenticate push requests with until then. + */ + LocalDevice requireActivatedDevice() throws AblyException { + LocalDevice localDevice = getLocalDevice(); + if (localDevice == null || localDevice.deviceIdentityToken == null) { + // Alternatively, we could store a queue of pending subscriptions in the + // device storage. But then, in order to know if this subscription operation + // succeeded, you would have to add a BroadcastReceiver in AndroidManifest.xml. + // Arguably that encourages just ignoring any errors, and forcing you to listen + // to the broadcast after push.activate has finished before subscribing is + // more robust. + throw AblyException.fromThrowable(new Exception("cannot use device before PubSubHttpClient.push.activate has finished")); + } + return localDevice; + } + + /** + * Builds a request for the local device, authenticated as it. + */ + interface LocalDeviceRequest { + HttpRequester.Request build(String deviceId, Param[] headers); + } + + /** + * Builds a request authenticated as the local device, or a failed request if push activation has + * not completed. The device ID and identity headers come from the same device read, so they + * cannot disagree if the device is reset concurrently. + */ + HttpRequester.Request localDeviceRequest(LocalDeviceRequest request) { + LocalDevice device; + try { + device = requireActivatedDevice(); + } catch (AblyException e) { + return rest.http.failedRequest(e); + } + Param[] headers = HttpUtils.mergeHeaders(super.pushRequestHeaders(false), device.deviceIdentityHeaders()); + return request.build(device.id, headers); + } + @Override Param[] pushRequestHeaders(boolean forLocalDevice) { Param[] headers = super.pushRequestHeaders(forLocalDevice); diff --git a/core-android/src/main/java/io/ably/pubsub/push/PushChannel.java b/core-android/src/main/java/io/ably/pubsub/push/PushChannel.java index 550ec6261..1c560559d 100644 --- a/core-android/src/main/java/io/ably/pubsub/push/PushChannel.java +++ b/core-android/src/main/java/io/ably/pubsub/push/PushChannel.java @@ -249,16 +249,6 @@ protected String getClientId() throws AblyException { } protected DeviceDetails getDevice() throws AblyException { - LocalDevice localDevice = rest.push.getActivationContext().getLocalDevice(); - if (localDevice == null || localDevice.deviceIdentityToken == null) { - // Alternatively, we could store a queue of pending subscriptions in the - // device storage. But then, in order to know if this subscription operation - // succeeded, you would have to add a BroadcastReceiver in AndroidManifest.xml. - // Arguably that encourages just ignoring any errors, and forcing you to listen - // to the broadcast after push.activate has finished before subscribing is - // more robust. - throw AblyException.fromThrowable(new Exception("cannot use device before PubSubHttpClient.push.activate has finished")); - } - return localDevice; + return rest.push.requireActivatedDevice(); } } diff --git a/lib/src/main/java/io/ably/pubsub/push/AdminFcmTopic.java b/lib/src/main/java/io/ably/pubsub/push/AdminFcmTopic.java new file mode 100644 index 000000000..3a9b77a44 --- /dev/null +++ b/lib/src/main/java/io/ably/pubsub/push/AdminFcmTopic.java @@ -0,0 +1,287 @@ +package io.ably.pubsub.push; + +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import io.ably.pubsub.http.HttpCore; +import io.ably.pubsub.http.HttpRequester; +import io.ably.pubsub.http.HttpScheduler; +import io.ably.pubsub.http.HttpUtils; +import io.ably.pubsub.http.PubSubBase; +import io.ably.pubsub.realtime.CompletionListener; +import io.ably.pubsub.types.AblyException; +import io.ably.pubsub.types.Callback; +import io.ably.pubsub.types.ErrorInfo; +import io.ably.pubsub.types.Param; +import io.ably.pubsub.util.Log; +import io.ably.pubsub.util.ParamsUtils; + +import java.util.Arrays; + +/** + * A single FCM topic, managed with push-admin privileges. Obtained from {@link AdminFcmTopics#get(String)}. + */ +public class AdminFcmTopic { + private static final String TAG = AdminFcmTopic.class.getName(); + + /** + * The FCM topic name. + */ + public final String name; + + private final PubSubBase rest; + private final PushBase push; + + AdminFcmTopic(PubSubBase rest, PushBase push, String name) { + this.rest = rest; + this.push = push; + this.name = name; + } + + /** + * Subscribes a device to the topic. Subscribing a device that is already subscribed succeeds. + *

+ * The device must be registered with Ably for FCM. + * + * @param deviceId The unique ID of the device. + * @throws AblyException + */ + public void subscribe(String deviceId) throws AblyException { + AdminFcmTopics.subscribeDeviceImpl(rest, push.pushRequestHeaders(deviceId), name, deviceId).sync(); + } + + /** + * Asynchronously subscribes a device to the topic. See {@link #subscribe(String)}. + * + * @param deviceId The unique ID of the device. + * @param listener A listener to be notified of success or failure. + *

+ * This listener is invoked on a background thread. + */ + public void subscribeAsync(String deviceId, CompletionListener listener) { + AdminFcmTopics.subscribeDeviceImpl(rest, push.pushRequestHeaders(deviceId), name, deviceId) + .async(new CompletionListener.ToCallback<>(listener)); + } + + /** + * Unsubscribes a device from the topic. Unsubscribing a device that is not subscribed succeeds. + * + * @param deviceId The unique ID of the device. + * @throws AblyException + */ + public void unsubscribe(String deviceId) throws AblyException { + AdminFcmTopics.unsubscribeDeviceImpl(rest, push.pushRequestHeaders(deviceId), name, deviceId).sync(); + } + + /** + * Asynchronously unsubscribes a device from the topic. See {@link #unsubscribe(String)}. + * + * @param deviceId The unique ID of the device. + * @param listener A listener to be notified of success or failure. + *

+ * This listener is invoked on a background thread. + */ + public void unsubscribeAsync(String deviceId, CompletionListener listener) { + AdminFcmTopics.unsubscribeDeviceImpl(rest, push.pushRequestHeaders(deviceId), name, deviceId) + .async(new CompletionListener.ToCallback<>(listener)); + } + + /** + * Requests that every FCM device registered for a clientId is subscribed to the topic. + *

+ * Ably accepts the request and subscribes the devices in the background, so success means the + * request was accepted, not that every device is subscribed. Devices registered for the clientId + * later are not subscribed. + * + * @param clientId The ID of the client whose devices to subscribe. + * @throws AblyException + */ + public void subscribeClient(String clientId) throws AblyException { + subscribeClientImpl(clientId).sync(); + } + + /** + * Asynchronously requests that every FCM device registered for a clientId is subscribed to the topic. + * See {@link #subscribeClient(String)}. + * + * @param clientId The ID of the client whose devices to subscribe. + * @param listener A listener to be notified of success or failure. + *

+ * This listener is invoked on a background thread. + */ + public void subscribeClientAsync(String clientId, CompletionListener listener) { + subscribeClientImpl(clientId).async(new CompletionListener.ToCallback<>(listener)); + } + + /** + * Requests that every FCM device subscribed to push notifications on any of the given channels + * is subscribed to the topic. + *

+ * Ably accepts the request and subscribes the devices in the background, so success means the + * request was accepted, not that every device is subscribed. This is a one-off batch, not a + * standing link: devices that subscribe to the channels later are not subscribed to the topic. + * + * @param channels The names of the channels whose subscribed devices to subscribe. + * @throws AblyException + */ + public void subscribeChannels(String[] channels) throws AblyException { + subscribeChannelsImpl(channels).sync(); + } + + /** + * Asynchronously requests that every FCM device subscribed to push notifications on any of the + * given channels is subscribed to the topic. See {@link #subscribeChannels(String[])}. + * + * @param channels The names of the channels whose subscribed devices to subscribe. + * @param listener A listener to be notified of success or failure. + *

+ * This listener is invoked on a background thread. + */ + public void subscribeChannelsAsync(String[] channels, CompletionListener listener) { + subscribeChannelsImpl(channels).async(new CompletionListener.ToCallback<>(listener)); + } + + /** + * Requests that every FCM device registered for a clientId is unsubscribed from the topic. + *

+ * Ably accepts the request and unsubscribes the devices in the background, so success means the + * request was accepted, not that every device is unsubscribed. + * + * @param clientId The ID of the client whose devices to unsubscribe. + * @throws AblyException + */ + public void unsubscribeClient(String clientId) throws AblyException { + unsubscribeClientImpl(clientId).sync(); + } + + /** + * Asynchronously requests that every FCM device registered for a clientId is unsubscribed from the topic. + * See {@link #unsubscribeClient(String)}. + * + * @param clientId The ID of the client whose devices to unsubscribe. + * @param listener A listener to be notified of success or failure. + *

+ * This listener is invoked on a background thread. + */ + public void unsubscribeClientAsync(String clientId, CompletionListener listener) { + unsubscribeClientImpl(clientId).async(new CompletionListener.ToCallback<>(listener)); + } + + /** + * Sends a push notification to every device subscribed to the topic. + *

+ * The payload carries FCM message fields ({@code data}, {@code notification}, {@code android}), + * which are passed through to FCM; Ably sets the topic itself. {@code android.priority} defaults + * to {@code high}. FCM requires every value in {@code data} to be a string. + * + * @param payload A JSON object containing the FCM message fields. + * @return The FCM message ID. + * @throws AblyException + */ + public String publish(JsonObject payload) throws AblyException { + return publishImpl(payload).sync(); + } + + /** + * Asynchronously sends a push notification to every device subscribed to the topic. + * See {@link #publish(JsonObject)}. + * + * @param payload A JSON object containing the FCM message fields. + * @param callback A callback returning the FCM message ID. + *

+ * This callback is invoked on a background thread. + */ + public void publishAsync(JsonObject payload, Callback callback) { + publishImpl(payload).async(callback); + } + + private HttpRequester.Request subscribeClientImpl(String clientId) { + Log.v(TAG, "subscribeClientImpl(): topic=" + name + ", clientId=" + clientId); + AblyException invalid = AdminFcmTopics.emptyArgument("topic", name, "clientId", clientId); + if (invalid != null) { + return rest.http.failedRequest(invalid); + } + JsonObject bodyJson = new JsonObject(); + bodyJson.addProperty("clientId", clientId); + return postSubscriptions(bodyJson); + } + + private HttpRequester.Request subscribeChannelsImpl(String[] channels) { + Log.v(TAG, "subscribeChannelsImpl(): topic=" + name + ", channels=" + Arrays.toString(channels)); + AblyException invalid = AdminFcmTopics.emptyArgument("topic", name); + if (invalid != null) { + return rest.http.failedRequest(invalid); + } + if (channels == null || channels.length == 0) { + return rest.http.failedRequest(AdminFcmTopics.missingArgument("channels")); + } + JsonArray channelsJson = new JsonArray(); + for (String channel : channels) { + invalid = AdminFcmTopics.emptyArgument("channel name", channel); + if (invalid != null) { + return rest.http.failedRequest(invalid); + } + channelsJson.add(channel); + } + JsonObject bodyJson = new JsonObject(); + bodyJson.add("channels", channelsJson); + return postSubscriptions(bodyJson); + } + + private HttpRequester.Request postSubscriptions(JsonObject bodyJson) { + final HttpCore.RequestBody body = HttpUtils.requestBodyFromGson(bodyJson, rest.options.useBinaryProtocol); + return rest.http.request(new HttpRequester.Execute() { + @Override + public void execute(HttpScheduler http, Callback callback) throws AblyException { + Param[] params = ParamsUtils.enrichParams(null, rest.options); + http.post(AdminFcmTopics.topicPath(name) + "/subscriptions", HttpUtils.defaultAcceptHeaders(rest.options.useBinaryProtocol), params, body, null, true, callback); + } + }); + } + + private HttpRequester.Request unsubscribeClientImpl(final String clientId) { + Log.v(TAG, "unsubscribeClientImpl(): topic=" + name + ", clientId=" + clientId); + AblyException invalid = AdminFcmTopics.emptyArgument("topic", name, "clientId", clientId); + if (invalid != null) { + return rest.http.failedRequest(invalid); + } + return rest.http.request(new HttpRequester.Execute() { + @Override + public void execute(HttpScheduler http, Callback callback) throws AblyException { + Param[] params = ParamsUtils.enrichParams(new Param[] { new Param("clientId", clientId) }, rest.options); + http.del(AdminFcmTopics.topicPath(name) + "/subscriptions", HttpUtils.defaultAcceptHeaders(rest.options.useBinaryProtocol), params, null, true, callback); + } + }); + } + + private HttpRequester.Request publishImpl(JsonObject payload) { + Log.v(TAG, "publishImpl(): topic=" + name + ", payload=" + payload); + AblyException invalid = AdminFcmTopics.emptyArgument("topic", name); + if (invalid != null) { + return rest.http.failedRequest(invalid); + } + if (payload == null || payload.entrySet().isEmpty()) { + return rest.http.failedRequest(AdminFcmTopics.missingArgument("payload")); + } + final HttpCore.RequestBody body = HttpUtils.requestBodyFromGson(payload, rest.options.useBinaryProtocol); + return rest.http.request(new HttpRequester.Execute() { + @Override + public void execute(HttpScheduler http, Callback callback) throws AblyException { + Param[] params = ParamsUtils.enrichParams(null, rest.options); + http.post(AdminFcmTopics.topicPath(name) + "/messages", HttpUtils.defaultAcceptHeaders(rest.options.useBinaryProtocol), params, body, messageIdResponseHandler, true, callback); + } + }); + } + + /** Reads the FCM message ID from the publish response, {@code {"topic": ..., "messageId": ...}}. */ + private static final HttpCore.ResponseHandler messageIdResponseHandler = new HttpCore.ResponseHandler() { + @Override + public String handleResponse(HttpCore.Response response, ErrorInfo error) throws AblyException { + // The publish has succeeded once the server accepts it, so a missing or malformed + // message ID is reported as null rather than as a failure. + JsonElement body = AdminFcmTopics.responseJson(response, error); + JsonElement messageId = body != null && body.isJsonObject() ? body.getAsJsonObject().get("messageId") : null; + return messageId != null && messageId.isJsonPrimitive() ? messageId.getAsString() : null; + } + }; +} diff --git a/lib/src/main/java/io/ably/pubsub/push/AdminFcmTopics.java b/lib/src/main/java/io/ably/pubsub/push/AdminFcmTopics.java new file mode 100644 index 000000000..0c63a1e09 --- /dev/null +++ b/lib/src/main/java/io/ably/pubsub/push/AdminFcmTopics.java @@ -0,0 +1,217 @@ +package io.ably.pubsub.push; + +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.JsonParseException; +import io.ably.pubsub.http.HttpCore; +import io.ably.pubsub.http.HttpRequester; +import io.ably.pubsub.http.HttpScheduler; +import io.ably.pubsub.http.HttpUtils; +import io.ably.pubsub.http.PubSubBase; +import io.ably.pubsub.types.AblyException; +import io.ably.pubsub.types.Callback; +import io.ably.pubsub.types.ErrorInfo; +import io.ably.pubsub.types.Param; +import io.ably.pubsub.util.Log; +import io.ably.pubsub.util.ParamsUtils; +import io.ably.pubsub.util.Serialisation; +import io.ably.pubsub.util.StringUtils; + +/** + * Manages FCM topics with push-admin privileges: subscribing any device to a topic, + * listing a device's topics, and publishing to a topic. + *

+ * FCM topics let FCM fan a single send out to every Android device subscribed to the topic, + * instead of Ably sending one FCM request per device. Ably resolves devices to their FCM + * registration tokens and calls FCM with the app's service account, so callers never handle + * FCM credentials or tokens. + *

+ * A topic exists once its first device subscribes; there is no create or delete call. Topic names + * must match {@code [a-zA-Z0-9-_.~%]{1,900}}. FCM topics are open: any instance of the app can + * subscribe to any topic directly through the Firebase SDK, so never put user-specific or + * sensitive data in a topic message. + */ +public class AdminFcmTopics { + private static final String TAG = AdminFcmTopics.class.getName(); + + private final PubSubBase rest; + private final PushBase push; + + AdminFcmTopics(PubSubBase rest, PushBase push) { + this.rest = rest; + this.push = push; + } + + /** + * Returns an {@link AdminFcmTopic} for managing subscriptions to, and publishing to, the named topic. + * An empty topic name is reported as an {@link AblyException} by the operations on the returned object. + * + * @param topic The FCM topic name. + * @return An {@link AdminFcmTopic} object. + */ + public AdminFcmTopic get(String topic) { + return new AdminFcmTopic(rest, push, topic); + } + + /** + * Retrieves the names of the FCM topics a device is subscribed to, + * including any it subscribed to other than through Ably. + * + * @param deviceId The unique ID of the device. + * @return The topic names. + * @throws AblyException + */ + public String[] list(String deviceId) throws AblyException { + return listImpl(rest, push.pushRequestHeaders(deviceId), deviceId).sync(); + } + + /** + * Asynchronously retrieves the names of the FCM topics a device is subscribed to, + * including any it subscribed to other than through Ably. + * + * @param deviceId The unique ID of the device. + * @param callback A callback returning the topic names. + *

+ * This callback is invoked on a background thread. + */ + public void listAsync(String deviceId, Callback callback) { + listImpl(rest, push.pushRequestHeaders(deviceId), deviceId).async(callback); + } + + /** + * Sends a push notification to every device subscribed to an FCM topic. + *

+ * The payload carries FCM message fields ({@code data}, {@code notification}, {@code android}), + * which are passed through to FCM; Ably sets the topic itself. {@code android.priority} defaults + * to {@code high}. FCM requires every value in {@code data} to be a string. + * + * @param topic The FCM topic name. + * @param payload A JSON object containing the FCM message fields. + * @return The FCM message ID. + * @throws AblyException + */ + public String publish(String topic, JsonObject payload) throws AblyException { + return get(topic).publish(payload); + } + + /** + * Asynchronously sends a push notification to every device subscribed to an FCM topic. + * See {@link #publish(String, JsonObject)}. + * + * @param topic The FCM topic name. + * @param payload A JSON object containing the FCM message fields. + * @param callback A callback returning the FCM message ID. + *

+ * This callback is invoked on a background thread. + */ + public void publishAsync(String topic, JsonObject payload, Callback callback) { + get(topic).publishAsync(payload, callback); + } + + static String topicPath(String topic) { + return "/push/fcmTopics/" + HttpUtils.encodeURIComponent(topic); + } + + static String deviceSubscriptionPath(String topic, String deviceId) { + return topicPath(topic) + "/subscriptions/" + HttpUtils.encodeURIComponent(deviceId); + } + + static HttpRequester.Request subscribeDeviceImpl(final PubSubBase rest, final Param[] headers, final String topic, final String deviceId) { + Log.v(TAG, "subscribeDeviceImpl(): topic=" + topic + ", deviceId=" + deviceId); + AblyException invalid = emptyArgument("topic", topic, "deviceId", deviceId); + if (invalid != null) { + return rest.http.failedRequest(invalid); + } + // The endpoint reads no body, but HTTP clients such as OkHttp reject a POST without one. + final HttpCore.RequestBody body = HttpUtils.requestBodyFromGson(new JsonObject(), rest.options.useBinaryProtocol); + return rest.http.request(new HttpRequester.Execute() { + @Override + public void execute(HttpScheduler http, Callback callback) throws AblyException { + Param[] params = ParamsUtils.enrichParams(null, rest.options); + http.post(deviceSubscriptionPath(topic, deviceId), headers, params, body, null, true, callback); + } + }); + } + + static HttpRequester.Request unsubscribeDeviceImpl(final PubSubBase rest, final Param[] headers, final String topic, final String deviceId) { + Log.v(TAG, "unsubscribeDeviceImpl(): topic=" + topic + ", deviceId=" + deviceId); + AblyException invalid = emptyArgument("topic", topic, "deviceId", deviceId); + if (invalid != null) { + return rest.http.failedRequest(invalid); + } + return rest.http.request(new HttpRequester.Execute() { + @Override + public void execute(HttpScheduler http, Callback callback) throws AblyException { + Param[] params = ParamsUtils.enrichParams(null, rest.options); + http.del(deviceSubscriptionPath(topic, deviceId), headers, params, null, true, callback); + } + }); + } + + static HttpRequester.Request listImpl(final PubSubBase rest, final Param[] headers, final String deviceId) { + Log.v(TAG, "listImpl(): deviceId=" + deviceId); + AblyException invalid = emptyArgument("deviceId", deviceId); + if (invalid != null) { + return rest.http.failedRequest(invalid); + } + return rest.http.request(new HttpRequester.Execute() { + @Override + public void execute(HttpScheduler http, Callback callback) throws AblyException { + Param[] params = ParamsUtils.enrichParams(new Param[] { new Param("deviceId", deviceId) }, rest.options); + http.get("/push/fcmTopics", headers, params, topicsResponseHandler, true, callback); + } + }); + } + + /** + * Checks arguments given as alternating names and values. + * + * @return An error naming the first null or empty argument, or null if there is none. + */ + static AblyException emptyArgument(String... namesAndValues) { + for (int i = 0; i < namesAndValues.length; i += 2) { + if (StringUtils.isNullOrEmpty(namesAndValues[i + 1])) { + return missingArgument(namesAndValues[i]); + } + } + return null; + } + + static AblyException missingArgument(String name) { + return AblyException.fromErrorInfo(new ErrorInfo(name + " cannot be empty", 400, 40000)); + } + + /** + * Decodes a response body as JSON, throwing the request's error if it failed. + * + * @return The decoded body, or null if the response has no body. + */ + static JsonElement responseJson(HttpCore.Response response, ErrorInfo error) throws AblyException { + if (error != null) { + throw AblyException.fromErrorInfo(error); + } + if (response == null || response.body == null || response.body.length == 0) { + return null; + } + return Serialisation.httpResponseHandler.handleResponse(response, null); + } + + /** Reads the list response, a plain (unpaginated) array of topic names. */ + private static final HttpCore.ResponseHandler topicsResponseHandler = new HttpCore.ResponseHandler() { + @Override + public String[] handleResponse(HttpCore.Response response, ErrorInfo error) throws AblyException { + JsonElement body = responseJson(response, error); + if (body == null || body.isJsonNull()) { + return new String[0]; + } + try { + if (!body.isJsonArray()) { + throw new JsonParseException("expected an array of topic names, got " + body); + } + return Serialisation.gson.fromJson(body, String[].class); + } catch (JsonParseException e) { + throw AblyException.fromErrorInfo(e, new ErrorInfo("Unexpected response listing FCM topics: " + e.getMessage(), 500, 50000)); + } + } + }; +} diff --git a/lib/src/main/java/io/ably/pubsub/push/PushBase.java b/lib/src/main/java/io/ably/pubsub/push/PushBase.java index 421959284..236bec7c3 100644 --- a/lib/src/main/java/io/ably/pubsub/push/PushBase.java +++ b/lib/src/main/java/io/ably/pubsub/push/PushBase.java @@ -29,7 +29,7 @@ public class PushBase { public PushBase(PubSubBase rest) { this.rest = rest; - this.admin = new Admin(rest); + this.admin = new Admin(rest, this); } public static class Admin { @@ -47,11 +47,16 @@ public static class Admin { * Spec: RSH1c */ public final ChannelSubscriptions channelSubscriptions; + /** + * An {@link AdminFcmTopics} object, for managing and publishing to FCM topics. + */ + public final AdminFcmTopics fcmTopics; - Admin(PubSubBase rest) { + Admin(PubSubBase rest, PushBase push) { this.rest = rest; this.deviceRegistrations = new DeviceRegistrations(rest); this.channelSubscriptions = new ChannelSubscriptions(rest); + this.fcmTopics = new AdminFcmTopics(rest, push); } /** diff --git a/lib/src/test/kotlin/io/ably/pubsub/push/AdminFcmTopicsTest.kt b/lib/src/test/kotlin/io/ably/pubsub/push/AdminFcmTopicsTest.kt new file mode 100644 index 000000000..ddfcb33d5 --- /dev/null +++ b/lib/src/test/kotlin/io/ably/pubsub/push/AdminFcmTopicsTest.kt @@ -0,0 +1,176 @@ +package io.ably.pubsub.push + +import com.google.gson.JsonObject +import com.google.gson.JsonParser +import io.ably.pubsub.http.PubSubHttpClient +import io.ably.pubsub.types.AblyException +import io.ably.pubsub.uts.infra.unit.MockHttpClient +import io.ably.pubsub.uts.infra.unit.PendingRequest +import io.ably.pubsub.uts.infra.unit.TestHttpClient +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Assert.assertThrows +import org.junit.Assert.assertTrue +import org.junit.Test +import java.util.concurrent.CopyOnWriteArrayList + +/** + * Checks the requests the push-admin FCM topics API sends, against a mock HTTP engine. + */ +class AdminFcmTopicsTest { + + private val requests = CopyOnWriteArrayList() + + /** A client whose every request is recorded and answered with [status] and [body]. */ + private fun client(status: Int = 200, body: Any = emptyMap()): PubSubHttpClient { + val mock = MockHttpClient { + onConnectionAttempt = { it.respondWithSuccess() } + onRequest = { request -> + requests.add(request) + request.respondWith(status, body) + } + } + return TestHttpClient { install(mock) } + } + + private val request get() = requests.single() + + private fun bodyJson(): JsonObject = JsonParser.parseString(String(request.body)).asJsonObject + + @Test + fun `subscribe posts to the device subscription path`() { + client(body = mapOf("topic" to "team-manutd", "deviceId" to "device-1")) + .push.admin.fcmTopics.get("team-manutd").subscribe("device-1") + + assertEquals("POST", request.method) + assertEquals("/push/fcmTopics/team-manutd/subscriptions/device-1", request.url.path) + } + + @Test + fun `unsubscribe deletes the device subscription path`() { + client(status = 204, body = "").push.admin.fcmTopics.get("team-manutd").unsubscribe("device-1") + + assertEquals("DELETE", request.method) + assertEquals("/push/fcmTopics/team-manutd/subscriptions/device-1", request.url.path) + } + + @Test + fun `topic and device id are percent-encoded in the path`() { + // FCM topic names may contain %, which the server requires sent as %25. + client().push.admin.fcmTopics.get("50%off").subscribe("device/1") + + // java.net.URL.path is the raw, still-encoded path. + assertEquals("/push/fcmTopics/50%25off/subscriptions/device%2F1", request.url.path) + } + + @Test + fun `subscribeClient posts the clientId to the topic subscriptions`() { + client(status = 202, body = mapOf("topic" to "team-manutd")) + .push.admin.fcmTopics.get("team-manutd").subscribeClient("client-1") + + assertEquals("POST", request.method) + assertEquals("/push/fcmTopics/team-manutd/subscriptions", request.url.path) + assertEquals("client-1", bodyJson().get("clientId").asString) + assertTrue(!bodyJson().has("channels")) + } + + @Test + fun `subscribeChannels posts the channels to the topic subscriptions`() { + client(status = 202, body = mapOf("topic" to "team-manutd")) + .push.admin.fcmTopics.get("team-manutd").subscribeChannels(arrayOf("a", "b")) + + assertEquals("/push/fcmTopics/team-manutd/subscriptions", request.url.path) + val channels = bodyJson().getAsJsonArray("channels").map { it.asString } + assertEquals(listOf("a", "b"), channels) + } + + @Test + fun `unsubscribeClient deletes the topic subscriptions with a clientId param`() { + client(status = 202, body = mapOf("topic" to "team-manutd")) + .push.admin.fcmTopics.get("team-manutd").unsubscribeClient("client-1") + + assertEquals("DELETE", request.method) + assertEquals("/push/fcmTopics/team-manutd/subscriptions", request.url.path) + assertTrue(request.url.query.split("&").contains("clientId=client-1")) + } + + @Test + fun `list gets the device's topics`() { + val topics = client(body = listOf("team-manutd", "alerts-goals")).push.admin.fcmTopics.list("device-1") + + assertEquals("GET", request.method) + assertEquals("/push/fcmTopics", request.url.path) + assertTrue(request.url.query.split("&").contains("deviceId=device-1")) + assertArrayEquals(arrayOf("team-manutd", "alerts-goals"), topics) + } + + @Test + fun `publish posts the payload to the topic messages and returns the message id`() { + val payload = JsonObject().apply { + add("data", JsonObject().apply { addProperty("homeScore", "2") }) + add("android", JsonObject().apply { addProperty("ttl", "120s") }) + } + + val messageId = client(body = mapOf("topic" to "team-manutd", "messageId" to "projects/p/messages/1")) + .push.admin.fcmTopics.publish("team-manutd", payload) + + assertEquals("POST", request.method) + assertEquals("/push/fcmTopics/team-manutd/messages", request.url.path) + assertEquals(payload, bodyJson()) + assertEquals("projects/p/messages/1", messageId) + } + + @Test + fun `publish returns null when the response has no message id`() { + assertNull(client().push.admin.fcmTopics.get("team-manutd").publish(JsonObject().apply { + add("data", JsonObject()) + })) + } + + @Test + fun `publish returns null when the response has no body`() { + assertNull(client(status = 204, body = "").push.admin.fcmTopics.publish("team-manutd", JsonObject().apply { + add("data", JsonObject()) + })) + } + + @Test + fun `list with an unexpected response shape raises AblyException`() { + val e = assertThrows(AblyException::class.java) { + client(body = mapOf("items" to listOf("team-manutd"))).push.admin.fcmTopics.list("device-1") + } + assertEquals(50000, e.errorInfo.code) + } + + @Test + fun `list returns an empty array when the response has no body`() { + assertArrayEquals(arrayOf(), client(status = 204, body = "").push.admin.fcmTopics.list("device-1")) + } + + @Test + fun `server errors are raised as AblyException`() { + val error = mapOf("error" to mapOf("code" to 40183, "statusCode" to 400, "message" to "needs a service account")) + val e = assertThrows(AblyException::class.java) { + client(status = 400, body = error).push.admin.fcmTopics.get("team-manutd").subscribe("device-1") + } + assertEquals(40183, e.errorInfo.code) + } + + @Test + fun `empty arguments fail without a request`() { + val fcmTopics = client().push.admin.fcmTopics + for (call in listOf<() -> Unit>( + { fcmTopics.get("").subscribe("device-1") }, + { fcmTopics.get(null).unsubscribeClient("client-1") }, + { fcmTopics.publish("", JsonObject().apply { add("data", JsonObject()) }) }, + )) { + assertEquals(40000, assertThrows(AblyException::class.java) { call() }.errorInfo.code) + } + assertThrows(AblyException::class.java) { fcmTopics.get("t").subscribe("") } + assertThrows(AblyException::class.java) { fcmTopics.get("t").subscribeChannels(arrayOf()) } + assertThrows(AblyException::class.java) { fcmTopics.publish("t", JsonObject()) } + assertThrows(AblyException::class.java) { fcmTopics.list(null) } + assertTrue(requests.isEmpty()) + } +}