RxJs: форма zip-оператора с потерями

Следующее даст вам желаемое поведение:

Observable.zip(s1.take(1), s2.take(1)).repeat()

В синтаксисе RxJs 5.5+ pipe:

zip(s1.pipe(take(1)), s2.pipe(take(1))).pipe(repeat());

const s1 = new Rx.Subject();
const s2 = new Rx.Subject();

Rx.Observable.zip(s1.take(1), s2.take(1)).repeat()
    .subscribe(console.log);

s1.next(1); s1.next(2); s2.next(10); s1.next(3); s1.next(4); s2.next(20); s1.next(5); s1.next(6); s1.next(7); s2.next(30);  
s2.next(40); s2.next(50); s2.next(60); s2.next(70); 

<script src="https://unpkg.com/@reactivex/[email protected]/dist/global/Rx.js"></script>

Объяснение:

• Оператор repeat (в его текущей реализации) повторно подписывается на наблюдаемый источник по завершении последнего, т.е. в данном конкретном случае он повторно подписывается на zip при каждой взаимной эмиссии.
• zip объединяет два наблюдаемых объекта и ждет, пока они оба излучают. combineLatest тоже подойдет, это не имеет значения из-за take(1)
• take(1) на самом деле заботится о взрыве памяти и определяет поведение с потерями

Если вы хотите брать последнее значение вместо первого из каждого потока при взаимной эмиссии, используйте это:

Observable.combineLatest(s1, s2).take(1).repeat()

В синтаксисе RxJs 5.5+ pipe:

combineLatest(s1.pipe(take(1)), s2.pipe(take(1))).pipe(repeat());

const s1 = new Rx.Subject();
const s2 = new Rx.Subject();

Rx.Observable.combineLatest(s1,s2).take(1).repeat()
    .subscribe(console.log);

s1.next(1); s1.next(2); s2.next(10); s1.next(3); s1.next(4); s2.next(20); s1.next(5); s1.next(6); s1.next(7); s2.next(30);  
s2.next(40); s2.next(50); s2.next(60); s2.next(70); 

<script src="https://unpkg.com/@reactivex/[email protected]/dist/global/Rx.js"></script>

Это дает последовательность [0, 2] [1, 5] [2, 8] [3, 12] ...

const interval1 = Rx.Observable.interval(1000)
const interval2 = Rx.Observable.interval(300)

const combined = Rx.Observable.combineLatest(interval1, interval2);
const fresh = combined.scan((acc, x) => { 
    return x[0] === acc[0] || x[1] === acc[1] ? acc : x 
  })
  .distinctUntilChanged() //fresh ones only

fresh.subscribe(console.log);

с возможно меньшим количеством операторов. Не уверен, насколько это эффективно.
CodePen

Для обновления №3

Тогда вам понадобится ключ для каждого исходного элемента.

// Simulated sources according to latest spec provided (update #3)
const source1 = Rx.Observable.from(['x','y','z'])
const source2 = Rx.Observable.from(['a','a','b','b','c'])

// Create keys for sources
let key1 = 0
let key2 = 0
const keyed1 = source1.map(x => [x, key1++])
const keyed2 = source2.map(x => [x, key2++])

const combined = Rx.Observable
  .combineLatest(keyed1, keyed2)
  .map(([keyed1, keyed2]) => [...keyed1, ...keyed2]) // to simplify scan below
combined.subscribe(console.log) // not the output, for illustration only
console.log('-------------------------------------')

const fresh = combined.scan((acc, x) => { 
    return x[1] === acc[1] || x[3] === acc[3] ? acc : x 
  })
  .distinctUntilChanged() //fresh ones only

const dekeyed = fresh
  .map(keyed => { return [keyed[0], keyed[2]] })
dekeyed.subscribe(console.log); // required output

Это производит

["x", "a"]  
["y", "a"]  
["z", "b"]  

CodePen (обновите страницу CodePen после открытия консоли для лучшего отображения)

Я думаю, что следующее должно всегда принимать последнее значение из каждого источника Observable.

