Rxjs: Как связать предмет в трубе?

Question

Я хотел бы интегрировать тему в канал так, чтобы более поздние операторы могли подписаться на более ранние наблюдаемые.

Контекст: я хочу реализовать более сложную «дебад», когда только определенные события могут блокировать некоторые другие события. Для этого сначала я хочу переопределить "debounce".

Вот что я имею в виду:

const debounced = myEventStream.pipe(
    useSubject(futureEvents => pipe(
        flatFilter(
            myEvent => race(
                interval(10).pipe(map(x => true)),
                futureEvents.pipe(/* filter(...), */ map(x => false))
            )
        )
    )),
);

useSubject предоставит следующие этапы трубы с наблюдаемыми будущими событиями. Таким образом, если существует flatFilter, события не фильтруются, если интервал срабатывает первым, а фильтруются, если наступает более новое событие.

Разумно ли такое useSubject - может ли оно работать? Такой useSubject уже существует?

Answer 1

Работает как шарм:

function mergeFilter<T>(
    predicate: (arg: T) => Observable<boolean>
): MonoTypeOperatorFunction<T> {
    return mergeMap(evt =>
        predicate(evt).pipe(
            take(1),
            filter(e => e),
            map(e => evt)
        )
    );
}

function useSubject<T, O>(
    fn: (o: Observable<T>) => OperatorFunction<T, O>
): OperatorFunction<T, O> {
    return function(input: Observable<T>): Observable<O> {
        const subject = new Subject<T>();
        const observer: PartialObserver<T> = subject;
        return input.pipe(
            tap(observer),
            fn(subject)
        );
    };
}

function debounceTime<T>(time: number) {
    return useSubject<T, T>(futureEvents =>
        mergeFilter(myEvent =>
            race(
                // allow event after 10ms
                interval(10, s).pipe(map(x => true)),
                // discard event when new event arrived
                futureEvents.pipe(map(x => false))
            )
        )
    );
}

Это все еще намного короче, чем исходная реализация rxjs debounce, но его производительность может быть хуже.