Merge pull request #19 from epferrari/will-idle-spec

Process unsubscribe requests transactionally in order

Author: epferrari

src/strongbus.ts

@@ -66,6 +66,14 @@ export class Bus<TEventMap extends object = object> implements Scannable<TEventM
   private readonly bus = new Map<EventKeys<TEventMap>|Events.WILDCARD, Set<EventHandlers.GenericHandler>>();
   private readonly lifecycle = new Map<Lifecycle, Set<EventHandlers.GenericHandler>>();
 
+  // Queue of unsubscription requests so that they are processed transactionally in order
+  private readonly _unsubQueue: {
+    token: string;
+    event: EventKeys<TEventMap>|Events.WILDCARD;
+    handler: EventHandlers.GenericHandler;
+  }[] = [];
+  private _purgingUnsubQueue: boolean = false;
+
   constructor(options?: Options) {
     this.options = {
       ...Bus.defaultOptions,
@@ -426,13 +434,35 @@ export class Bus<TEventMap extends object = object> implements Scannable<TEventM
   private cacheListener(event: EventKeys<TEventMap>|Events.WILDCARD, handler: EventHandlers.GenericHandler): Events.Subscription {
     const token = randomId();
     const sub = generateSubscription(() => {
+      this._unsubQueue.push({
+        token,
+        event,
+        handler
+      });
+      this.purgeUnsubQueue();
+    });
+    this.subscriptionCache.set(token, sub);
+    return sub;
+  }
+
+  private purgeUnsubQueue() {
+    if(this._purgingUnsubQueue) {
+      // There is another purge loop running.
+      return;
+    }
+
+    this._purgingUnsubQueue = true;
+
+    while(this._unsubQueue.length) {
+      const {token, event, handler} = this._unsubQueue.shift();
       if(this.subscriptionCache.has(token)) {
         this.subscriptionCache.delete(token);
+        // lifecycle events may trigger additional unsubs, which will be pushed to the end of queue and handled in a subsequent iteration of this loop
         this.removeListener(event, handler);
       }
-    });
-    this.subscriptionCache.set(token, sub);
-    return sub;
+    }
+
+    this._purgingUnsubQueue = false;
   }
 
   private removeListener(event: EventKeys<TEventMap>|Events.WILDCARD, handler: EventHandlers.GenericHandler): void {

src/strongbus_spec.ts

@@ -431,6 +431,7 @@ describe('Strongbus.Bus', () => {
     let onActive: jasmine.Spy;
     let onWillIdle: jasmine.Spy;
     let onIdle: jasmine.Spy;
+    let onError: jasmine.Spy;
 
     beforeEach(() => {
       bus.hook('willAddListener', onWillAddListener = jasmine.createSpy('onWillAddListener'));
@@ -441,6 +442,7 @@ describe('Strongbus.Bus', () => {
       bus.hook('active', onActive = jasmine.createSpy('onActive'));
       bus.hook('willIdle', onWillIdle = jasmine.createSpy('onWillIdle'));
       bus.hook('idle', onIdle = jasmine.createSpy('onIdle'));
+      bus.hook('error', onError = jasmine.createSpy('error'));
     });
 
     it('allows subscription to meta events', () => {
@@ -561,6 +563,18 @@ describe('Strongbus.Bus', () => {
       sub2();
     });
 
+    it('raises "error" when errors are thrown in the listener', () => {
+      const error = new Error('Error in callback');
+      bus.on('bar', () => {
+        throw error;
+      })
+      bus.emit('bar', true);
+      expect(onError).toHaveBeenCalledWith({
+        error,
+        event: 'bar'
+      });
+    });
+
     describe('given bus has delegates', () => {
       let delegate: DelegateTestBus;
       let onDelegateWillAddListener: jasmine.Spy;
@@ -627,6 +641,16 @@ describe('Strongbus.Bus', () => {
         expect(onActive).toHaveBeenCalledTimes(1);
       });
 
+      it('handles unsubscribes fired from hooks', async () => {
+        const sub1 = bus.on('foo', () => {});
+        const sub2 = bus.on('foo', () => {});
+        bus.hook('willRemoveListener', () => sub2());
+
+        sub1();
+        expect(onWillIdle).toHaveBeenCalledTimes(1);
+        expect(onIdle).toHaveBeenCalledTimes(1);
+      });
+
       it('raises "idle" events independently of delegates', () => {
         const foosub = bus.on('foo', onTestEvent);
         const fooSub2 = delegate.on('foo', onTestEvent);