javascript - 将两个不同类 rxjs 的两个不同可观察值分组

Question

我有两个 Observables 发出 2 个不同类的数据:

1.用户类别:

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

2.UserWebsocket类

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

我想通过 user_id 或 id 属性合并和分组两个可观察量。这样合并分组类型的结果就是:

[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);
});