通过流切换 API 调用答案

【问题标题】：Toggling API call by a stream通过流切换 API 调用
【发布时间】：2017-05-29 10:53:30
【问题描述】：

这里：

import Rx from 'rxjs';

function fakeApi(name, delay, response) {
  return new Rx.Observable(observer => {
    console.log(`${name}: Request.`)
    let running = true;
    const id = setTimeout(() => {
      console.log(`${name}: Response.`)
      running = false;
      observer.next(response);
      observer.complete();
    }, delay);
    return () => {
      if(running) console.log(`${name}: Cancel.`)
      clearTimeout(id);
    }
  })
}

function apiSearch() { return fakeApi('Search', 4000, "This is a result of the search."); }


//============================================================

const messages$ = new Rx.Subject();

const toggle$ = messages$.filter(m => m === 'toggle');

const searchDone$ = toggle$.flatMap(() => 
  apiSearch().takeUntil(toggle$)
);

searchDone$.subscribe(m => console.log('Subscriber:', m))

setTimeout(() => {
  // This one starts the API call.
  toggle$.next('toggle'); 
}, 2000)

setTimeout(() => {
  // This one should only cancel the API call in progress, not to start a new one.
  toggle$.next('toggle'); 
}, 3000)

setTimeout(() => {
  // And this should start a new request again...
  toggle$.next('toggle'); 
}, 9000)

我的意图是启动 API 调用并在它正在进行时通过相同的toggle$ 信号停止它。代码的问题是toggle$ 每次都会启动一个新的 API 调用。我希望它在已经运行一个呼叫时不要启动新呼叫，而只是停止已经在运行的呼叫。当apiSearch() 正在运行时，我应该以某种方式从toggle$ 流中“取消订阅”最外层的flatMap。我想有必要重组代码来实现这种行为...... RxJS 的做法是什么？

更新：经过更多调查和用户指南查找，我得到了这个：

const searchDone$ = toggle$.take(1).flatMap(() =>
  apiSearch().takeUntil(toggle$)
).repeat()

按应有的方式工作。感觉还是有点神秘。你们 RxJS 的人会这样解决吗？

【问题讨论】：

如果你能画一个简单的大理石图，它可能会更容易帮助你:)。用油漆或类似物，甚至用手拍一张照片，真的很简单

标签： javascript rxjs reactive-programming reactivex

【解决方案1】：

我认为您的解决方案只能工作一次，因为您使用的是take(1)。你可以这样做：

const searchDone$ = toggle$
    .let(observable => {
        let pending;

        return observable
            .switchMap(() => {
                let innerObs;

                if (pending) {
                    innerObs = Observable.empty();
                } else {
                    pending = innerObs = apiSearch();
                }

                return innerObs.finally(() => pending = null);
            });
    });

我使用let() 只是为了包装pending 而没有在父范围内声明它。 switchMap() 运算符会自动为您取消订阅，而无需使用 take*()。

您的测试setTimeouts 的输出将如下所示：

Search: Request.
Search: Cancel.
Search: Request.
Search: Response.
Subscriber: This is a result of the search.

【讨论】：