RxJS：如果请求很慢则显示加载

Question

本想用 RxJS 优雅的解决这个问题，但是尝试了各种方法后，我不知道该怎么做......

我的需求很普遍：我打了一个休息电话，即。我有一个承诺。 如果响应很快，我只想使用结果。 如果来得慢，我想显示一个微调器，直到请求完成。 这是为了避免旋转器闪烁，然后是数据。

也许可以通过制作两个可观察对象来完成：一个带有承诺，另一个带有超时并将微调器显示为副作用。 我尝试了switch()，但没有取得多大成功，可能是因为另一个可观察对象没有产生值。

有人实现过类似的东西吗？

Answer 1

基于@PhiLho 的answer，我编写了一个可管道操作符，它正是这样做的：

export function executeDelayed<T>(
    fn : () => void,
    delay : number,
    thisArg? : any
) : OperatorFunction<T, T> {
    return function executeDelayedOperation(source : Observable<T>) : Observable<T> {
        let timerSub = timer(delay).subscribe(() => fn());
        return source.pipe(
            tap(
                () => {
                    timerSub.unsubscribe();
                    timerSub = timer(delay).subscribe(() => fn());
                },
                undefined,
                () => {
                    timerSub.unsubscribe();
                }
            )
        );
    }
}

基本上它返回一个函数，该函数获取Observable source。
然后它使用给定的delay 启动一个timer。
如果此计时器发出next-事件，则调用该函数。
但是，如果源发出 next，则 timer 将被取消并开始一个新的。
在源的complete中，timer终于被取消了。 然后可以像这样使用此运算符：

this.loadResults().pipe(
    executeDelayed(
        () => this.startLoading(),
        500
    )
).subscribe(results => this.showResult())

我自己并没有编写很多运算符，所以这种运算符实现可能不是最好的，但它确实有效。
欢迎任何关于如何优化它的建议:)

编辑： 正如@DauleDK 提到的，在这种情况下，错误不会停止计时器，fn 将在delay 之后调用。如果这不是你想要的，你需要在tap中添加一个onError-callback，它调用timerSub.unsubscribe()：

export function executeDelayed<T>(
    fn : () => void,
    delay : number,
    thisArg? : any
) : OperatorFunction<T, T> {
    return function executeDelayedOperation(source : Observable<T>) : Observable<T> {
        let timerSub = timer(delay).subscribe(() => fn());
        return source.pipe(
            tap(
                () => {
                    timerSub.unsubscribe();
                    timerSub = timer(delay).subscribe(() => fn());
                },
                () => timerSub.unsubscribe(),   // unsubscribe on error
                () => timerSub.unsubscribe()
            )
        );
    }
}

Answer 2

这是我用过的一个例子。我们在这里假设您也获得了想要发送到服务器的数据作为 Observable，称为query$。传入的查询将触发loadResults 函数，该函数应返回一个承诺并将结果放入results$ observable。

现在的诀窍是使用observable$.map(() => new Date()) 来获取最后一个发出值的时间戳。

然后我们可以比较最后一个查询的时间戳和来自服务器的最后一个响应。

由于您不仅希望显示加载动画，还希望在显示动画之前等待 750 毫秒，因此我们引入了延迟时间戳。更多解释请参见下面的 cmets。

最后我们得到了包含true 或false 的isLoading$ Observable。订阅它，以便在显示/隐藏加载动画时收到通知。

const query$ = ... // From user input.

const WAIT_BEFORE_SHOW_LOADING = 750;

const results$ = query$.flatMapLatest(loadResults);

const queryTimestamp$ = query$.map(() => new Date());
const resultsTimestamp$ = results$.map(() => new Date());
const queryDelayTimestamp$ = (
    // For every query coming in, we wait 750ms, then create a timestamp.
    query$
    .delay(WAIT_BEFORE_SHOW_LOADING)
    .map(() => new Date())
);

const isLoading$ = (
    queryTimestamp$.combineLatest(
        resultsTimestamp$,
        queryDelayTimestamp$,
        (queryTimestamp, resultsTimestamp, delayTimestamp) => {
            return (
                // If the latest query is more recent than the latest
                // results we got we can assume that
                // it's still loading.
                queryTimestamp > resultsTimestamp &&
                // But only show the isLoading animation when delay has passed
                // as well.
                delayTimestamp > resultsTimestamp
            );
        }
    )
    .startWith(false)
    .distinctUntilChanged()
);

Answer 3

好吧，我在通勤中想了想，找到了解决办法……

你可以在http://plnkr.co/edit/Z3nQ8q找到我的实验场

简而言之，解决方案是实际订阅可观察到的处理微调器（而不是尝试以某种方式组合它）。 如果 Rest 请求的结果出现在 observable 触发之前，我们只是取消微调器的可处置（订阅），所以它什么也不做。 否则，observable 将触发并显示其微调器。然后我们可以在收到响应后隐藏它。

代码：

function test(loadTime)
{
  var prom = promiseInTime(loadTime, { id: 'First'}); // Return data after a while
  var restO = Rx.Observable.fromPromise(prom);

  var load = Rx.Observable.timer(750);
  var loadD = load.subscribe(
    undefined,
    undefined,
    function onComplete() { show('Showing a loading spinner'); });

  restO.subscribe(
    function onNext(v) { show('Next - ' + JSON.stringify(v)); },
    function onError(e) { show('Error - ' + JSON.stringify(e)); loadD.dispose(); },
    function onComplete() { show('Done'); loadD.dispose(); }
  );
}

test(500);
test(1500);

不确定这是否是使用 RxJS 执行此操作的惯用方式，但它似乎有效...... 当然，欢迎使用其他解决方案。

Answer 4

这是我的解决方案：

public static addDelayedFunction<T>(delayedFunction: Function, delay_ms: number): (mainObs: Observable<T>) => Observable<T> {
    const stopTimer$: Subject<void> = new Subject<void>();
    const stopTimer = (): void => {
        stopTimer$.next();
        stopTimer$.complete();
    };
    const catchErrorAndStopTimer = (obs: Observable<T>): Observable<T> => {
        return obs.pipe(catchError(err => {
            stopTimer();
            throw err;
        }));
    };
    const timerObs: Observable<any> = of({})
        .pipe(delay(delay_ms))
        .pipe(takeUntil(stopTimer$))
        .pipe(tap(() => delayedFunction()));
    return (mainObs: Observable<T>) => catchErrorAndStopTimer(
        of({})
            .pipe(tap(() => timerObs.subscribe()))
            .pipe(mergeMap(() => catchErrorAndStopTimer(mainObs.pipe(tap(stopTimer)))))
    );
}

Answer 5

就在获取数据之前，即。创建微调器，为创建微调器的函数设置超时。假设您愿意等待半秒钟，直到显示微调器......它会是这样的：

spinnerTimeout = setTimeout(showSpinner, 500)
fetch(url).then(data => {
  if (spinner) {
    clearTimeout(spinnerTimeout) //this is critical
    removeSpinnerElement()
  }
  doSomethingWith(data)
});

编辑：如果不明显，如果数据早于 500 毫秒（ish）到达，则 clearTimer 会停止 showSpinner 执行。