Refactor pubsub

Author: tolyo

@types/services/pubsub/pubsub.d.ts

@@ -23,144 +23,66 @@ export class PubSubProvider {
  *  - Minimal memory churn & stable hidden-class shapes
  *  - Fast publish using flat arrays
  *  - Preserves listener order
+ *  - All publishes are asynchronous (via queueMicrotask)
  */
 export class PubSub {
-  static $nonscope: boolean;
-  disposed: boolean;
+  /** @private {Object<string, Array<{fn: Function, context: any}>>} */
+  private _topics;
+  /** @private */
+  private _disposed;
   /**
-   * The next available subscription key.  Internally, this is an index into the
-   * sparse array of subscriptions.
-   *
-   * @private {number}
+   * Set instance to initial state
    */
-  private key;
+  reset(): void;
   /**
-   * Array of subscription keys pending removal once publishing is done.
-   *
-   * @private {!Array<number>}
-   * @const
+   * Checks if instance has been disposed.
+   * @returns {boolean} True if disposed.
    */
-  private pendingKeys;
-  /**
-   * Lock to prevent the removal of subscriptions during publishing. Incremented
-   * at the beginning of {@link #publish}, and decremented at the end.
-   *
-   * @private {number}
-   */
-  private publishDepth;
-  /**
-   * Sparse array of subscriptions. Each subscription is represented by a tuple
-   * comprising a topic identifier, a function, and an optional context object.
-   * Each tuple occupies three consecutive positions in the array, with the
-   * topic identifier at index n, the function at index (n + 1), the context
-   * object at index (n + 2), the next topic at index (n + 3), etc. (This
-   * representation minimizes the number of object allocations and has been
-   * shown to be faster than an array of objects with three key-value pairs or
-   * three parallel arrays, especially on IE.)
-   *
-   * Once a subscription is removed via {@link unsubscribe} or {@link unsubscribeByKey}, the three
-   * corresponding array elements are deleted, and never reused. This means the
-   * total number of subscriptions during the lifetime of the pubsub channel is
-   * limited by the maximum length of a JavaScript array to (2^32 - 1) / 3 =
-   * 1,431,655,765 subscriptions, which should suffice for most applications.
-   *
-   * @private {!Array<?>}
-   * @const
-   */
-  private subscriptions;
-  /**
-   * Map of topics to arrays of subscription keys.
-   *
-   * @private {!Object<!Array<number>>}
-   */
-  private topics;
-  /**
-   * Subscribes a function to a topic.  The function is invoked as a method on
-   * the given `opt_context` object, or in the global scope if no context
-   * is specified.  Subscribing the same function to the same topic multiple
-   * times will result in multiple function invocations while publishing.
-   * Returns a subscription key that can be used to unsubscribe the function from
-   * the topic via {@link unsubscribeByKey}.
-   *
-   * @param {string} topic Topic to subscribe to.
-   * @param {Function} fn Function to be invoked when a message is published to
-   *     the given topic.
-   * @param {Object=} opt_context Object in whose context the function is to be
-   *     called (the global scope if none).
-   * @return {number} Subscription key.
-   */
-  subscribe(topic: string, fn: Function, opt_context?: any | undefined): number;
+  isDisposed(): boolean;
   /**
-   * Subscribes a single-use function to a topic.  The function is invoked as a
-   * method on the given `opt_context` object, or in the global scope if
-   * no context is specified, and is then unsubscribed.  Returns a subscription
-   * key that can be used to unsubscribe the function from the topic via
-   * {@link unsubscribeByKey}.
-   *
-   * @param {string} topic Topic to subscribe to.
-   * @param {Function} fn Function to be invoked once and then unsubscribed when
-   *     a message is published to the given topic.
-   * @param {Object=} opt_context Object in whose context the function is to be
-   *     called (the global scope if none).
-   * @return {number} Subscription key.
+   * Dispose the instance, removing all topics and listeners.
    */
-  subscribeOnce(
-    topic: string,
-    fn: Function,
-    opt_context?: any | undefined,
-  ): number;
+  dispose(): void;
   /**
-   * Unsubscribes a function from a topic.  Only deletes the first match found.
-   * Returns a Boolean indicating whether a subscription was removed.
-   *
-   * @param {string} topic Topic to unsubscribe from.
-   * @param {Function} fn Function to unsubscribe.
-   * @param {Object=} opt_context Object in whose context the function was to be
-   *     called (the global scope if none).
-   * @return {boolean} Whether a matching subscription was removed.
+   * Subscribe a function to a topic.
+   * @param {string} topic - The topic to subscribe to.
+   * @param {Function} fn - The callback function to invoke when published.
+   * @param {*} [context] - Optional `this` context for the callback.
+   * @returns {() => boolean} A function that unsubscribes this listener.
    */
-  unsubscribe(
-    topic: string,
-    fn: Function,
-    opt_context?: any | undefined,
-  ): boolean;
+  subscribe(topic: string, fn: Function, context?: any): () => boolean;
   /**
-   * Removes a subscription based on the key returned by {@link subscribe}.
-   * No-op if no matching subscription is found.  Returns a Boolean indicating
-   * whether a subscription was removed.
-   *
-   * @param {number} key Subscription key.
-   * @return {boolean} Whether a matching subscription was removed.
+   * Subscribe a function to a topic only once.
+   * Listener is removed before the first invocation.
+   * @param {string} topic - The topic to subscribe to.
+   * @param {Function} fn - The callback function.
+   * @param {*} [context] - Optional `this` context for the callback.
+   * @returns {() => boolean} A function that unsubscribes this listener.
    */
-  unsubscribeByKey(key: number): boolean;
+  subscribeOnce(topic: string, fn: Function, context?: any): () => boolean;
   /**
-   * Publishes a message to a topic.  Calls functions subscribed to the topic in
-   * the order in which they were added, passing all arguments along.
-   *
-   * If this object was created with async=true, subscribed functions are called
-   * via `queueMicrotask`.  Otherwise, the functions are called directly, and if
-   * any of them throw an uncaught error, publishing is aborted.
-   *
-   * @param {string} topic Topic to publish to.
-   * @param {...*} var_args Arguments that are applied to each subscription
-   *     function.
-   * @return {boolean} Whether any subscriptions were called.
+   * Unsubscribe a specific function from a topic.
+   * Matches by function reference and optional context.
+   * @param {string} topic - The topic to unsubscribe from.
+   * @param {Function} fn - The listener function.
+   * @param {*} [context] - Optional `this` context.
+   * @returns {boolean} True if the listener was found and removed.
    */
-  publish(topic: string, ...var_args: any[]): boolean;
+  unsubscribe(topic: string, fn: Function, context?: any): boolean;
   /**
-   * Clears the subscription list for a topic, or all topics if unspecified.
-   * @param {string=} opt_topic Topic to clear (all topics if unspecified).
+   * Get the number of subscribers for a topic.
+   * @param {string} topic
+   * @returns {number}
    */
-  clear(opt_topic?: string | undefined): void;
+  getCount(topic: string): number;
   /**
-   * Returns the number of subscriptions to the given topic (or all topics if
-   * unspecified). This number will not change while publishing any messages.
-   * @param {string=} opt_topic The topic (all topics if unspecified).
-   * @return {number} Number of subscriptions to the topic.
+   * Publish a value to a topic asynchronously.
+   * All listeners are invoked in the order they were added.
+   * @param {string} topic - The topic to publish.
+   * @param {...*} args - Arguments to pass to listeners.
+   * @returns {boolean} True if any listeners exist for this topic.
    */
-  getCount(opt_topic?: string | undefined): number;
-  isDisposed(): boolean;
-  dispose(): void;
+  publish(topic: string, ...args: any[]): boolean;
 }
 export const EventBus: PubSub;
 /**

src/directive/channel/channel.js

@@ -13,7 +13,7 @@ export function ngChannelDirective($eventBus) {
 
       const hasTemplateContent = element.childNodes.length > 0;
 
-      const key = $eventBus.subscribe(channel, (value) => {
+      const unsubscribe = $eventBus.subscribe(channel, (value) => {
         if (hasTemplateContent) {
           if (isObject(value)) {
             scope.$merge(value);
@@ -23,9 +23,7 @@ export function ngChannelDirective($eventBus) {
         }
       });
 
-      scope.$on("$destroy", () => {
-        $eventBus.unsubscribeByKey(key);
-      });
+      scope.$on("$destroy", () => unsubscribe());
     },
   };
 }

src/directive/channel/channel.spec.js

@@ -18,7 +18,6 @@ describe("channel", () => {
       });
 
     spyOn(EventBus, "subscribe").and.callThrough();
-    unsubscribeSpy = spyOn(EventBus, "unsubscribeByKey").and.callThrough();
   });
 
   it("should subscribe to the specified EventBus channel", () => {
@@ -42,11 +41,12 @@ describe("channel", () => {
   });
 
   it("should unsubscribe from the EventBus when the scope is destroyed", () => {
+    EventBus.reset();
     element = $compile('<div ng-channel="testChannel"></div>')($scope);
-
+    expect(EventBus.getCount("testChannel")).toEqual(1);
     $scope.$destroy();
 
-    expect(unsubscribeSpy).toHaveBeenCalled();
+    expect(EventBus.getCount("testChannel")).toEqual(0);
   });
 
   it("should handle templates when EventBus emits a value", async () => {