feat: add buffer reset functionality

Author: iNViTiON

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@invition/rxjs-sharelatest",
-  "version": "1.0.2",
+  "version": "1.0.3",
   "description": "An RxJS operator designed to limit source invocation and guarantee a swift initial response",
   "main": "./dist/index.js",
   "module": "./dist/index.mjs",

src/reuseSubject.ts

@@ -25,7 +25,7 @@ export class ReuseSubject<T> extends ReplaySubject<T> {
     this['_buffer'] = __buffer;
   }
 
-  public getBuffer(): (T | number)[] {
-    return (this['_buffer'] as (T | number)[]).slice();
-  }
+  public getBuffer = (): (T | number)[] => (this['_buffer'] as (T | number)[]).slice();
+
+  public resetBuffer = (): void => void (this['_buffer'].length = 0);
 }

src/shareLatest-bufferReset.spec.ts

@@ -0,0 +1,52 @@
+import { afterEach, beforeEach, describe, expect, it } from '@jest/globals';
+import { firstValueFrom, map, Observable, of, pipe, Subject, Subscription, tap, timeout, timer } from 'rxjs';
+import { shareLatest } from './shareLatest';
+
+describe('RxJS shareLatest operator with bufferLifetime', () => {
+  let count: number;
+  let subject: Subject<number>;
+  let value: number | undefined;
+  let randomNumber: number;
+  let observable: Observable<number>;
+  let subscription: Subscription;
+  let bufferReset: Subject<void>;
+
+  beforeEach(() => {
+    count = 0;
+    value = undefined;
+    randomNumber = Math.random() * 1000;
+    subject = new Subject();
+    bufferReset = new Subject();
+    observable = subject.pipe(
+      tap(() => ++count),
+      map(v => v * 2),
+      shareLatest(undefined, bufferReset.asObservable()),
+    );
+    subscription = observable.subscribe(newValue => (value = newValue));
+  });
+
+  afterEach(() => {
+    subscription?.unsubscribe();
+  });
+
+  it(`should't emit value if buffer expired`, async () => {
+    const InvSqrtMagic = 0x5f3759df;
+    const timeoutWithMagic = pipe(timeout({ first: 10, with: () => of(InvSqrtMagic) }));
+
+    subject.next(randomNumber);
+    await firstValueFrom(timer(0));
+    subscription.unsubscribe();
+    expect(count).toBe(1);
+    expect(value).toBe(randomNumber * 2);
+
+    let value2 = await firstValueFrom(observable.pipe(timeoutWithMagic));
+    expect(count).toBe(1);
+    expect(value2).toBe(randomNumber * 2);
+
+    // reset buffer
+    bufferReset.next();
+    value2 = await firstValueFrom(observable.pipe(timeoutWithMagic));
+    expect(count).toBe(1);
+    expect(value2).toBe(InvSqrtMagic);
+  });
+});

src/shareLatest.ts

@@ -1,14 +1,21 @@
 import type { MonoTypeOperatorFunction, Observable } from 'rxjs';
-import { asapScheduler, observeOn, share, timer } from 'rxjs';
+import { EMPTY, asapScheduler, ignoreElements, observeOn, share, takeUntil, tap, timer } from 'rxjs';
 import { ReuseSubject } from './reuseSubject';
 
-export function shareLatest<T>(bufferLifetime?: number): MonoTypeOperatorFunction<T> {
+export function shareLatest<T>(bufferLifetime?: number, bufferReset?: Observable<unknown>): MonoTypeOperatorFunction<T> {
   let previousBuffer: (number | T)[]
   let subject: ReuseSubject<T>;
   const connector = (): ReuseSubject<T> => subject = new ReuseSubject<T>(previousBuffer, bufferLifetime);
   return (source$: Observable<T>) =>
     source$.pipe(
       observeOn(asapScheduler),
+      takeUntil(bufferReset?.pipe(
+        // Reset the buffer when the lifetime observable emits.
+        tap(() => subject.resetBuffer()),
+        // Ignore all emissions from the lifetime observable to prevent it from completing the source.
+        // This make takeUntil sit here as a noop but just manage subscriber for bufferLifetimeObs.
+        ignoreElements(),
+      ) ?? EMPTY),
       share({
         connector,
         resetOnComplete: true,