여러 번 관찰 할 수있는 ReactiveX 필터링 및 병합

Question

다음과 같은 관찰 가능 항목을 만드는 데 문제가 있습니다.
미리 정의 된 값 배열을 받길 원합니다.
다른 것들로 필터링하고 개별 관찰 가능 항목으로 작업 할 수 있습니다. 는 이러한 필터링 관찰 가능한 병합하는 데 시간이 올 때
그리고는, 나는 원래

//Not sure the share is necessary, just thought it would tie it all together 
const input$ = Observable.from([0,1,0,1]).share(); 
const ones$ = input$.filter(n => n == 1); 
const zeroes$ = input$.filter(n => n == 0); 

const zeroesChanged$ = zeroes$.mapTo(2); 
const onesChanged$ = ones$.mapTo(3); 
const allValues$ = Observable.merge(onesChanged$,zeroesChanged$); 

allValues$.subscribe(n => console.log(n)); 
//Outputs 3,3,2,2 
//Expected output 3,2,3,2 

편집에서 순서를 유지하려면 : 내가 내 질문에 충분한 특정되지 않았습니다 죄송합니다. 부작용을 드라이버로 구분하는 cycleJS라는 라이브러리를 사용하고 있습니다. 그래서 내가 내주기에서 뭐하는 거지 것은 내가 그것을 위해 테스트를 작성하고 싶어 할 때이

export function socketCycle({ SOCKETIO }) { 
    const serverConnect$ = SOCKETIO.get('connect').map(serverDidConnect); 
    const serverDisconnect$ = SOCKETIO.get('disconnect').map(serverDidDisconnect); 
    const serverFailedToConnect$ = SOCKETIO.get('connect_failed').map(serverFailedToConnect); 
    return { ACTION: Observable.merge(serverConnect$, serverDisconnect$, serverFailedToConnect$) }; 
} 

지금 내 문제가 발생합니다. 나는 잘못된 문제 (농담을 사용)에서 일한 다음과 함께 시도했다.

const inputConnect$ = Observable.from(['connect', 'disconnect', 'connect', 'disconnect']).share(); 
const expectedOutput$ = Observable.from([ 
    serverDidConnect(), 
    serverDidDisconnect(), 
    serverDidConnect(), 
    serverDidDisconnect(), 
]); 
const socketIOMock = { 
    get: (evt) => { 
    if (evt === 'connect') { 
     return inputConnect$.filter(s => s === 'connect'); 
    } else if (evt === 'disconnect') { 
     return inputConnect$.filter(s => s === 'disconnect'); 
    } 
    return Observable.empty(); 
    }, 
}; 
const { ACTION } = socketCycle({ SOCKETIO: socketIOMock }); 
Observable.zip(ACTION, expectedOutput$).subscribe(
    ([output, expectedOutput]) => { expect(output).toEqual(expectedOutput); }, 
    (error) => { expect(true).toBe(false) }, 
() => { done(); }, 
); 

어쩌면 내가 테스트 할 수있는 또 다른 방법이 있을까? 코드 아래

Answer 1

는 다른 딸의 요소 사이의 타이밍을 보장 스트림 실제로 배열을 조작 할 rxjs를 사용할 필요 없습니다 의해 파괴됨. 특히, connect 이벤트가 항상 이벤트 소스에서 disconnect 이벤트 이전에 오더라도, connect Observable의 이벤트가 항상 해당 이벤트 항목 인 disconnect Observable보다 먼저 나오는 것은 아닙니다. 정상적인 시간대에서는이 레이스 컨디션은 매우 드물 긴하지만 위험 할 수 있습니다.이 테스트는 최악의 경우를 보여줍니다.

표시된 좋은 기능은 이벤트와 처리기의 결과 사이의 매퍼 (mapper)에 불과하다는 것입니다.

const event_handlers = new Map({ 
'connect': serverDidConnect, 
'disconnect': serverDidDisconnect, 
'connect_failed': serverFailedToConnect 
}); 
const ACTION = input$.map(event_handlers.get.bind(event_handlers)); 

주의 할 : 당신이 이벤트 유형을 통해 일반적으로이 모델을 계속 할 수있는 경우에, 당신은 심지어 표현력 혜택을 일반 데이터 구조에 매핑을 인코딩 할 수 있습니다 당신이 딸 스트림을 통해 감소 된 경우 (또는 다른 이전 값 (예 : debounceTime)을 고려하면 리팩터링은 그리 간단하지 않으며 "보존 순서"라는 새로운 정의에 의존합니다. 대개의 경우, reduce + 더 복잡한 누적기로 재생산하는 것은 여전히 ​​가능할 것입니다.

Answer 2

당신에게 욕망 결과를 줄 수있을 수도 있지만 이럴 스트림이 분할되면

Rx.Observable.combineLatest(
    Rx.Observable.from([0,0,0]), 
    Rx.Observable.from([1,1,1]) 
).flatMap(value=>Rx.Observable.from(value)) 
.subscribe(console.log)