如何有条件地合并单个Observable流中的对象？

Question

甲流constains以下对象

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 } 
] 

万一ID相同，则对象被合并。如果没有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

没有必要拆分流并重新合并它。您可以使用scan收集的对象和filter出不符合那些调理

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>