fromEvent() should emit error when NodeJS EventEmitter emits error event

[felixfbecker]

error events are special in NodeJS. They make the process crash if there is no handler for them. That makes them the standardised event name for any error, be it on a stream, socket, child process or custom event emitters.

As a user, if I want to convert a stream into an Observable for example:

Observable.fromEvent(net.connect(1234), 'data')

I would expect a socket error to be emitted as an Observable error, so I can handle it with operators (like reconnect etc). Instead, a socket error would crash the process.

https://nodejs.org/api/events.html#events_error_events

[midnight-wonderer]

But it is very specific to node IMHO.
You can create your own factory function by wrapping Observable.fromEvent with Observable.defer and handle the error case there (maybe by merging normal events with error events).

[felixfbecker]

Yep, it is specific to node. But so is bindNodeCallback, and fromEvent is documented as handling NodeJS EventEmitters specifically. The use case for wrapping NodeJS EventEmitters is huge.

[david-driscoll]

@MidnightWonderer that's what I figured at first, but @felixfbecker is right, we do say we support it.

It should be easy enough to add the new behavior given the code is already forking at node based emitters. I imagine we just need to add something like sourceObj.addListener('error', subscriber.error.bind(subscriber));

[felixfbecker]

yep, it would just need a

sourceObj.once('error', err => subscriber.error(err))

[jbaudanza]

What if you are creating multiple observables off the same EventEmitter?

const socket = net.connect(1234);
const data$ = Observable.fromEvent(socket, 'data');
const drain$ = Observable.fromEvent(socket, 'drain');

Do we want the same error to be sent to both observables?

[mattpodwysocki]

@felixfbecker I would be more comfortable with the following which uses a merge to add in the error logic:

const event$ = Observable.merge(
  Observable.fromEvent(socket, 'data'),
  Observable.fromEvent(socket, 'error').mergeMap(x => Observable.throw(x))
);

Otherwise, feel free to move forward some of the rx-node project from v4

[benlesh]

The proper answer here is to use the above suggestion.

In RxJS 6+, that would be something like this:

import { merge, fromEvent } from 'rxjs';
import { tap } from 'rxjs/operators';

const event$ = merge(
  fromEvent(socket, 'data'),
  // convert error events to errors.
  fromEvent(socket, 'error').pipe(tap(err => { throw err; }));
);