Mind Games: Asynchronous Handling in EventEmitter

Anyone who has used node.js knows that it implements the Observer design pattern with a set of event listeners much like Python’s, called EventEmitter. You create (or extend) an EventEmitter, emit the event you want when you need to trigger it, and elsewhere in your code you listen for that event and handle it.

// framework code
const ev = EventEmitter();
ev.emit('event', data);

// user code, add event listener
ev.on('event', data => process(data));

EventEmitter is especially useful when building a library or framework, because it decouples events from their handling: the emitter need not care who handles the event or how. But unlike Python, a language that executes sequentially, node.js is fully asynchronous — which means event handling may not proceed the way you imagine. Recently, while building an API framework in nodejs, I ran into exactly this wall.

Here’s the problem: I wanted my framework flexible enough that, before and after handling a given route, users could inject code of their own (middleware can’t fully satisfy this need), something like:

class Application() {
  ...
  _setupRoutes() {
    const preprocessor = (req, res, next) => {
      this.server.emit('preprocessing', req, res, next);
    };

    const postprocessor = (req, res, next) => {
      this.server.emit('postprocessing', req, res, next);
    };

    this.routes.forEach(r => {
      const validator = getValidator(this.server, r);
      const routePath = {
        path: `/${prefix}/${r.path}`,
        name: r.name,
        needAuth: r.needAuth
      };
      if (r.version) routePath.version = r.version;
      ...

      this.server[r.method](
        routePath, validator, preprocessor,
        r.action, postprocessor);
    });
  }
  ...
}

That way, a framework user could write:

server.on('preprocessing', (req, res, next) => {
    // preprocessing hook!
});

The ideal was plump; reality was bony. This implementation has several problems:

1. If the preprocessing listener is an async function, preprocessor() finishes later than r.action() — which is not what we want!
2. If there are multiple preprocessing listeners (as a framework, you must guarantee this flexibility), who calls next() to hand control over to r.action()?
3. If preprocessor() throws, how do we abort the execution of r.action()?
4. With multiple preprocessing listeners, how do we guarantee their execution order?

These are thorny to deal with, and a simple solution is hard to find. Callbacks are clearly the wrong path — they make listener code hideous. We know the elegant solution to async in JavaScript is Promise, so let’s give Promise a try:

const preprocessor = (req, res, next) => {
  const promise = this.server.emit('preprocessing', req, res);
  promise.then(() => next()).catch(err => {
    req.log.warn(err, 'preprocessing hook raised error');
    next(err);
  });
};

If emit could return a Promise, we’d solve the first three problems. We take back control over when next() runs. We don’t pass next to the listeners when we emit(); instead, once the listeners have finished, we resolve/reject the returned promise and only then proceed. That way next() or next(err) runs only after preprocessor() completes, handing control to r.action() or to the error-handling flow.

However, a normal EventEmitter returns only a boolean from emit(); it does not return a Promise. That’s no obstacle — a monkey patch will teach EventEmitter this trick.

Monkey patching in JavaScript is simple, and you’ve probably done it yourself:

> Array.prototype.sum = function() {
  return this.reduce((acc, item) => acc + item, 0);
}
[Function]
> a = [1, 2, 3]
[ 1, 2, 3 ]
> a.sum()
6

That is a monkey patch: by injecting a new function into Array.prototype, we changed Array’s behavior. Likewise, we only need to inject a new function into EventEmitter that returns a Promise, and our needs are met. Be careful, though: when monkey patching, don’t alter existing functions — that causes all sorts of trouble. So we should create a new one:

EventEmitter.prototype.emitAsync = function(type) {
  ...
}

In this function we can execute all the event handlers. Since multiple handlers may be sync or async, we need to unify them. Sync execution is a special case of async execution, so we can convert the result of a sync execution into a Promise, like this:

let result;
try {
  result = handler(args);
} catch (err) {
  result = new Promise((res, rej) => rej(err));
}
if (result instanceof Promise) {
  /* we're good */
}
else {
  result = new Promise(res => res(result));
}

This way, whatever a handler produces — even a thrown exception — is wrapped asynchronously into a Promise. Of course, as a framework we cannot know whether a listener is sync or async, so we require that a listener return a Promise if it handles things asynchronously.

If we have multiple event listeners, we can use Promise.All() to aggregate all the async events together.

By and large we’ve now built what we wanted; only the fourth problem remains: with multiple preprocessing listeners, how do we guarantee execution order? Promise.All() runs all listeners in parallel, which is useful, but often we’d rather have listeners run one at a time in order, whether async or sync.

If you know Observable, this problem is easy to solve with it. You just create an Observable for each handler’s execution, then use concatAll() to run them in sequence.

const mode = sequential ? 'concatAll' : 'mergeAll';
Rx.Observable.from(handlers)
.map(handler =>
  new Rx.Observable.create(observer => {
    let result;
    try {
      result = handler.apply(self, args);
    } catch (e) {
      result = new Promise((_res, _rej) => _rej(e));
    }

    if (result instanceof Promise) {
      result.then(data => {
        observer.onNext(data);
        observer.onCompleted();
      }).catch(err => observer.onError(err));
    } else {
      observer.onNext(true);
      observer.onCompleted();
    }
  })
)[mode]()

When we need sequential execution, we use concatAll(); when we need parallel execution, we can use flatMap() (map() + mergeAll() is equivalent to flatMap()).

With that, all our problems are solved. We can monkey patch an emitAsync() for parallel handling of async (and sync) listeners, and likewise monkey patch an emitAsyncSeq() for sequential execution of async (and sync) listeners.

Interested readers can head over to https://github.com/tyrchen/node-eventasync to see the complete implementation. If you want to use it in your own project, just run:

$ npm install eventasync