feat: replay to first subscriber subject

Author: KatoakDR

electron/common/observable/__tests__/replay-first-subscriber-only-subject.test.ts

@@ -0,0 +1,161 @@
+import type * as rxjs from 'rxjs';
+import { ReplayFirstSubscriberOnlySubject } from '../replay-first-subscriber-only.subject';
+
+describe('ReplayFirstSubscriberOnlySubject', () => {
+  let subscriber1NextSpy: jest.Mock;
+  let subscriber2NextSpy: jest.Mock;
+
+  let subscriber1ErrorSpy: jest.Mock;
+  let subscriber2ErrorSpy: jest.Mock;
+
+  let subscriber1CompleteSpy: jest.Mock;
+  let subscriber2CompleteSpy: jest.Mock;
+
+  let subject: rxjs.SubjectLike<string>;
+
+  beforeEach(() => {
+    subject = new ReplayFirstSubscriberOnlySubject<string>();
+
+    subscriber1NextSpy = jest.fn();
+    subscriber2NextSpy = jest.fn();
+
+    subscriber1ErrorSpy = jest.fn();
+    subscriber2ErrorSpy = jest.fn();
+
+    subscriber1CompleteSpy = jest.fn();
+    subscriber2CompleteSpy = jest.fn();
+  });
+
+  afterEach(() => {
+    jest.clearAllMocks();
+    jest.clearAllTimers();
+  });
+
+  it('should replay buffered events to the first subscriber only', () => {
+    // buffer events for the first subscriber
+    subject.next('A');
+    subject.next('B');
+
+    // first subscriber
+    subject.subscribe({
+      next: subscriber1NextSpy,
+      error: subscriber1ErrorSpy,
+      complete: subscriber1CompleteSpy,
+    });
+
+    // new event to all current subscribers
+    subject.next('C');
+
+    // second subscriber
+    subject.subscribe({
+      next: subscriber2NextSpy,
+      error: subscriber2ErrorSpy,
+      complete: subscriber2CompleteSpy,
+    });
+
+    // new event to all current subscribers
+    subject.next('D');
+
+    // done
+    subject.complete();
+
+    // First subscriber receives all buffered and new events.
+    expect(subscriber1NextSpy).toHaveBeenCalledTimes(4);
+    expect(subscriber1NextSpy).toHaveBeenNthCalledWith(1, 'A');
+    expect(subscriber1NextSpy).toHaveBeenNthCalledWith(2, 'B');
+    expect(subscriber1NextSpy).toHaveBeenNthCalledWith(3, 'C');
+    expect(subscriber1NextSpy).toHaveBeenNthCalledWith(4, 'D');
+    expect(subscriber1ErrorSpy).toHaveBeenCalledTimes(0);
+    expect(subscriber1CompleteSpy).toHaveBeenCalledTimes(1);
+
+    // Subsequent subscribers only receive new events.
+    expect(subscriber2NextSpy).toHaveBeenCalledTimes(1);
+    expect(subscriber2NextSpy).toHaveBeenNthCalledWith(1, 'D');
+    expect(subscriber2ErrorSpy).toHaveBeenCalledTimes(0);
+    expect(subscriber2CompleteSpy).toHaveBeenCalledTimes(1);
+  });
+
+  it('should emit error to the first subscriber only', () => {
+    // buffer events for the first subscriber
+    subject.next('A');
+    subject.error(new Error('test error'));
+
+    // first subscriber
+    subject.subscribe({
+      next: subscriber1NextSpy,
+      error: subscriber1ErrorSpy,
+      complete: subscriber1CompleteSpy,
+    });
+
+    // new event to all current subscribers
+    subject.next('C');
+
+    // second subscriber
+    subject.subscribe({
+      next: subscriber2NextSpy,
+      error: subscriber2ErrorSpy,
+      complete: subscriber2CompleteSpy,
+    });
+
+    // new event to all current subscribers
+    subject.next('D');
+
+    // done
+    subject.complete();
+
+    // First subscriber receives all buffered events then errors.
+    expect(subscriber1NextSpy).toHaveBeenCalledTimes(1);
+    expect(subscriber1NextSpy).toHaveBeenNthCalledWith(1, 'A');
+    expect(subscriber1ErrorSpy).toHaveBeenCalledTimes(1);
+    expect(subscriber1CompleteSpy).toHaveBeenCalledTimes(0);
+
+    // Subsequent subscribers only receive new events.
+    expect(subscriber2NextSpy).toHaveBeenCalledTimes(1);
+    expect(subscriber2NextSpy).toHaveBeenNthCalledWith(1, 'D');
+    expect(subscriber2ErrorSpy).toHaveBeenCalledTimes(0);
+    expect(subscriber2CompleteSpy).toHaveBeenCalledTimes(1);
+  });
+
+  it('should emit error to all subscribers', () => {
+    // buffer events for the first subscriber
+    subject.next('A');
+    subject.next('B');
+
+    // first subscriber
+    subject.subscribe({
+      next: subscriber1NextSpy,
+      error: subscriber1ErrorSpy,
+      complete: subscriber1CompleteSpy,
+    });
+
+    // new event to all current subscribers
+    subject.next('C');
+
+    // second subscriber
+    subject.subscribe({
+      next: subscriber2NextSpy,
+      error: subscriber2ErrorSpy,
+      complete: subscriber2CompleteSpy,
+    });
+
+    // new event to all current subscribers
+    subject.error(new Error('test error'));
+    subject.next('D'); // no subscriber will receive this because of the error
+
+    // done
+    subject.complete();
+
+    // First subscriber receives all buffered and new events.
+    expect(subscriber1NextSpy).toHaveBeenCalledTimes(3);
+    expect(subscriber1NextSpy).toHaveBeenNthCalledWith(1, 'A');
+    expect(subscriber1NextSpy).toHaveBeenNthCalledWith(2, 'B');
+    expect(subscriber1NextSpy).toHaveBeenNthCalledWith(3, 'C');
+    expect(subscriber1ErrorSpy).toHaveBeenCalledTimes(1);
+    expect(subscriber1CompleteSpy).toHaveBeenCalledTimes(0);
+
+    // Subsequent subscribers only receive new events.
+    expect(subscriber2NextSpy).toHaveBeenCalledTimes(0);
+    expect(subscriber2ErrorSpy).toHaveBeenCalledTimes(1);
+    expect(subscriber2CompleteSpy).toHaveBeenCalledTimes(0);
+  });
+});

