Concat Наблюдаемые

Question

У меня есть некоторые функции для работы с БД:

setupData(param) { 
    return Observable.create((observer) => { 
     this.db.executeSql('SELECT .... ?', param).then(() => { 
      console.log('DB QUERY DONE'); 
      observer.complete(); 
     }, (error) => { 
      observer.error(error); 
     }); 
    }); 
} 

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

processData() { 

    ... 

    let observers = []; 
    storage.forEach((data) => { 
     observers.push(this.setupData(data)); 
    }); 

    Observable.concat(observers).subscribe(() => { 
     console.log('NEXT'); 
    },() => { 
     console.log('ERROR'); 
    },() => { 
     console.log('COMPLETE'); 
    }); 
} 

Так что мой выход:

СЛЕДУЮЩИЙ
Вперед Вперед

...
СЛЕДУЮЩИЙ
СЛЕДУЮЩИЙ
ПОЛНОЕ

, но я никогда не видел "DB QUERY DONE". На каждого наблюдаемого нет подписки. Если я заменю concat на forkJoin - я вижу, что я ожидаю, но мне нужно запустить SEQUENTIALLY, а не параллельно ... Есть ли какое-то решение?

Answer 1

Что вам действительно нужно - concatMap. Вы можете выполнять этот оператор на наблюдаемом, и что он будет делать.

1. Примите следующее значение.

2. Выполните функцию над этим значением, которое должно вернуть наблюдаемое.

3. Начните слушать это наблюдаемое и отправить значения от этого наблюдаемого до следующего.

4. Как только это наблюдаемое прекратило излучать, сделайте следующее значение, чтобы источник наблюдался и повторялся.

код мудрым это выглядит следующим образом:

const processData =() => { 
    let storageIds$ = Rx.Observable.from(["1", "2", "3"]); 

    storageIds$.concatMap((val) => { 
     return setupData(val); 
    }).subscribe(() => { 
     console.log('NEXT'); 
    },() => { 
     console.log('ERROR'); 
    },() => { 
     console.log('COMPLETE'); 
    }); 
} 

Вы можете видеть, что я создал наблюдаемую storageIds$ из чисел 1, 2, 3. Тогда я выполняю concatMap над этим наблюдаемым. Я просто вызываю метод setupData, который возвращает наблюдаемый, и немедленно возвращаю его. Это обеспечит вам ожидаемое поведение.

Полный jsbin пример можно найти здесь: http://jsbin.com/katame/edit?js,console