[1.x] Improve subscription cancellation

client-js/main/client/client.js

@@ -183,6 +183,16 @@ export class Client {
     throw new Error('Not implemented in abstract base.');
   }
 
+  /**
+   * Immediately cancels all active subscriptions.
+   *
+   * This endpoint is handy to use when an end-user chooses to end her session
+   * with the web app. E.g. all subscriptions should be cancelled upon user sign-out.
+   */
+  cancelAllSubscriptions() {
+    throw new Error('Not implemented in abstract base.');
+  }
+
   /**
    * Subscribes to the given `Topic` instance.
    *

client-js/main/client/composite-client.js

@@ -90,6 +90,13 @@ export class CompositeClient extends Client {
     return this._subscribing.subscribeTo(entityType, this);
   }
 
+  /**
+   * @override
+   */
+  cancelAllSubscriptions() {
+    this._subscribing.cancelAllSubscriptions();
+  }
+
   /**
    * @override
    */

client-js/main/client/firebase-client.js

@@ -35,7 +35,6 @@ import {ActorRequestFactory} from './actor-request-factory';
 import {AbstractClientFactory} from './client-factory';
 import {CommandingClient} from "./commanding-client";
 import {CompositeClient} from "./composite-client";
-import {HttpClient} from './http-client';
 import {HttpEndpoint} from './http-endpoint';
 import {FirebaseDatabaseClient} from './firebase-database-client';
 import {FirebaseSubscriptionService} from './firebase-subscription-service';