electron/common/observable/index.ts

@@ -0,0 +1 @@
+export * from './replay-first-subscriber-only.subject';

electron/common/observable/replay-first-subscriber-only.subject.ts

@@ -0,0 +1,58 @@
+import * as rxjs from 'rxjs';
+
+/**
+ * A custom `ReplaySubject` that only replays the buffered events
+ * to the first subscriber. All subsequent subscribers will only
+ * receive new events, as if subscribed to a `Subject`.
+ * https://stackoverflow.com/a/69390202/470818
+ */
+export class ReplayFirstSubscriberOnlySubject<T>
+  implements rxjs.SubjectLike<T>
+{
+  /**
+   * Buffers events and replays them to the first subscriber.
+   * Continues to emit new events.
+   */
+  private replaySubject: rxjs.Subject<T>;
+
+  /**
+   * Subject used for all subsequent subscribers.
+   * They only receive new emits, not buffered events before subscribing.
+   */
+  private subject?: rxjs.Subject<T>;
+
+  constructor() {
+    this.replaySubject = new rxjs.ReplaySubject<T>();
+  }
+
+  public next(value: T): void {
+    // Notify the first subscriber of the value.
+    this.replaySubject.next(value);
+    // Notify all subsequent subscribers of the value.
+    this.subject?.next(value);
+  }
+
+  public error(error: Error): void {
+    // Notify the first subscriber of the error.
+    this.replaySubject.error(error);
+    // Notify all subsequent subscribers of the error.
+    this.subject?.error(error);
+  }
+
+  public complete(): void {
+    // Notify the first subscriber of the completion.
+    this.replaySubject.complete();
+    // Notify all subsequent subscribers of the completion.
+    this.subject?.complete();
+  }
+
+  public subscribe(observer: Partial<rxjs.Observer<T>>): rxjs.Unsubscribable {
+    const subscription = (this.subject ?? this.replaySubject).subscribe(
+      observer
+    );
+    if (!this.subject) {
+      this.subject = new rxjs.Subject<T>();
+    }
+    return subscription;
+  }
+}