rxjs，递归调用api，直到获取所有项目答案

【问题标题】：rxjs, recursively call api until all items are fetchedrxjs，递归调用api，直到获取所有项目
【发布时间】：2018-04-29 17:15:21
【问题描述】：

我需要调用 api 并从端点获取所有项目。但是，我们一次只能获取 100 个项目。

响应如下所示：

{
    elements: Array<any>,
    totalCount: number,
}

端点将是/api/items?drop=${drop}&take=100，drop 用于分页。

我认为这将在某处涉及 scan 和 takeWhile 运算符。这是我得到的：

    const subject = new Subject();
    const rec = subject.pipe(
        scan((acc: number, curr: number) => (curr + 100), 0),
        switchMap(drop => this.http.get(`api/items?drop=${drop}&take=100`)),
        takeWhile((r: any) => r.elements.length === 100),
        tap(r => subject.next())
    );

【问题讨论】：

你知道会有多少商品吗？
只返回没有记录从 API 并增加直到没有
@Antoniossss 不，我不知道物品的数量
使用变量“page”（取值0,1,2,...），然后从page*100询问到(page+1)*100-1，或者使用slice (页*100,(页+1)*100)
blog.angularindepth.com/rxjs-understanding-expand-a5f8b41a3602

标签： angular rxjs

【解决方案1】：

你不需要知道有多少元素会被返回，只是永远不会有多少:)

Observable.range(0, 1000000).pipe(
  concatMap(page => this.http.get(`api/items?drop=${page * take}&take=${take}`))
  takeWhile(results => results.elements.length === take)
)

起初我以为使用range 作为源会堆积请求并破坏分页的目的，但 concatMap 会自动限制。

来自learnrxjs.io/operators/transformation/concatmap

注意 concatMap 和 mergeMap 的区别。因为 concatMap 在前一个完成之前不会订阅下一个 observable，所以延迟 2000ms 的来自源的值将首先发出。与立即订阅内部 observable 的 mergeMap 相比，具有较小延迟（1000 毫秒）的 observable 将发出，然后是需要 2000 毫秒才能完成的 observable。

这是对 concatMap 的测试，它显示在之前的结果发出之前不会调用 getPage

const take = 100;
const getPage = (page) => {
  console.log('reading page', page);
  return Rx.Observable.of(page).delay(1000);
}

Rx.Observable.range(0,3)
  .concatMap(getPage)
  .subscribe(results => console.log('concatMap results', results));

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.10/Rx.js"&gt;&lt;/script&gt;

与 mergeMap 对比，它会立即对getPage 进行所有调用

const take = 100;
const getPage = (page) => {
  console.log('reading page', page);
  return Rx.Observable.of(page).delay(1000);
}

Rx.Observable.range(0,3)
  .mergeMap(getPage)
  .subscribe(results => console.log('mergeMap results', results));

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.10/Rx.js"&gt;&lt;/script&gt;

【讨论】：

【解决方案2】：

似乎我通过使用外部变量让它工作了，但它并不是很漂亮......

const take = 100;
const subject = new BehaviorSubject(0);
// var used for pagination
let i = 0;
const rec = subject.pipe(
    map((acc: number) => (acc * 100)),
    switchMap(drop => this.http.get(`api/items?drop=${drop}&take=${take}`)),
    map((r: any) => r.elements),
    tap(items=> {
        if (items.length === take)
            subject.next(++i);
    })
);
return rec;

【讨论】：

【解决方案3】：

希望这会有所帮助:)

我会选择更简单的东西，订阅有点必要，但我想我们可以牺牲一些花哨的可读性。您可能应该考虑错误和重试，因为这显然只是幸福的道路

const subject = new Subject();
const createUrl = (drop = 0) => `api/items?drop=${drop}&take=100`
const fetchApi = url => Observable.fromPromise(fetch(url))

const requestItems = (pagination = 0) => {
    fetchApi(createUrl(pagination))
    .take(1)
    .subscribe(response => {
        if(response.elements.length === 100) {
            requestItems(pagination++)
        }
        subject.next(response)
    })
}

requestItems()

【讨论】：

我认为那里有一个错误。在订阅中，您正在执行 subject.next。但是，除非我错过了什么，否则所述主题不会在任何地方使用？

【解决方案4】：

详细阐述@cartant 的建议，您可以解决您的需求，而不必依赖主题，而是使用expand 运算符，按照以下 sn-p 行

let counterOfItemsFetched = 0;
function api() {
    // console.log('counterOfItemsFetched', counterOfItemsFetched)
    return counterOfItemsFetched < 1000 ? 100 : 1;
}

function apiAsynCall() {
    return Observable.of(api()).delay(1000);
}

Observable.of(0).pipe(
    expand(() => apiAsynCall().pipe(tap(itemsFetched => counterOfItemsFetched = counterOfItemsFetched + itemsFetched))),
    filter(counter => counter > 0),
    tap(() => console.log(counterOfItemsFetched)),
    takeWhile(counter => counter ===  100),
)
.subscribe(
    null,
    null,
    () => console.log('counter value at the end', counterOfItemsFetched)
)

【讨论】：