Default eventbus to async

Author: tolyo

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

@@ -15,7 +15,7 @@ export class PubSubProvider {
   $get: () => PubSub;
 }
 export class PubSub {
-  static $nonscope: any;
+  static $nonscope: boolean;
   /**
    * Runs a function asynchronously.
    *
@@ -24,21 +24,7 @@ export class PubSub {
    * @param {Object} context Context in which to run the function.
    * @param {Array} args Arguments to pass to the function.
    */
-  private static runAsync_;
-  /**
-   * Topic-based publish/subscribe channel.  Maintains a map of topics to
-   * subscriptions.  When a message is published to a topic, all functions
-   * subscribed to that topic are invoked in the order they were added.
-   * Uncaught errors abort publishing.
-   *
-   * Topics may be identified by any nonempty string, <strong>except</strong>
-   * strings corresponding to native Object properties, e.g. "constructor",
-   * "toString", "hasOwnProperty", etc.
-   *
-   * @param {boolean=} async Enable asynchronous behavior.  Recommended for
-   *     new code.  See notes on the publish() method.
-   */
-  constructor(async?: boolean | undefined);
+  private static runAsync;
   disposed: boolean;
   /**
    * The next available subscription key.  Internally, this is an index into the
@@ -69,8 +55,9 @@ export class PubSub {
    * 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
+   * 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 =
@@ -86,10 +73,6 @@ export class PubSub {
    * @private {!Object<!Array<number>>}
    */
   private topics;
-  /**
-   * @private @const {boolean}
-   */
-  private async_;
   /**
    * 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

src/services/pubsub/pubsub.js

@@ -19,7 +19,7 @@ export class PubSubProvider {
 }
 
 export class PubSub {
-  static $nonscope;
+  static $nonscope = true;
   /**
    * Topic-based publish/subscribe channel.  Maintains a map of topics to
    * subscriptions.  When a message is published to a topic, all functions
@@ -30,10 +30,8 @@ export class PubSub {
    * strings corresponding to native Object properties, e.g. "constructor",
    * "toString", "hasOwnProperty", etc.
    *
-   * @param {boolean=} async Enable asynchronous behavior.  Recommended for
-   *     new code.  See notes on the publish() method.
    */
-  constructor(async = false) {
+  constructor() {
     this.disposed = false;
 
     /**
@@ -68,8 +66,9 @@ export class PubSub {
      * 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
+     * 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 =
@@ -86,11 +85,6 @@ export class PubSub {
      * @private {!Object<!Array<number>>}
      */
     this.topics = {};
-
-    /**
-     * @private @const {boolean}
-     */
-    this.async_ = Boolean(async);
   }
 
   /**
@@ -176,7 +170,7 @@ export class PubSub {
    * @param {Object} context Context in which to run the function.
    * @param {Array} args Arguments to pass to the function.
    */
-  static runAsync_(fn, context, args) {
+  static runAsync(fn, context, args) {
     queueMicrotask(() => {
       fn.apply(context, args);
     });
@@ -264,45 +258,16 @@ export class PubSub {
     if (keys) {
       const args = var_args;
 
-      if (this.async_) {
-        // For each key in the list of subscription keys for the topic, schedule
-        // the function to be applied to the arguments in the appropriate context.
-        for (let i = 0; i < keys.length; i++) {
-          const key = keys[i];
-
-          PubSub.runAsync_(
-            this.subscriptions[key + 1],
-            this.subscriptions[key + 2],
-            args,
-          );
-        }
-      } else {
-        this.publishDepth++;
-
-        try {
-          for (
-            let i = 0, len = keys.length;
-            i < len && !this.isDisposed();
-            i++
-          ) {
-            const key = keys[i];
-
-            this.subscriptions[key + 1].apply(
-              this.subscriptions[key + 2],
-              args,
-            );
-          }
-        } finally {
-          this.publishDepth--;
-
-          if (this.pendingKeys.length > 0 && this.publishDepth === 0) {
-            let pendingKey;
-
-            while ((pendingKey = this.pendingKeys.pop())) {
-              this.unsubscribeByKey(pendingKey);
-            }
-          }
-        }
+      // For each key in the list of subscription keys for the topic, schedule
+      // the function to be applied to the arguments in the appropriate context.
+      for (let i = 0, l = keys.length; i < l; i++) {
+        const key = keys[i];
+
+        PubSub.runAsync(
+          this.subscriptions[key + 1],
+          this.subscriptions[key + 2],
+          args,
+        );
       }
 
       return true;
@@ -362,4 +327,4 @@ export class PubSub {
   }
 }
 
-export const EventBus = new PubSub(true);
+export const EventBus = new PubSub();

src/services/pubsub/pubsub.spec.js

@@ -1,6 +1,7 @@
 import { PubSub } from "./pubsub.js";
 import { createInjector } from "../../core/di/injector.js";
 import { Angular } from "../../angular.js";
+import { wait } from "../../shared/utils.js";
 
 describe("PubSubProvider", () => {
   it("should be injectable", () => {
@@ -9,7 +10,6 @@ describe("PubSubProvider", () => {
     const $injector = createInjector(["test"]);
     expect($injector.has("$eventBus")).toBeTrue();
     expect($injector.get("$eventBus") instanceof PubSub).toBeTrue();
-    expect($injector.get("$eventBus").async_).toBeTrue();
   });
 });
 
@@ -117,18 +117,20 @@ describe("PubSub", function () {
     expect(pubsub.getCount("X")).toBe(0);
   });
 
-  it("should subscribe once correctly", function () {
+  it("should subscribe once correctly", async function () {
     let called;
     let context;
 
     called = false;
     pubsub.subscribeOnce("someTopic", () => {
       called = true;
     });
+    await wait();
     expect(pubsub.getCount("someTopic")).toBe(1);
     expect(called).toBe(false);
 
     pubsub.publish("someTopic");
+    await wait();
     expect(pubsub.getCount("someTopic")).toBe(0);
     expect(called).toBe(true);
 
@@ -140,10 +142,12 @@ describe("PubSub", function () {
       },
       context,
     );
+    await wait();
     expect(pubsub.getCount("someTopic")).toBe(1);
     expect(context.called).toBe(false);
 
     pubsub.publish("someTopic");
+    await wait();
     expect(pubsub.getCount("someTopic")).toBe(0);
     expect(context.called).toBe(true);
 
@@ -156,11 +160,13 @@ describe("PubSub", function () {
       },
       context,
     );
+    await wait();
     expect(pubsub.getCount("someTopic")).toBe(1);
     expect(context.called).toBe(false);
     expect(context.value).toBe(0);
 
     pubsub.publish("someTopic", 17);
+    await wait();
     expect(pubsub.getCount("someTopic")).toBe(0);
     expect(context.called).toBe(true);
     expect(context.value).toBe(17);
@@ -227,7 +233,7 @@ describe("PubSub", function () {
     }, 0);
   });
 
-  it("should subscribe once with bound function correctly", function () {
+  it("should subscribe once with bound function correctly", async function () {
     const context = { called: false, value: 0 };
 
     function subscriber(value) {
@@ -236,17 +242,21 @@ describe("PubSub", function () {
     }
 
     pubsub.subscribeOnce("someTopic", subscriber.bind(context));
+    await wait();
+
     expect(pubsub.getCount("someTopic")).toBe(1);
     expect(context.called).toBe(false);
     expect(context.value).toBe(0);
 
     pubsub.publish("someTopic", 17);
+    await wait();
+
     expect(pubsub.getCount("someTopic")).toBe(0);
     expect(context.called).toBe(true);
     expect(context.value).toBe(17);
   });
 
-  it("should subscribe once with partial function correctly", function () {
+  it("should subscribe once with partial function correctly", async function () {
     let called = false;
     let value = 0;
 
@@ -256,17 +266,21 @@ describe("PubSub", function () {
     }
 
     pubsub.subscribeOnce("someTopic", subscriber.bind(null, true));
+    await wait();
+
     expect(pubsub.getCount("someTopic")).toBe(1);
     expect(called).toBe(false);
     expect(value).toBe(0);
 
     pubsub.publish("someTopic", 17);
+    await wait();
+
     expect(pubsub.getCount("someTopic")).toBe(0);
     expect(called).toBe(true);
     expect(value).toBe(17);
   });
 
-  it("should handle self resubscribe correctly", function () {
+  it("should handle self resubscribe correctly", async function () {
     let value = null;
 
     function resubscribe(iteration, newValue) {
@@ -275,18 +289,26 @@ describe("PubSub", function () {
     }
 
     pubsub.subscribeOnce("someTopic", resubscribe.bind(null, 0));
+    await wait();
+
     expect(pubsub.getCount("someTopic")).toBe(1);
     expect(value).toBeNull();
 
     pubsub.publish("someTopic", "foo");
+    await wait();
+
     expect(pubsub.getCount("someTopic")).toBe(1);
     expect(value).toBe("foo:0");
 
     pubsub.publish("someTopic", "bar");
+    await wait();
+
     expect(pubsub.getCount("someTopic")).toBe(1);
     expect(value).toBe("bar:1");
 
     pubsub.publish("someTopic", "baz");
+    await wait();
+
     expect(pubsub.getCount("someTopic")).toBe(1);
     expect(value).toBe("baz:2");
   });
@@ -352,50 +374,57 @@ describe("PubSub", function () {
       expect(record.y).toBe("y");
     }
 
-    it("should call subscribed functions on publish", function () {
+    it("should call subscribed functions on publish", async function () {
       pubsub.subscribe(SOME_TOPIC, foo);
       pubsub.subscribe(SOME_TOPIC, bar, context);
 
       expect(pubsub.publish(SOME_TOPIC, { x: "x", y: "y" })).toBe(true);
+      await wait();
       expect(fooCalled).toBe(true, "foo() must have been called");
       expect(barCalled).toBe(true, "bar() must have been called");
     });
 
-    it("should not call unsubscribed functions on publish", function () {
+    it("should not call unsubscribed functions on publish", async function () {
       pubsub.subscribe(SOME_TOPIC, foo);
       pubsub.subscribe(SOME_TOPIC, bar, context);
 
       pubsub.publish(SOME_TOPIC, { x: "x", y: "y" });
+      await wait();
+      expect(fooCalled).toBe(true, "foo() must have been called");
+      expect(barCalled).toBe(true, "bar() must have been called");
       fooCalled = false;
       barCalled = false;
       expect(pubsub.unsubscribe(SOME_TOPIC, foo)).toBe(true);
-
       expect(pubsub.publish(SOME_TOPIC, { x: "x", y: "y" })).toBe(true);
+
+      await wait();
       expect(fooCalled).toBe(false, "foo() must not have been called");
       expect(barCalled).toBe(true, "bar() must have been called");
     });
 
-    it("should only call functions subscribed to the correct topic", function () {
+    it("should only call functions subscribed to the correct topic", async function () {
       pubsub.subscribe(SOME_TOPIC, bar, context);
       pubsub.subscribe("differentTopic", foo);
 
       pubsub.publish(SOME_TOPIC, { x: "x", y: "y" });
       fooCalled = false;
       barCalled = false;
 
+      await wait();
       expect(pubsub.publish(SOME_TOPIC, { x: "x", y: "y" })).toBe(true);
       expect(fooCalled).toBe(false, "foo() must not have been called");
       expect(barCalled).toBe(true, "bar() must have been called");
     });
 
-    it("should trigger functions if not arguments are provided", function () {
+    it("should trigger functions if not arguments are provided", async function () {
       let called = false;
       pubsub.subscribe(SOME_TOPIC, () => {
         called = true;
         0;
       });
 
       pubsub.publish(SOME_TOPIC);
+      await wait();
 
       expect(pubsub.publish(SOME_TOPIC)).toBe(true);
       expect(called).toBeTrue();