// Copyright 2017-2026 @pezkuwi/rpc-core authors & contributors // SPDX-License-Identifier: Apache-2.0 import type { Observable } from 'rxjs'; import { catchError, distinctUntilChanged, publishReplay, refCount, tap } from 'rxjs'; import { stringify } from '@pezkuwi/util'; import { refCountDelay } from './refCountDelay.js'; export type DrrResult = (source$: Observable) => Observable; interface Options { delay?: number; skipChange?: boolean; skipTimeout?: boolean; } function CMP (a: unknown, b: unknown): boolean { return stringify({ t: a }) === stringify({ t: b }); } function ERR (error: Error): Observable { throw error; } function NOOP (): void { // empty } /** * Shorthand for distinctUntilChanged(), publishReplay(1) and refCount(). * * @ignore * @internal */ export function drr ({ delay, skipChange = false, skipTimeout = false }: Options = {}): DrrResult { return (source$: Observable): Observable => source$.pipe( catchError(ERR), skipChange ? tap(NOOP) : distinctUntilChanged(CMP), // eslint-disable-next-line @typescript-eslint/no-deprecated publishReplay(1), skipTimeout // eslint-disable-next-line @typescript-eslint/no-deprecated ? refCount() : refCountDelay(delay) ); }