typescript - 使用 rxjs5 在 typescript 中构建无限滚动列表

Question

我正在尝试使用 TypeScript 和 rxjs 构建一个无限滚动的列表。也就是说，我希望应用程序从后端获取几页结果，然后在用户滚动到底部附近时获取更多结果。

我有一个 Observable，用 Observable.prototype.expand() 构建，这将给我所有结果，最终从服务器获取所有页面。但是由于 Observable 的性质，我不能暂停这个过程。一旦我订阅，它不可避免地会尽快获取所有结果。我需要一个不同的解决方案，以我需要的速度从结果流中提取结果。

事情变得更加复杂，因为我无法直接从 API 获取第二页，每个页面都包含我获取下一页所需的信息。回复看起来像这样:

interface GraphApiResponse {
    data?: any[];
    paging?: {
        cursors: {
            before: string,
            after: string
        },
        next?: string,
        previous?: string
    };
}

paging.next 的存在表示还有另一页 paging.cursors.after用于实际检索它。

我似乎无法弄清楚如何在不弄乱它的情况下实现它。然而，无限列表似乎是一个常见的问题，不太可能没有好的解决方案。我应该如何着手实现它，而不会使事情变得一团糟？

我尝试过的事情

可迭代的 promise

我的第一个想法是使用一个可迭代的 Promises，但是我不知道我会得到多少结果，迫使我构建一个无限的 Iterable<Promise<Response?>>谁的 Promises 都将解析为 undefined在某一点之后。然而，由于它是无限的，我无法正常迭代它(它会用 Promises 填充整个可用内存)，实际上在结果处于那种形式时使用结果意味着在前一个的 resolve 函数中获得每个 Promise。

这个解决方案似乎可行，但随着我编写的每一行，它的可读性越来越差，而且越来越复杂。

将其与行为主体合并

在谷歌上搜索这个问题时，我发现了 a related SO question以及a GitHub issue on rxjs backpressure ，两者都包含 Ben Lesh 的代码片段，显然可以用来向 Observable 添加背压，遗憾的是，无论我尝试什么，我都无法让源 Observable 发射它的值比它生成它们的速度慢，它们总是被缓冲某处，这意味着网络请求无论如何都会发生。

来自 GitHub:

// this behavior subject is basically your "give me the next batch" mechanism.
// in this example, we're going to make 5 async requests back to back before requesting more.
const BATCH_SIZE = 5;
const requests = new BehaviorSubject(BATCH_SIZE); // start by requesting five items

// for every request, pump out a stream of events that represent how many you have left to fulfill
requests.flatMap((count) => Observable.range(0, count).map(n => count - n - 1))
  // then concat map that into an observable of what you want to control with backpressure
  // you might have some parameterization here you need to handle, this example is simplified
  // handle side effects with a `do` block
  .concatMap(() => getSomeObservableOfDataHere().do(stuffWithIt), (remaining) => remaining)
  // narrow it down to when there are no more left to request,
  // and pump another batch request into the BehaviorSubject
  .filter(remaining => remaining === 0)
  .mapTo(BATCH_SIZE)
  .subscribe(requests);

来自 StackOverflow:

// start with 5 values
const controller = new Rx.BehaviorSubject(5);

// some observable source, in this case, an interval.
const source = Rx.Observable.interval(100)

const controlled = controller.flatMap(
      // map your count into a set of values
      (count) => source.take(count), 
      // additional mapping for metadata about when the block is done
      (count, value, _, index) => {
        return { value: value, done: count - index === 1 }; 
      })
      // when the block is done, request 5 more.
      .do(({done}) => done && controller.next(5))
      // we only care about the value for output
      .map(({value}) => value);


// start our subscription
controlled.subscribe(x => {
  console.log(x)
});

我对此可能是错误的，但在我看来，一旦我订阅了一个 Observable，它就会尽可能快地产生它的值，没有办法减慢它的速度，所以这可能不是一个解决方案。

使用 ixjs

似乎ixjs旨在解决我的问题，但是该存储库已经很长时间没有更新了。显然有 a reimplementation in TypeScript ，但是这似乎处于早期开发阶段并且没有很好地记录 jet。

对于实际上非常简单的问题，我宁愿不依赖于少数人使用的框架。

重构应用

我在网上搜索了 TypeScript(使用 Angular)中无限滚动列表的实现。我当前的方法是拥有一个服务，该服务提供一个对象，可用于获取所有结果。然后我有一个显示它们的组件。替代品似乎是 doing the checking for scroll position right in the service that queries the backend , 或 having the component fetch a new Observable from the backend service when the user scrolls .

这两种解决方案都会迫使我混合使用目前整齐分开的代码。我更希望让服务返回一些东西，我可以将这些东西提供给组件，而组件不必知道网络请求，或者服务不必知道滚动位置。

Answer 1

我建议您查看 mergeScan运算符代替。看起来它可能很适合这里。

MergeScan 类似于 expand 运算符，因为它将来自先前请求的数据作为累加器返回，但与 expand 不同直到时间结束才继续运行。

基本上假设您有一个函数makeRequest(params)，它接受一个请求并返回一个Observable，最终解析为一个响应，一个流表示滚动事件，我们将调用 fetchMore$，您可以像这样创建按需获取服务:

// This is abstracted as a simple "fetch" concept but in reality should
// be hooked up to your scroll handler, properly debounced etc.
this.fetchMore$
  .mergeScan(
    // Make the request
    (acc, _) => makeRequest(acc.paging.next ? acc.paging.cursors.after : ''),
    {paging: {}}, // Initial request body
    1 // Maximum concurrency, i.e. how many requests can be in flight at once 
  )
  .pluck('data')
  .subscribe(data => {/*Do something with the data*/});

我将并发设置为 1，因为虽然您可能有多个请求在进行中，但目前无法保证顺序，因此如果用户滚动得非常快，结果可能是 acc 不同步，而对于并发度为 1 时，数据将始终有序。