Более простой / более идиоматический способ сделать "почтовый индекс с условным" в RXJS?

Question

Я пытаюсь сжать два Observable в RXJS, принимая значения каждого в парах, когда выполняется определенное условие.Я нашел способ сделать это без использования zip, но мне было интересно, знает ли кто-нибудь более идиоматический способ сделать это.

const { zip, merge, Subject } = require("rxjs");
const rxop = require("rxjs/operators");

const a$ = new Subject();
const b$ = new Subject();

// Zipping a$ and b$ on conditional a.n === b.n
const zippedWithCondition$ = merge(
  a$.pipe(
    rxop.mergeMap(aMsg => b$.pipe(
      rxop.find(bMsg => aMsg.n === bMsg.n),
      rxop.map(bMsg => [aMsg, bMsg])
    ))
  ),
  b$.pipe(
    rxop.mergeMap(bMsg => a$.pipe(
      rxop.find(aMsg => aMsg.n === bMsg.n),
      rxop.map(aMsg => [aMsg, bMsg])
    ))
  )
);
const withConditionSub = zippedWithCondition$.subscribe(msg => {
  console.log("[ZIPPED WITH CONDITION]", msg);
});
a$.next({n: 0, type: "a"});
b$.next({n: 1, type: "b"});
a$.next({n: 1, type: "a"});
a$.next({n: 2, type: "a"});
b$.next({n: 2, type: "b"});
b$.next({n: 0, type: "b"});
a$.next({n: 3, type: "a"});
b$.next({n: 3, type: "b"});
withConditionSub.unsubscribe();

// Zipping a$ and b$ without a conditional
const normalZipped$ = zip(a$, b$);
const normalZippedSub = normalZipped$.subscribe(msg => {
  console.log("[NORMAL ZIP]", msg);
});
a$.next({n: 0, type: "a"}); // same order as above
b$.next({n: 1, type: "b"});
a$.next({n: 1, type: "a"});
a$.next({n: 2, type: "a"});
b$.next({n: 2, type: "b"});
b$.next({n: 0, type: "b"});
a$.next({n: 3, type: "a"});
b$.next({n: 3, type: "b"});
normalZippedSub.unsubscribe();

Вывод:

[ZIPPED WITH CONDITION] [ { n: 1, type: 'a' }, { n: 1, type: 'b' } ]
[ZIPPED WITH CONDITION] [ { n: 2, type: 'a' }, { n: 2, type: 'b' } ]
[ZIPPED WITH CONDITION] [ { n: 0, type: 'a' }, { n: 0, type: 'b' } ]
[ZIPPED WITH CONDITION] [ { n: 3, type: 'a' }, { n: 3, type: 'b' } ]

[NORMAL ZIP] [ { n: 0, type: 'a' }, { n: 1, type: 'b' } ]
[NORMAL ZIP] [ { n: 1, type: 'a' }, { n: 2, type: 'b' } ]
[NORMAL ZIP] [ { n: 2, type: 'a' }, { n: 0, type: 'b' } ]
[NORMAL ZIP] [ { n: 3, type: 'a' }, { n: 3, type: 'b' } ]

См.как для моего [ZIPPED WITH CONDITION] совпадения n для каждой пары.Для обычного почтового индекса этого не происходит, потому что сообщения пришли не в порядке.

Так есть ли лучший способ сделать это?Может быть, таким способом, который расширяет любое количество наблюдаемых при условии?

Answer 1

В своем решении вы создаете 2n + 2m наблюдаемых.Другими словами, если a$ испускает m значений и b$ испускает n значений, вы создаете 2n + 2m наблюдаемых.

Чтобы увидеть это, вам просто нужно добавить complete методы к a$ и b$ к вашей последовательности тестовых данных, как в следующем примере

a$.next({n: 0, type: "a"});
b$.next({n: 1, type: "b"});
a$.next({n: 1, type: "a"});
a$.next({n: 2, type: "a"});
b$.next({n: 2, type: "b"});
b$.next({n: 0, type: "b"});
a$.next({n: 3, type: "a"});
b$.next({n: 3, type: "b"});
a$.complete();
b$.complete();
withConditionSub.unsubscribe();

Я думаю,что для того, чтобы найти хорошее решение, вы должны выполнить свои требования, учитывая также переменную time .Например: «a$ и b$ испускают всегда объекты со свойством n, увеличенным на 1 последовательно - никаких дырок в последовательной последовательности не предвидится - и т. Д. И т. Д.

Если требования, связанные со временемне может быть указан, трудно представить решение RxJS