尝试使用 RXJS 创建动态列表时遇到问题

Question

我正在尝试创建一个服务工作者，它将通过 API 向我提供有关用户的信息。

我的想法是我不想发出无用的请求，比如两次请求同一个用户，所以我试图实现一个函数来检查我们是否存储了用户数据，如果没有从服务器获取它.

我尝试使用mapTo、share、shareReplay等管道

users = new BehaviorSubject<Array<User>>([]);

getUserInfo(id: string) {
 return new Observable((sub) => {
   const users = this.users.getValue();
   let user = users.find((u) => u._id === id);

   if (user) {
     sub.next(user);
     sub.complete();
     return;
   }

   this.api.get('users/' + id, this.userService.getAuthToken(), ).then((data) => {
     user = new User(data, this);
     users.push(user);
     this.users.next(users);
     sub.next(user);
     sub.complete();
   }).catch((err) => {
     sub.error(err);
   });
 });
}

因此，当我使用相同的 id 同时调用 getUser 2 次时，它应该只请求一次。

Answer 1

要向行为主体中的数组添加值，您无需更改数组的内容，而是创建一个新数组，然后让行为主体发出该新数组。

this.users.next([..this.users.value, user]);

从一个值创建一个新的可观察对象，只需使用

of(user);

为什么你的 api 返回一个承诺？ http 服务返回 observables，因此您应该能够只返回映射到用户对象的 http。

return this.api.get('users/' + id, this.userService.getAuthToken(), ).pipe(
  map(functionToMapResponseToAUserObject),
  tap(sideEfectFunctionToInsertNewUserIntoBehaviorSubject)
)

因此，如果您修改 api 以返回可观察对象而不是承诺，则您的服务可能是

getUserInfo(id: string): Observable<User> {
  const users = this.users.getValue();
  const user = users.find((u) => u._id === id);
  return user
    ? of(user)
    : this.api.get('users/' + id, this.userService.getAuthToken(), ).pipe(
      map(response => new User(response)),
      tap(user => { this.users.next([...users, user]); })
    )
}

如果您无法修改 api 使用 from 将 promise 转换为 observable

from(this.api.get('users/' + id, this.userService.getAuthToken(), ))

要停止对同一个 id 的多个请求，您需要一个对象来跟踪正在加载的请求

loading: { [key]: Observable<User> } = {};

在 api 调用中添加共享重播，这样 api 调用就不会重复，并将其粘贴在加载缓存中

 let loaded = false;
 const user$ = this.api.get('users/' + id, this.userService.getAuthToken(), ).pipe(
    map(response => new User(response)),
    tap(user => {
      this.users.next([...users, user]);
      if (loading[id]) {
        delete loading[id];
      }
      loaded = true;
    }),
    shareReplay()
 );

 if (!loaded) {
   loading[id] = user$;
 }

然后您可以在触发 api 调用之前查看是否正在加载，并从加载缓存中返回 observable。

Answer 2

请尝试这种方法，我相信它会满足您的需求

usersSubject = new Subject<User>();
usersCache = this.usersSubject.pipe(
  scan<Record<string, User>>((cache, user) => {
    cache[user.id] = user;
    return cache;
  }, {}),
  shareReplay(1) // keeping the last version of the cache
);


getUserInfo(id: string): Observable<User> {
  return this.usersCache.pipe(
    pluck(id),
    switchMap(user => iif(() => !user,
      this.api.get('users/' + id, this.userService.getAuthToken()).pipe(
        tap(user => this.usersSubject.next(user)) // caching the user
      ),
      of(user)
    ),
    first() // auto unsubscription, because usersCache will keep pushing values
  )
}

更新： 确保保持订阅打开以保持缓存完好无损（我猜它是一个服务，所以在构造函数中这样做）：

this.usersCache.subscribe()

Answer 3

Adrian 的回答很接近，但没有完全回答您问题的一部分：

因此，当我使用相同的 id 同时调用 getUser 2 次时，它应该只请求一次。

Adrian 的解决方案没有这样做，因为如果请求尚未解决，用户将不会被存储，您最终会请求用户两次。

通常推荐的方法是将缓存存储为可观察对象的地图。

伪打字稿，语法可能是错误的，我也会假设您可以将 api 调用变成可观察的，如果不是，请寻找 Adrian 关于如何处理它的答案：

users: Map<string,Observable<User>> = new Map();


getUserInfo(id: string): Observable<User> {
  if(!this.users.get(id)){
     const userObservable = this.api.get('users/' + id, this.userService.getAuthToken(), ) // Or turn the promise to observable with `from`
       .pipe(
         map(response => new User(response)),
         share(),
       );
     this.users.set(id, userObservable) 
  }

  return this.users.get(id)

}