How do I merge an array of Observables with RxJS 6.x and Node.js?

Question

For learning purposes, I'm creating a Node app that will need to take x RxJS observables from an array and combine into a single stream of events. I want to know when events occur in any observable, in any order (not in any sequence or full completion). I feel it should be in a single merged stream of events. Basically, the first event that comes through from any observable will complete.

For this, I felt merge() will do the trick. As merge doesn't take arrays directly as a parameter, I'm using reduce so far to help merge.

However, the end result is not an observable, but a function. I can't subscribe to it either. A simplified version of the code can be seen below.

How can I alter this Node 10.14.2, RxJS 6.4.x code to return an observable and not a "[function]" that I can tack a .subscribe() to?

const { Observable } = require('rxjs');
const { merge } = require('rxjs/operators');

const observables = [
    Observable.create(observer => observer.next('Hello')),
    Observable.create(observer => observer.next('Hello')),
    Observable.create(observer => observer.next('Hello'))
];

const mergedObservables = observables.reduce((merged, observable) => {
    console.log(observable);
    return merge(merged, observable);
});

// outputs:
// Observable { _isScalar: false, _subscribe: [Function] }
// Observable { _isScalar: false, _subscribe: [Function] }

console.log(mergedObservables);

// outputs:
// [Function]

mergedObservables.subscribe();
// error:
// TypeError: mergedObservables.subscribe is not a function

seniorquico seniorquico · Accepted Answer · 2019-04-22T04:54:37

EDIT: You're importing the merge operator as opposed to the static merge function. The former operates on events from a source observable while the latter creates a new observable from one or more source observables. While the thought about the spread syntax below simplifies your code, it wasn't your real issue.

It appears Node.js >= 5.0 supports the spread operator for arrays in function calls (you don't specify which version of Node.js you're using, though). The following should work if you're using a modern version of Node.js:

const { Observable, merge } = require('rxjs')

const observables = [
    Observable.create(observer => observer.next('Hello')),
    Observable.create(observer => observer.next('Hello')),
    Observable.create(observer => observer.next('Hello'))
]

const mergedObservables = merge(...observables)
mergedObservables.subscribe(event => { console.log(event) })

How do I merge an array of Observables with RxJS 6.x and Node.js?

1 Answers