Rxjs - Слияние асинхронного ответа с ответом синхронизации

Question

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

я некоторые асинхронные данные использование value и объединить их с самим значением.

Позвольте мне показать вам код:

Rx.Observable.fromPromise(somePromise) // doesn't really matter 
    .flatMap(value => { // this is the original value I have 
    const promise = aFunctionThatReturnsPromise(value); 
    return Promise.all([promise, value])); // I created a promise using this value, but I still want to keep value as well 
    }) 
    .map(([promiseValue, originalValue]) => ({ // I merge new produced values with the value itself and get rid of array. 
    value1: promiseValue.value1, 
    value2: promiseValue.value2 
    originalValue: originalValue 
    })) 
    // ...rest of the chain... 

Как можно видеть, это означает, чтобы создать объединенный обещание с Promise.all просто держать value и фазы очистки с map, чтобы сделать вещи лучше для следующих частей цепи.

Есть ли встроенный Конструкция RxJs для этого типа рисунка? Или есть другой подход, который дает более понятное решение?

Answer 1

Если вы хотите сделать что-то с входным значением в сочетании с выходной стоимостью людей, как правило, пишут:

Rx.Observable.of('foo') 
    .mergeMap(i => aFunctionThatReturnsPromise(i) 
    .map(asyncResult => ({ foo: i, result: asyncResult}) 
) 

Эта модель может быть упрощена с помощью mergeMapresultSelector (flatMap если rxjs4) в следующее:

Rx.Observable.of('foo') 
    .mergeMap(
    i => aFunctionThatReturnsPromise(i), 
    (i, asyncResult) => ({ syncValue: i, asyncResult }) 
) 
// ... rest of the chain 

Это дает вам доступ к исходному входному значению и каждому полученному выходному значению. Дополнительный бонус заключается в том, что mergeMap обернет ваше обещание в наблюдаемое, не используя Rx.Observable.fromPromise().

Answer 2

Использования чистой RxJS, это общая для гнездовых наблюдаемых цепей сохранить предыдущие значения в контексте:

Rx.Observable.fromPromise(somePromise) 
    .flatMap(value => 
    Rx.Observable.fromPromise(aFunctionThatReturnsPromise(value)) 
     .map((promiseValue) => ({ 
     value1: promiseValue.value1, 
     value2: promiseValue.value2, 
     originalValue: value 
     })) 
) 
// ...rest of the chain... 

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