RxJS Подписаться с двумя аргументами

Question

У меня есть две наблюдаемые, которые я хочу объединить и в подписке использовать либо оба аргумента, либо только один.Я пробовал .ForkJoin, .merge, .concat, но не смог добиться нужного мне поведения.

Пример:

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return obs1.concat(obs2);
}

Затем при использовании этой функции:

service.save().subscribe((first, second) => {
    console.log(first); // int e.g. 1000
    console.log(second); // Boolean, e.g. true
});

или

service.save().subscribe((first) => {
    console.log(first); // int e.g. 1000
});

Есть ли возможность получить именно такое поведение?

Надеюсь, кто-то может помочь!

РЕДАКТИРОВАТЬ:

В моем конкретном случае использования obs1<int> и obs2<bool> - это два разных почтовых запроса: obs1<int> - это действительная функция сохранения, а obs2<bool> проверяет, запущена ли другая служба.

значение obs1<int> необходимо для перезагрузки страницы после завершения запроса, а значение obs2<bool> необходимо для отображения сообщения, если служба запущена - независимо от obs1<int>.

Так что если obs2<bool> испускается до obs1<int>, это не проблема, сообщение отображается перед перезагрузкой.Но если obs1<int> испускается до obs2<bool>, страница перезагружается, и сообщение может больше не отображаться.

Я говорю это, потому что с данными ответами существуют различные способы поведения, переданные значения доили после onComplete другой наблюдаемой, и это может повлиять на вариант использования.

Answer 1

Вы можете использовать forkJoin для этой цели.Назовите их параллельно, а затем, если кто-то из них присутствует, сделайте что-нибудь.

let numberSource = Rx.Observable.of(100);
let booleanSource = Rx.Observable.of(true);

Rx.Observable.forkJoin(
  numberSource,
  booleanSource
).subscribe( ([numberResp, booleanResp]) => {
  if (numberResp) {
    console.log(numberResp);
    // do something
  } else if (booleanResp) {
    console.log(booleanResp);
    // do something
  }
});

Answer 2

Есть несколько операторов, которые выполняют это:

CombineLatest

Этот оператор объединяет последние значения, излучаемые обеими наблюдаемыми, как показано на мраморной диаграмме:

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return combineLatest(obs1, obs2);
}

save().subscribe((val1, val2) => { 
   // logic
});

Zip

Оператор Zip будет ожидать, пока обе наблюдаемые передадут значения, прежде чем испустить одну.

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return zip(obs1, obs2);
}

save().subscribe((vals) => { 
   // Note Vals = [val1, val2]
   // Logic
});

Или если вы хотите использовать деструктуризацию с массивом

save().subscribe(([val1, val2]) => {
   // Logic
});

WithLatestFrom

WithLatestFrom испускает комбинациюпоследние значения, испускаемые наблюдаемыми, обратите внимание, что этот оператор пропускает любые значения, которые не имеют соответствующего значения из других наблюдаемых.

save: obs1.pipe(withLatestFrom(secondSource))

save().subscribe(([val1, val2]) => {
    // Logic
});

Answer 3

Точный используемый оператор зависит от конкретных деталей того, что вы пытаетесь решить.

Допустимый вариант - использовать combineLatest - Документы :

obs1$: Observable<int>;
obs2$: Observable<Boolean>;
combined$ = combineLatest(obs1$, obs2$);

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

Answer 4

Concat испускает два потока через поток, одно за другим завершено, это не то, что вам нужно.

Слияние будет генерировать оба события одинаковым образом, но в том порядке, в котором они действительно завершатся, а не в том, что вам нужно.

То, что вы хотите, это значение обоих элементов в одном и том же поточном событии.Это будут делать forkJoin, zip и integraLatest, когда вы запутываетесь в том, что все они выдают массив значений, к которым вы не обращаетесь должным образом при подписке.

zip испускает каждый раз, когда все элементы, упакованные вместе, испускают последовательно, поэтому, если наблюдаемый 1 испускает 1,2,3, а наблюдаемый два испускает 4,5;выбросы из почтового индекса будут [1,4], [2,5].

Объединение последнего будет излучать каждый раз, когда излучается, поэтому вы получите что-то вроде [1,4], [2,4], [2,5], [3,5] (в зависимости от точного порядка излучения).

Наконец, forkJoin испускает только один раз, после того, как каждый элемент внутри него фактически завершен, a и затем завершает себя.Это, вероятно, то, что вы хотите больше всего на свете, так как вы, похоже, «экономите».если ни один из этих примеров потоков не завершится, forkJoin никогда не выдаст, но если они оба завершат работу после своего окончательного значения, forkjoin выдаст только одну эмиссию: [2,5].Я предпочитаю это, поскольку это «самая безопасная» операция, которая гарантирует, что все потоки завершаются правильно и не создают утечек памяти.И обычно при «сохранении» вы ожидаете только одну эмиссию, поэтому она также более явная.Когда бы вы ни увидели forkJoin, вы знаете, что имеете дело с одним потоком эмиссии.

Лично я бы сделал это следующим образом:

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return forkJoin(obs1, obs2);
}

service.save().subscribe(([first, second]) => {
    console.log(first); // int e.g. 1000
    console.log(second); // Boolean, e.g. true
});

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

Answer 5

Вы можете использовать статический метод zip вместо оператора concat.

save(): Observable<any> {
   return zip(obs1, obs2);
}

Тогда вы сможете сделать следующее:

service.save().subscribe((x) => {
    console.log(x[0]); // int e.g. 1000
    console.log(x[1]); // Boolean, e.g. true
});