使用 Rxjs 构建 Angular Store

Question

我有这个 Angular 应用程序，我想在其中为聊天对象创建自定义商店。

（代码沙盒：https://codesandbox.io/s/chat-store-yw4yz）

注意：我知道有像 ngrx 这样的工具，但这是一个更大项目的一部分，我更喜欢使用这个自定义实现，以便更好地了解商店的行为。

上下文

使其快速高效的想法是在存储中存储两种类型的记录：

• 带有key = chat id 和value = chat object 的字典（作为地图）
• 按groupId 对聊天进行分组的索引（作为地图），其中key = groupId 和value = chatId

所以我有这个接口定义：

export interface ChatStoreModel {
  chats: Map<number, Chat>;
  chatsByGroup: Map<number, number[]>;
}

当聊天对象被保存到 Store 中时，会发生两件事：

• 聊天对象被保存到存储chats
• 聊天 id 保存在相应的 groupId 中的 store chatsByGroup

问题

无法从群组中获取聊天对象。我执行以下操作：

• 使用 groupId 从商店 chatsByGroup 获取聊天 ID 列表
• 对于每个聊天ID，从商店chats获取元素

代码：

getGroupChats$(groupId: number): Observable<Chat[]> {
    return this._store.select("chatsByGroup").pipe(
      map((chatsByGroup) => chatsByGroup.get(groupId) || []),
      switchMap((ids) => {
        if (!ids || !ids.length) {
          return of([]);
        }
        const chatsList$: Observable<Chat>[] = ids.map((id) =>
          this._store.select("chats").pipe(map((chats) => chats.get(id)))
        );
        return forkJoin([...chatsList$]).pipe(
          map((list) => {
            return ([] as Chat[]).concat(...list);
          })
        );
      }),
      shareReplay(1)
    );
  }

调试此代码，它到达return forkJoin 行，并且 ids 有聊天列表，但它从未到达map((list) => { 行，因此应用程序不显示聊天列表。

任何帮助将不胜感激，谢谢！

Answer 1

每当任何可观察对象完成但没有发出任何值时， forkJoin 也将在那个时候完成并且它不会发出 什么都可以

这是来自RxJS ForkJoin Doc。我的猜测是，您传递给 forkJoin 的 Observable 之一完成后没有发出任何东西，因此 forkJoin 立即完成而不发出任何东西。

尝试使用defaultIfEmpty 运算符以确保不会发生这种情况。我建议这样做：

const chatsList$: Observable<Chat>[] = ids.map((id) =>
  this._store.select("chats").pipe(map((chats) => chats.get(id))),
  defaultIfEmpty(null),
);

---

forkJoin 将等待所有通过的 observables 完成

这又来自文档。这意味着您传递给 forkJoin 的所有 Observable 都必须完成，以便 forkJoin 自己发出值。

我的猜测是this._store.select("chats").pipe(map((chats) => chats.get(id))), 这条线会无限运行，所以你需要提供take(1) 才能完成它，否则forkJoin 不会发出任何东西。所以尝试这样的事情：

const chatsList$: Observable<Chat>[] = ids.map((id) =>
  this._store.select("chats").pipe(map((chats) => chats.get(id))),
  take(1),
  defaultIfEmpty(null),
);

---

最后，如果你不想使用take(1)，我的意思是你想订阅 Observable，只要它发出值，你可能想使用combineLatest 而不是forkJoin。那是因为forkJoin 等待完成发出一些东西，而combineLatest 将发出提供的 Observables 的最新值。