Rxjs. Вложенная условная наблюдаемая, которая делает паузу "главной" наблюдаемой

Question

У меня есть obserable:

const observable = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
})

, и я хотел бы подписаться таким образом:

observable.subscribe({
  next(x) { 
    console.log('got value ' + x); 
    if (x == 2) {
      const observable2 = new Observable(subscriber => {
       subscriber.next('a');
       subscriber.next('b');
       subscriber.next('c');
      })  
     observable2.subscribe(...)   
    }      
  },
  error(err) { console.error('something wrong occurred: ' + err); },
  complete() { console.log('done'); }
});

У меня есть Observable, который "выбрасывает" некоторые данные и, если данные равны некоторымконкретное значение мне нужно

1. Пауза основной Наблюдаемый

2. Подписаться на Наблюдаемый в блоке if

3. Получить некоторые данные из "условной" наблюдаемой

4. Резюме основной наблюдаемой

Answer 1

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

Итак, для этого конкретного случая, вот как я бы решил:

const source = of(1, 2, 3);

const subscription = source
  .pipe(
    /**
     * `Pause` the main observable
     * Won't to get the next emitted value until this inner subscription is done
     */
    concatMap(v => {
      // The if block
      if (v === 2) {
        // Getting some data from the condition
        return of('a', 'b', 'c');
      }

      return of('default value');
    }),
  )
  .subscribe(v => console.log('value received!', v))

concatMap является одним из операторов высокого порядка .

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

StackBlitz .

Редактировать после прочтения первого комментария

const microS1$ = ...;

microS1$
 .pipe(
  // Insert your condition here
  filter(content => typeof content === 'string')
 )
 .subscribe(d => {
  // Send the data to the second microservice
  microserviceTwo.sendData(d);
 })