使用 Rxjs 构建 Angular Store

Posted 2023-03-16

【中文标题】使用 Rxjs 构建 Angular Store

【英文标题】：Build Angular Store with Rxjs

【发布时间】：2021-01-10 16:47:59

【问题描述】：

我有这个 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) => 行，因此应用程序不显示聊天列表。

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

【参考方案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 的最新值。

【讨论】：

谢谢！有效！另外，如果我想监听每个聊天对象的变化，我不应该使用take(1)，对吧？

@adrisons 没错，take(1) 只会传递一个值然后完成。