无限滚动的可观察分页以保持最新答案

【问题标题】：Observable Pagination for infinite scroll to stay up to date无限滚动的可观察分页以保持最新
【发布时间】：2021-03-22 03:17:08
【问题描述】：

我正在尝试让我的 observable 在分页期间自动更新。

仅页面触发器

当load()被调用时，页面observable被触发，但是订阅observable在更新时没有被触发......

pageSub = new BehaviorSubject<number>(1);
page = 1;

// in constructor...

this.posts = this.pageSub
  .pipe(
    mergeMap((page: number) => this.urql
      .subscription(this.GET_POSTS,
        {
          first: this.ITEMS.toString(),
          offset: (this.ITEMS * (page - 1)).toString()
        }
      )
    ),
    scan((acc, value) => [...acc, ...value]),
  );

// called from button

load() {
  this.page += 1;
  this.pageSub.next(this.page);
}

仅订阅触发器

这里确实触发了订阅更改，但是当调用 load() 时，不会触发页面更改...

this.posts = combineLatest([this.pageSub, this.urql
  .subscription(this.GET_POSTS,
    {
      first: this.ITEMS.toString(),
      offset: (this.ITEMS * (this.page - 1)).toString()
    }
  )]).pipe(
    map((arr: any[]) => arr[1])
  );

当然，有一种方法可以让 Observable 随着页面更改和订阅更改而更新？

我应该补充一点，重置变量声明 this.posts 不起作用，因为它刷新了页面并且不是预期的行为。

有什么想法吗？

谢谢，

【问题讨论】：

你能用小演示创建一个 stackblitz，因为据我所知，你的第一个解决方案应该可以工作。你能告诉我们`this.urql`背后是什么
如果我调用load()，第一个解决方案确实有效，但是，从技术上讲，它是一个可观察的页面而不是组合的可观察，因此当订阅更改时数据不会更新，只有当页面更改时.这有意义吗？
我不确定您所说的“订阅更改”是什么意思，以及当您引用 this.urql.subscrition(...) 时会调用什么
我有一个外部数据库。通常，如果我更新数据库值，它将立即在网站上更新，因为 observable 是使用 websockets 对数据库的订阅。但是，我希望这在使用分页时仍然有效。第一个示例实际上订阅了分页更改，但不订阅数据库更改。当触发页面更改时（由pageSub.next()），页面会更新。但是，当数据库值发生更改时，它不会立即在网站上更新。我希望在 db 值更改和 pageSub.next() 上触发 observable。

标签： rxjs pagination observable combinelatest mergemap

【解决方案1】：

知道了：

mainStream: BehaviorSubject<Observable<any>>;

posts!: Observable<any>;

constructor(private urql: UrqlModule) {

  this.mainStream = new BehaviorSubject<Observable<any>>(this.getPosts());

  this.posts = this.mainStream.pipe(mergeAll());

}

async load() {
  this.page += 1;
  this.mainStream.next(this.combine(this.mainStream.value, this.getPosts()));
}

getPosts() {
  return this.urql.subscription(this.GET_POSTS,
    {
      first: this.ITEMS.toString(),
      offset: (this.ITEMS * (this.page - 1)).toString()
    }
  )
}

combine(o: Observable<any>, o2: Observable<any>) {
  return combineLatest([o, o2]).pipe(
    map((arr: any) => arr.reduce((acc: any, cur: any) => acc.concat(cur)))
  );
}

【讨论】：