Rxjs：如何在管道中绑定主题？答案

【问题标题】：Rxjs: How to bind subject in pipe?Rxjs：如何在管道中绑定主题？
【发布时间】：2019-06-28 18:26:22
【问题描述】：

我想将一个主题集成到一个管道中，以便以后的操作员可以订阅早期的 observables。

上下文：我想实现一个更复杂的“去抖动”，其中只有某些事件可以阻止某些其他事件。为此，首先我想重新实现“去抖动”。

这就是我的想法：

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 是否已经存在？

【问题讨论】：

标签： rxjs

【解决方案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 实现短很多，但它的性能可能更差。

【讨论】：