관찰 가능한 단일 스트림에서 개체를 조건부로 병합하는 방법은 무엇입니까?

Question

스트림을 다음 개체 경우

const data = [ 
    { type: 'gps', id: 1, val: 1 }, 
    { type: 'gps', id: 2, val: 2 }, 
    { type: 'speed', id: 2, val: 3 }, 
    { type: 'gps', id: 3, val: 4 }, 
    { type: 'speed', id: 4, val: 5 }, 
    { type: 'gps', id: 4, val: 6 }, 
    { type: 'gps', id: 5, val: 7 } 
] 

IDS는이 객체 병합, 동일를 constains. 어떤 ID가 일치하지 않는 경우, 객체는 무시됩니다 :

[ 
    [{type: 'gps', id:2, val:2}, { type: 'speed', id: 2, val: 3 }], 
    [{ type: 'speed', id: 4, val: 5 },{ type: 'gps', id: 4, val: 6 }] 
] 

내 생각이 그룹에 있던 같은 종류의 개체, 두 개의 새로운 스트림

Rx.Observable.from(data) 
    .groupBy((x) => x.type) 
    .flatMap((g) => ...) 
    .... 

다음과 결말 다시 압축/병합 id이 동일한 경우

Rx에서이를 지정하는 방법을 잘 모르겠습니다. 이것이 좋은 접근 방법인지 확실하지 않습니다.

Answer 1

스트림을 분할하고 다시 병합 할 필요가 없습니다. 당신은 상태로 객체와 부합하지 않는 것들을 밖으로 filter를 수집하는 scan을 사용할 수 있습니다

const data = [ 
 
    { type: 'gps', id: 1, val: 1 }, 
 
    { type: 'gps', id: 2, val: 2 }, 
 
    { type: 'speed', id: 2, val: 3 }, 
 
    { type: 'gps', id: 3, val: 4 }, 
 
    { type: 'speed', id: 4, val: 5 }, 
 
    { type: 'gps', id: 4, val: 6 }, 
 
    { type: 'gps', id: 5, val: 7 } 
 
] 
 

 
const generator$ = Rx.Observable.from(data) 
 

 
generator$ 
 
    .scan((acc, x) => { 
 
    if (R.contains(x.id, R.pluck('id', acc))) { 
 
     acc.push(x); 
 
    } else { 
 
     acc = [x] 
 
    } 
 
    return acc 
 
    }, []) 
 
    .filter(x => x.length > 1) 
 
    .subscribe(console.log)

<script src="https://cdnjs.cloudflare.com/ajax/libs/ramda/0.23.0/ramda.min.js"></script> 
 
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.1/Rx.min.js"></script>