@@ -84,9 +83,9 @@ class SpineSubscription extends Subscription {
 class EntitySubscription extends SpineSubscription {
 
   /**
-   * @param {Function} unsubscribe the callbacks that allows to cancel the subscription
+   * @param {Function} unsubscribe the callback that allows to cancel the subscription
    * @param {{itemAdded: Observable, itemChanged: Observable, itemRemoved: Observable}} observables
-   *        the observables for entity changes
+   *        the observables for entity change
    * @param {SubscriptionObject} subscription the wrapped subscription object
    */
   constructor({
@@ -102,7 +101,13 @@ class EntitySubscription extends SpineSubscription {
    * @return {EntitySubscriptionObject} a plain object with observables and unsubscribe method
    */
   toObject() {
-    return Object.assign({}, this._observables, {unsubscribe: () => this.unsubscribe()});
+    return Object.assign({},
+        this._observables,
+        {
+          unsubscribe: () => {
+            return this.unsubscribe();
+          }
+        });
   }
 }
 
@@ -130,7 +135,13 @@ class EventSubscription extends SpineSubscription {
    */
   toObject() {
     return Object.assign(
-        {}, {eventEmitted: this._observable}, {unsubscribe: () => this.unsubscribe()}
+        {},
+        {eventEmitted: this._observable},
+        {
+          unsubscribe: () => {
+            return this.unsubscribe();
+          }
+        }
     );
   }
 }
@@ -243,6 +254,7 @@ class FirebaseSubscribingClient extends SubscribingClient {
     return new EntitySubscription({
       unsubscribedBy: () => {
         FirebaseSubscribingClient._unsubscribe(pathSubscriptions);
+        this._subscriptionService.cancelSubscription(subscription);
       },
       withObservables: {
         itemAdded: ObjectToProto.map(itemAdded.asObservable(), typeUrl),
@@ -263,12 +275,20 @@ class FirebaseSubscribingClient extends SubscribingClient {
     return new EventSubscription({
       unsubscribedBy: () => {
         FirebaseSubscribingClient._unsubscribe([pathSubscription]);
+        this._subscriptionService.cancelSubscription(subscription);
       },
       withObservable: ObjectToProto.map(itemAdded.asObservable(), EVENT_TYPE_URL),
       forInternal: subscription
     });
   }
 
+  /**
+   * @override
+   */
+  cancelAllSubscriptions() {
+    this._subscriptionService.cancelAllSubscriptions();
+  }
+
   /**
    * Unsubscribes the provided Firebase subscriptions.
    *

client-js/main/client/firebase-subscription-service.js

@@ -38,8 +38,8 @@ import {Status} from '../proto/spine/core/response_pb';
 const DEFAULT_KEEP_UP_INTERVAL = new Duration({minutes: 2});
 
 /**
- * A service that manages the active subscriptions periodically sending requests to keep them
- * running.
+ * A service that manages the active subscriptions periodically sending requests
+ * to keep them running.
  */
 export class FirebaseSubscriptionService {
 
@@ -84,6 +84,33 @@ export class FirebaseSubscriptionService {
     }
   }
 
+  /**
+   * Immediately cancels all active subscriptions previously created through this service.
+   */
+  cancelAllSubscriptions() {
+    const activeSubscriptions = this._subscriptions.filter(s => !s.closed);
+    if (activeSubscriptions.length > 0) {
+      const subscriptionMessages = activeSubscriptions.map(s => s.internal())
+      this._endpoint.cancelAll(subscriptionMessages);
+      activeSubscriptions.forEach(s => {
+        s.unsubscribe() /* Calling RxJS's `unsubscribe` to stop propagating the updates. */
+        this._removeSubscription(s)
+      })
+    }
+  }
+
+  /**
+   * Immediately cancels the given subscription, including cancelling it on the server-side.
+   *
+   * @param {SubscriptionObject} subscription the subscription to cancel
+   */
+  cancelSubscription(subscription) {
+    this._endpoint.cancelSubscription(subscription)
+        .then(() => {
+          this._removeSubscription(subscription)
+        });
+  }
+
   /**
    * Indicates whether this service is running keeping up subscriptions.
    *
@@ -150,10 +177,22 @@ export class FirebaseSubscriptionService {
    * Removes the provided subscription from subscriptions list, which stops any attempts
    * to update it. In case no more subscriptions are left, stops this service.
    *
+   * In case the passed subscription is not known to this service, does nothing.
+   *
+   * @param subscription a subscription to cancel;
+   *                     this method accepts values of both `SpineSubscription`
+   *                     and Proto `Subscription` types,
+   *                     and operates based on the subscription ID
    * @private
    */
   _removeSubscription(subscription) {
-    const index = this._subscriptions.indexOf(subscription);
+    let id;
+    if (typeof subscription.id === 'function') {
+      id = subscription.id();
+    } else {
+      id = subscription.getId().getValue();
+    }
+    const index = this._subscriptions.findIndex(item => item.id() === id);
     this._subscriptions.splice(index, 1);
 
     if (this._subscriptions.length === 0) {

client-js/main/client/subscribing-client.js

@@ -85,6 +85,13 @@ export class SubscribingClient {
     throw new Error('Not implemented in abstract base.');
   }
 
+  /**
+   * Cancels all subscriptions, which were created through this instance of subscribing client.
+   */
+  cancelAllSubscriptions() {
+    throw new Error('Not implemented in abstract base.');
+  }
+
   /**
    * Returns a new topic factory instance which can be further used for the `Topic` creation.
    *
@@ -123,4 +130,13 @@ export class NoOpSubscribingClient extends SubscribingClient {
   subscribeToEvents(topic) {
     throw new Error(SUBSCRIPTIONS_NOT_SUPPORTED);
   }
+
+  /**
+   * Always throws an error.
+   *
+   * @override
+   */
+  cancelAllSubscriptions() {
+    throw new Error(SUBSCRIPTIONS_NOT_SUPPORTED);
+  }
 }

client-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "spine-web",
-  "version": "1.9.0-SNAPSHOT.9",
+  "version": "1.9.0-SNAPSHOT.10",
   "license": "Apache-2.0",
   "description": "A JS client for interacting with Spine applications.",
   "homepage": "https://spine.io",

firebase-web/src/main/java/io/spine/web/firebase/FirebaseCredentials.java

@@ -50,8 +50,7 @@
  *
  * <p>See <a href="https://firebase.google.com/docs/database/rest/auth">Firebase REST docs</a>.
  */
-@SuppressWarnings("deprecation")
-// Use deprecated `GoogleCredential` to retain backward compatibility.
+@SuppressWarnings("deprecation" /*`GoogleCredential` is used to retain backward compatibility.*/)
 public final class FirebaseCredentials implements HttpRequestInitializer {
 
     private static final String AUTH_DATABASE = "https://www.googleapis.com/auth/firebase.database";

firebase-web/src/main/java/io/spine/web/firebase/subscription/FirebaseSubscriptionBridge.java

@@ -209,13 +209,18 @@ public Builder setSubscriptionLifeSpan(Duration subscriptionLifeSpan) {
         /**
          * Creates a new instance of {@code FirebaseQueryBridge}.
          *
+         * <p>Mandatory fields are {@link #setFirebaseClient(FirebaseClient) firebaseClient}
+         * and {@link #setSubscriptionService(SubscriptionServiceImplBase) subscriptionService}.
+         *
          * @return new instance of {@code FirebaseQueryBridge}
          */
         public FirebaseSubscriptionBridge build() {
             checkState(firebaseClient != null,
-                       "Firebase database client is not set to FirebaseSubscriptionBridge.");
+                       "Mandatory Firebase database client" +
+                               " is not specified for `FirebaseSubscriptionBridge`.");
             checkState(subscriptionService != null,
-                       "Subscription Service is not set to FirebaseSubscriptionBridge.");
+                       "Mandatory Subscription Service is not specified" +
+                               " for `FirebaseSubscriptionBridge`.");
             return new FirebaseSubscriptionBridge(this);
         }
     }

firebase-web/src/main/java/io/spine/web/firebase/subscription/HealthLog.java

@@ -47,7 +47,7 @@
  * <p>To understand whether a client is still listening to the {@code Topic} updates, she
  * periodically sends a {@link FirebaseSubscriptionBridge#keepUp(Subscription) keepUp(Subscription)}
  * request. The server records the timestamps of these requests in this log and counts the client
- * alive, as long as the {@linkplain #withTimeout(Duration)} configured} timeout does not pass
+ * alive, as long as the {@linkplain #withTimeout(Duration) configured} timeout does not pass
  * since the last request.
  */
 final class HealthLog {
@@ -113,4 +113,15 @@ boolean isStale(Topic topic) {
         Duration elapsed = between(lastUpdate, now);
         return compare(elapsed, expirationTimeout) > 0;
     }
+
+    /**
+     * Removes the given {@code Topic} from this health log.
+     *
+     * <p>In case this topic is not known to this registry, does nothing, allowing
+     * to safely clear the health log from stale topics potentially residing in storage
+     * on either client- or server-sides.
+     */
+    void remove(Topic topic) {
+        updateTimes.remove(topic.getId());
+    }
 }

firebase-web/src/main/java/io/spine/web/firebase/subscription/SubscriptionRepository.java

@@ -34,7 +34,6 @@
 import com.google.protobuf.Duration;
 import io.spine.client.Subscription;
 import io.spine.client.Topic;
-import io.spine.json.Json;
 import io.spine.web.firebase.FirebaseClient;
 import io.spine.web.firebase.NodePath;
 import io.spine.web.firebase.NodePaths;
@@ -44,6 +43,7 @@
 import java.util.Optional;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static io.spine.json.Json.fromJson;
 import static io.spine.util.Exceptions.illegalStateWithCauseOf;
 import static io.spine.web.firebase.subscription.LazyRepository.lazy;
 
@@ -81,7 +81,7 @@ final class SubscriptionRepository {
     /**
      * Fetches all the existing subscriptions from the Firebase and activates them.
      *
-     * <p>After calling this method, all the new subscriptions are automatically activates on this
+     * <p>After calling this method, all the new subscriptions are automatically activated on this
      * server instance.
      */
     void subscribeToAll() {
@@ -152,6 +152,7 @@ private void delete(Topic topic) {
         checkNotNull(topic);
         NodePath path = pathForSubscription(topic);
         firebase.delete(path);
+        healthLog.remove(topic);
     }
 
     private static NodePath pathForSubscription(Topic topic) {
@@ -200,7 +201,7 @@ private static String asJson(DataSnapshot snapshot) {
         }
 
         private Topic loadTopic(String json) {
-            Topic topic = Json.fromJson(json, Topic.class);
+            Topic topic = fromJson(json, Topic.class);
             repository.healthLog.put(topic);
             return topic;
         }
@@ -216,7 +217,31 @@ private void deleteOrActivate(Topic topic) {
 
         @Override
         public void onChildRemoved(DataSnapshot snapshot) {
-            // NOP.
+            String json = asJson(snapshot);
+            Optional<Topic> topic = parseTopic(json);
+            topic.ifPresent(t -> {
+                HealthLog healthLog = repository.healthLog;
+                if (healthLog.isKnown(t)) {
+                    repository.delete(t);
+                }
+            });
+        }
+
+        /**
+         * Safely parses the {@code Topic} from the passed JSON.
+         *
+         * @param json
+         *         JSON to parse
+         * @return parsed {@code Topic} wrapped as {@code Optional},
+         *         or {@code Optional.empty()} if there was a parsing error
+         */
+        private static Optional<Topic> parseTopic(String json) {
+            try {
+                Topic topic = fromJson(json, Topic.class);
+                return Optional.of(topic);
+            } catch (RuntimeException e) {
+                return Optional.empty();
+            }
         }
 
         @Override

integration-tests/js-tests/package.json

@@ -1,6 +1,6 @@
 {
   "name": "client-js-tests",
-  "version": "1.9.0-SNAPSHOT.9",
+  "version": "1.9.0-SNAPSHOT.10",
   "license": "Apache-2.0",
   "description": "Tests of a `spine-web` JS library against the Spine-based application.",
   "scripts": {