javascript - RxJS:将源 Observable 划分为 3 个或更多 Observable 的优雅方式

Question

我有一个套接字连接，它发出带有标识符的消息。我想为每种类型的消息创建一个单独的可观察对象。我有一些解决方案，但它们都感觉笨拙或可能存在性能问题。我对 RxJS 比较陌生，所以我不知道我可能会走进的陷阱。

我的第一直觉是为每种类型创建一个过滤的可观察对象:

const receive_message = Rx.fromEvent(socket, 'data').pipe(share());
const message_type_a = receive_message.pipe(filter(message => message.type === 'a'));
const message_type_b = receive_message.pipe(filter(message => message.type === 'b'));
const message_type_c = receive_message.pipe(filter(message => message.type === 'c'));
const message_type_d = receive_message.pipe(filter(message => message.type === 'd'));

我认为这会导致性能问题，因为它会在每次收到任何消息时对每种消息类型执行此检查。

我考虑过像这样进行多级分区:

const receive_message = Rx.fromEvent(socket, 'data');
const [message_type_a, not_a] = receive_message.pipe(partition(message => message.type === 'a'));
const [message_type_b, not_b] = not_a.pipe(partition(message => message.type === 'b'));
const [message_type_c, message_type_d] = not_b.pipe(partition(message => message.type === 'c'));

这非常笨拙，我不确定它是否比过滤器解决方案更高效。

接下来我尝试使用这样的主题:

const message_type_a = new Rx.Subject();
const message_type_b = new Rx.Subject();
const message_type_c = new Rx.Subject();
const message_type_d = new Rx.Subject();

Rx.fromEvent(socket, 'data').subscribe(function (message) {
    switch (message.type) {
      case 'a':
        message_type_a.next(message);
        break;
      case 'b':
        message_type_b.next(message);
        break;
      case 'c':
        message_type_c.next(message);
        break;
      case 'd':
        message_type_d.next(message);
        break;
      default:
        console.log('Uh oh');
    }
  },
  console.log,
  function () {
    message_type_a.complete();
    message_type_b.complete();
    message_type_c.complete();
    message_type_d.complete();
  }
);

同样，这很笨拙，每当我使用主题时，我都会问自己这是否是“Rx”做事的方式。

理想情况下我可以做这样的事情:

const [
  message_type_a,
  message_type_b,
  message_type_c,
  message_type_d
] = Rx.fromEvent(socket, 'data').pipe(partitionMany(message.type));

是否有任何优雅的解决方案，或者我像这样拆分可观察到的源的总体方法是否存在根本性缺陷？

这是我的第一个问题，所以我希望我做得很好。提前致谢!

Answer 1

我将您的 switch case 解决方案更改为性能更高的解决方案。

const message_type_a = new Rx.Subject();
const message_type_b = new Rx.Subject();
const message_type_c = new Rx.Subject();
const message_type_d = new Rx.Subject();

subjects = {
    'a': message_type_a,
    'b': message_type_b,
    'c': message_type_c,
    'd': message_type_d
}

Rx.fromEvent(socket, 'data').pipe(tap(message => 
subjects[message.type].next(message))).subscribe();