将两个不同类 rxjs 的两个不同 observable 分组

Question

我有两个 Observables 发射 2 个不同类的数据：

1.用户类：

{
"id"=>25,
"username"=>"Chris",
.. other data 
}

2.UserWebsocket类

 {
     "user_id"=>25,
     "age"=>25,
      .. other data 
  }

我想通过 user_id 或 id 属性合并和分组这两个 observables。这样合并和分组类型的结果是：

[User|undefined,UserWebsocket|undefined]

为了更好地说明问题，我做了一个简单的大理石图：

这是怎么发生的？

Answer 1

正如马丁的评论所建议的那样，使用 combineLatest 并为每个部分设置一个起始值。假设你有 myUser$ 和 myUserWebsocket$ observables，你可以使用下面的代码来组合它们。首先添加一个起始值并将可观察值存储在新值中。将它们与combineLatest 结合使用。你必须订阅 uncold combined$ observable。

更新

另一种方法是使用merge 并将其与withLatestFrom 结合使用。

// Using RxJS v6+
import { merge } from 'rxjs';
import { startWith, map, withLatestFrom } from 'rxjs/operators';

// Your allready defined observables named e.g. myUser$ and myUserWebsocket$
// Add a starting value, since if user fires, websocket has no value yet.
const user$ = myUser$.pipe(
  startWith(undefined)
)
const userWebsocket$ = myUserWebsocket$.pipe(
  startWith(undefined)
)

const combined$ = merge(
  user$,
  userWebsocket$
).pipe(
  withLatestFrom(user$),
  withLatestFrom(userWebsocket$),
  map(([[merge, latestUser], latestWebsocket]) => {
    if (merge.user_id) { // merge is typeof userWebsocket, make sure user does not have user_id property
      if (latestUser && merge.user_id === latestUser.id) {
        return [latestUser, merge];
      } else {
        return [undefined, merge];
      }
    } else if (merge.id && !merge.user_id) { // merge is typeof user
      if (latestWebsocket && merge.id === latestWebsocket.user_id) {
        return [merge, latestWebsocket];
      } else {
        return [merge, undefined];
      }
    }
  })
);

combined$.subscribe([user, websocket] => {
  console.log(user, websocket);
});