const source1 = Observable.interval(1000).publish();
const source2 = Observable.interval(300).publish();

source1.connect();
source2.connect();

Observable.defer(() => Observable.forkJoin(
        source1.takeUntil(source2.skipUntil(source1)),
        source2.takeUntil(source1.skipUntil(source2))
    ))
    .take(1)
    .repeat()
    .subscribe(console.log);

Живая демонстрация: http://jsbin.com/vawewew/11/edit?js,console

Это печатает:

[ 0, 2 ]
[ 1, 5 ]
[ 2, 8 ]
[ 3, 12 ]
[ 4, 15 ]
[ 5, 18 ]

Возможно, вам придется превратить source1 и source2 в горячие наблюдаемые объекты, если они еще не были.

Редактировать:

Основная часть - source1.takeUntil(source2.skipUntil(source1)). Принимает значения от source1 до source2. Но в то же время он будет игнорировать source1, пока source2 не выдаст хотя бы одно значение :).

forkJoin() Observable ожидает завершения работы обоих источников, запоминая последнее излучение от каждого из них.

Затем мы хотим повторить процесс и поэтому используем take(1) для завершения цепочки и .repeat() для немедленной повторной подписки.

Вы упоминаете размер буфера 1 и задаетесь вопросом, поможет ли это заархивировать два ReplaySubject с размером буфера 1?

Я добавляю еще один ответ для ясности, поскольку он идет после принятого ответа (но основывается на моем предыдущем ответе).

Простите меня, если я неправильно понял, но я ожидал, что решение будет обрабатывать переключение уровней выбросов:

затем я переключаюсь между их скоростью излучения,

Поставляемый тест не переключает скорость излучения до тех пор, пока после не остановится первый поток,

Stream1: 1 2    3 4    5 6 7                 
Stream2:     10     20    30 40 50 60 70

поэтому я попробовал еще один тест

Stream1: 1 2      3 4     5 6
Stream2:    10 20    30 40   50 60

Тестовые данные для этого потока

s1.next(1); s1.next(2); s2.next(10); s2.next(20); s1.next(3); s1.next(4);
s2.next(30); s2.next(40); s1.next(5); s1.next(6);  s2.next(50); s2.next(60);

Насколько я понимаю, принятый ответ не проходит этот тест.
Он выводит

[1, 10]
[3, 20]
[4, 30]
[5, 40]
[6, 50]

тогда как я ожидал увидеть

[1, 10]
[3, 30]
[5, 50]

если оператор должен быть симметричным (коммутативным?)

Улучшение моего предыдущего ответа

Это решение построено на основе базовых операторов, поэтому его легче понять. Я не могу говорить об его эффективности, возможно, проверим это в другой итерации.

const s1 = new Rx.Subject();
const s2 = new Rx.Subject();

const tagged1 = s1.map(x=>[x,1])
const tagged2 = s2.map(x=>[x,2])
const merged = tagged1.merge(tagged2)
const fresh = merged.scan((acc, x) => { 
    return x[1] === acc[1] ? acc : x 
  })
  .distinctUntilChanged() //fresh ones only
const dekeyed = fresh.map(keyed => keyed[0])
const paired = dekeyed.pairwise()
let index = 0
const sequenced = paired.map(x=>[x,index++])
const alternates = sequenced.filter(x => x[1] % 2 === 0)
const deindexed = alternates.map(x=>x[0])

или в более компактной форме, если желательно

let index = 0
const output = 
  s1.map(x=>[x,1]).merge(s2.map(x=>[x,2])) // key by stream id
  .scan((acc, x) => { 
    return x[1] === acc[1] ? acc : x 
  })
  .distinctUntilChanged()       //fresh ones only
  .map(keyed => keyed[0])       // de-key
  .pairwise()                   // pair
  .map(x=>[x,index++])          // add a sequence no
  .filter(x => x[1] % 2 === 0)  // take even sequence
  .map(x=>x[0])                 // deindex

Для тестирования CodePen (обновите страницу CodePen после открытия консоли для лучшего отображения)