javascript - 将 Array 序列化为元素，使用 RxJS 进行转换，并将元素组装回数组

Question

在 Angular2 中，我有很多 Observable<any[]> (发出数组的 Observable)在 http.get() 中产生的后代或通过 websocket 操作提供，因此不 .complete() 但随着时间的推移发出多个值。

我经常需要使用 RxJS 运算符转换数组中的元素(我不想使用 Array.prototype.* 转换!)并将单个元素组装回数组，作为单个实体发出。

但我不知道如何将元素组装回数组。

例子:

const n$ = new Subject();

const output = n$
    // create an observable emitting the individual elements
    // of the array
    .mergeMap(n => n)

    // some kind of transform on the elements
    .distinct((n1, n2) => n1 == n2)
    .map(n => n*n)

    // how to assemble back to an array here???
    // not working:
    // .buffer(n$)
    // also not working (subject does not complete!)
    // .toArray()

output.subscribe(v => console.log(v))

n$.next([1,1,1,2,3]);
n$.next([4,5,5,6]);

// Wanted output:
// [1, 4, 9]
// [16, 25, 36]

Answer 1

如果您有多个值并想要一个值(数组)， reduce toArray应该是你所追求的:

Rx.Observable.from([0, 1, 1, 2, 3])
    .distinct()
    .map((n) => n * n)
    // .reduce((acc, n) => { acc.push(n); return acc; }, [])
    .toArray()
    .subscribe((a) => { console.log(a); })

如果你有一个 Observable<any[]> , 只需将其放入 mergeMap :

const output = n$
    .mergeMap((a) => Rx.Observable.from(a)
        .distinct()
        .map((n) => n * n)
        // .reduce((acc, n) => { acc.push(n); return acc; }, [])
        .toArray()
    )
    .subscribe(a => { console.log(a); });