fix(broker): issue with wrongly intercepting Subject next and publish

Author: marcj

packages/broker/src/broker.ts

@@ -219,31 +219,33 @@ export class BrokerQueueChannel<T> {
 
 export class RefCountedSubject<T> extends Subject<T> {
     private refCount = 0;
-    private readonly onFirst: () => void;
-    private readonly onLast: () => void;
-    public skipPublish = 0;
 
-    constructor(onFirst: () => void, onLast: () => void) {
+    constructor(protected onFirst: () => void, protected onLast: () => void, protected onPublish: (value: T) => void) {
         super();
-        this.onFirst = onFirst;
-        this.onLast = onLast;
     }
 
     // @ts-ignore
     override subscribe(...args: Parameters<Subject<T>['subscribe']>): Subscription {
-        if (this.refCount++ === 1) {
-            // We skip 1
+        if (this.refCount++ === 0) {
             this.onFirst();
         }
 
         const sub = super.subscribe(...args);
         sub.add(() => {
-            if (--this.refCount === 1) {
+            if (--this.refCount === 0) {
                 this.onLast();
             }
         });
         return sub;
     }
+
+    override next(value: T, publish = true): void {
+        if (publish) {
+            this.onPublish(value);
+        } else {
+            super.next(value);
+        }
+    }
 }
 
 const subjectFinalizer = new FinalizationRegistry<{
@@ -275,10 +277,7 @@ class BrokerBusSubjectHandle {
         this.releaseChannel = this.channel.subscribe(value => {
             for (const subjectRef of this.subjects) {
                 const subject = subjectRef.deref();
-                if (subject) {
-                    subject.skipPublish++;
-                    subject.next(value);
-                }
+                if (subject) subject.next(value, false);
             }
         }).catch((e) => {
             this.errorHandler.subscribeFailed(this.channel.name, ensureError(e));
@@ -295,19 +294,15 @@ class BrokerBusSubjectHandle {
             () => {
                 this.releaseSubject(subjectRef!);
             },
+            (value) => {
+                this.publish(value);
+            },
         );
         subjectRef = new WeakRef(subject);
         subjectFinalizer.register(subject, {
             handle: this,
             subjectRef,
         });
-        subject.subscribe(value => {
-            if (subject.skipPublish) {
-                subject.skipPublish--;
-                return;
-            }
-            this.publish(value);
-        });
         return subject;
     }
 

packages/broker/tests/broker.spec.ts

@@ -181,32 +181,66 @@ test('bus subject gc', async () => {
 });
 
 test('bus subject 1', async () => {
-
     const bus = new BrokerBus(await adapterFactory());
     const handles: BrokerBus['subjectHandles'] = (bus as any).subjectHandles;
 
     type Events = { type: 'user-created', id: number } | { type: 'user-deleted', id: number };
 
     const caughtEvents: Events[] = [];
+    const caughtEvents2: Events[] = [];
 
     const subject1 = bus.subject<Events>('/events');
     const subject2 = bus.subject<Events>('/events');
     expect(handles.get('/events')!.isSubscribed).toBe(false);
     const sub = subject2.subscribe((event) => {
         caughtEvents.push(event);
     });
+    const sub2 = subject1.subscribe((event) => {
+        caughtEvents2.push(event);
+    });
     expect(handles.get('/events')!.isSubscribed).toBe(true);
 
     subject1.next({ type: 'user-created', id: 2 });
     await sleep(0.1);
     expect(handles.size).toBe(1);
     expect(caughtEvents.length).toBe(1);
+    expect(caughtEvents2.length).toBe(1);
     sub.unsubscribe();
+    sub2.unsubscribe();
     await sleep(0.1);
     expect(handles.size).toBe(0);
 });
 
 test('bus subject 2', async () => {
+    const adapter = await adapterFactory();
+    const bus1 = new BrokerBus(adapter);
+    const bus2 = new BrokerBus(adapter);
+
+    type Events = { type: 'user-created', id: number } | { type: 'user-deleted', id: number };
+
+    const caughtEvents: Events[] = [];
+    const caughtEvents2: Events[] = [];
+
+    const subject1 = bus1.subject<Events>('/events');
+    const subject2 = bus2.subject<Events>('/events');
+    const sub = subject2.subscribe((event) => {
+        caughtEvents.push(event);
+    });
+    const sub2 = subject1.subscribe((event) => {
+        caughtEvents2.push(event);
+    });
+
+    await sleep(0.1);
+    subject1.next({ type: 'user-created', id: 2 });
+    await sleep(0.1);
+    expect(caughtEvents.length).toBe(1);
+    expect(caughtEvents2.length).toBe(1);
+    sub.unsubscribe();
+    sub2.unsubscribe();
+    await sleep(0.1);
+});
+
+test('bus subject 3', async () => {
 
     const bus = new BrokerBus(await adapterFactory());
     const handles: BrokerBus['subjectHandles'] = (bus as any).subjectHandles;