사슬 관측 가능한 배출량?

Question

두 개의 독립적 인 Observable이 있는데, 첫 번째 값을 내릴 때마다 두 번째 값을 내야하는 방식으로 결합하고 싶습니다.

이 시나리오에서 첫번째 Observable은 가변 콜백 타이밍 (delay 또는 timer의 사용이 허용되지 않음)이있는 비동기 Node.js 스타일의 메서드를 기반으로하는 무한 시퀀스입니다. 두 번째 Observable은 독립적 인 값의 유한 순서입니다.

아마도도 I 바라는 어떤 표현할 수

I: ---A--D-----Q-G---B-> (I is infinite) 
F: -3512-|->    (F is finite) 
O: ---3--5-----1-2-|-> (O is finite) 

출력 시퀀스 (O)는 (I)의 타이밍에 기초하여의 값 (F)로 한정된다.

은 여기 (I) :

// I'm stuck with this, it cannot change... 
function asyncOperation(callback) { 
    const delayMsec = Math.floor(Math.random() * 1000); 
    timers.setTimeout(callback, delayMsec, null, delayMsec); 
} 

const callbackObservable = Rx.Observable.bindNodeCallback(asyncOperation); 

const infiniteCallbackSequence = Rx.Observable 
    .defer(callbackObservable) 
    .expand(x => callbackObservable()); 

그리고 편의를 위해

, 여기 (F) :

const finiteSequence = Rx.Observable.from([3, 5, 1, 2]); 

I 돈 (내 here의 이전 질문에 따라이 빌드.) combine*, merge* 또는 *map이 내가 원하는 것을 생성 할 수 있도록 충분히 잘 이해하고 있습니다. 그것은 (O)가 방출 될 때마다 (I)에 take(1)을 원하는 것처럼 보입니다.

설명 된 동작을 어떻게 얻을 수 있습니까?

Answer 1

Observable.zip은 사용하기에 적합합니까?

const Rx = require('rxjs'); 
const timers = require('timers'); 

const finite = Rx.Observable.from([3, 5, 1, 2]); 

// I'm stuck with this, it cannot change... 
function asyncOperation(callback) { 
    const delayMsec = Math.floor(Math.random() * 1000); 
    timers.setTimeout(callback, delayMsec, null, delayMsec); 
} 

const callbackObservable = Rx.Observable.bindNodeCallback(asyncOperation); 

const infiniteCallbackSequence = Rx 
    .Observable 
    .defer(callbackObservable) 
    .expand(x => callbackObservable()); 

const combined = Rx.Observable.zip(
    finite, 
    infiniteCallbackSequence 
).forEach(v => console.log(v)); 

노드 6에서 실행 :

➜ $ node test.js 
[ 3, 816 ] 
[ 5, 297 ] 
[ 1, 95 ] 
[ 2, 677 ] 
➜ $