events: add once method to use promises with EventEmitter

[mcollina]

This change adds a EventEmitter.once() method that wraps ee.once in a
promise. I've been using this model for some months now, and it works extremely well for me.

const { once, EventEmitter } = require('events');
async function run() {
  const ee = new EventEmitter();
  process.nextTick(() => {
    ee.emit('myevent', 42);
  });
  const [value] = await once(ee, 'myevent');
  console.log(value);
  const err = new Error('kaboom');
  process.nextTick(() => {
    ee.emit('error', err);
  });
  try {
    await once(ee, 'myevent');
  } catch (err) {
    console.log('error happened', err);
  }
}
run();

Checklist

• make -j4 test (UNIX), or vcbuild test (Windows) passes
• tests and/or benchmarks are included
• documentation is changed or added
• commit message follows commit guidelines

[nodejs-github-bot]

@mcollina build started: https://ci.nodejs.org/blue/organizations/jenkins/node-test-pull-request-lite-pipeline/detail/node-test-pull-request-lite-pipeline/2592/pipeline

[mcollina]

Also @davidmarkclements that had the original idea: would you like to be included as "co-authored-by"?

[lpinca]

Why in core? Can't this be an npm module?

[addaleax]

Thank you! 💙

[Fishrock123]

Same question as @lpinca -

Why in core? Can't this be an npm module?

I mean, I understand why it would be in core, I think -- that is, to make yet more node.js core apis usable with await.

But I also think that since it doesn't appear to need to be in core, it should at very least have an npm module with adoption to show the weight of it's usefulness...

[cjihrig]

Fixes #20893 ?

[jasnell]

While this could live outside of core, it's been frequently requested and aligns with our promises support. I'm +1 on adding it

[vsemozhetbyt]

Should not this be events.once(emitter, name)? Otherwise, this seems to be a static EventEmitter method like EventEmitter.listenerCount() above. If I understand correctly, this would be called as events.once() in the REPL. I see that it can be used as EventEmitter static method, but in examples, it is imported alongside with the class and this can be confusing for users not aware of EventEmitter.EventEmitter = EventEmitter; core binding.

[lpinca]

If I'm not wrong it even works with a third party EventEmitter assuming that it supports once() and removeListener() which kind of reinforces "why in core?". In my opinion this goes against the small core principle.

• It is not more efficient than a userland implementation.
• It doesn't add something that can't be done in userland.
• As far as I know It is not needed in core itself. Is the plan to make core API use it?

[mcollina]

I think this API (or something similar) belongs in core for three main reasons:

1. it aligns with our strategic initiative of “promisifying core APIs”.
2. it is extremely hard to interact with stream using async/await without an utility like this.
3. making sure that 12.0.0 has this will shorten the adoption cycle by 1 year.

I think we can use this API in a lot of test code within core.

I’m using it from a userland module, but tiny utility modules are hard to market at this point.

I’m happy to provide more examples if needed.

[mcollina]

@lpinca we have been using it for 1 year and a half in Pino test suite (and copy-paste in a lot of other places): https://github.com/pinojs/pino/blob/master/test/basic.test.js#L43 as an example.

[lpinca]

Yes, ok but that does not answer "why in core?". A npm module would be better for this in my opinion. Anyway I'm not blocking this and it already has a few TSC approvals.

[mcollina]

cc @nodejs/tsc adding to tsc-agenda for visibility

[apapirovski]

LGTM. I don't disagree with the comment about this possibly belonging in user-land but I'm also happy with it in Node. Implementation is super clean 👍

[apapirovski]

This is just personal preference but having this at the start of the function body would be clearer for me. When I read the code, the secondListener is referenced before its defined which makes it somewhat confusing. I know it's kind of nitpicky tho so non-blocking...

(Maybe naming it errorRejectListener would also be clearer?)

[sam-github]

eventListener and errorListener (or something of the like) would be clearer than main and second.

[ChALkeR]

Related: #20909

[Trott]

No strong opinion, but if there's concerns especially around API design, maybe land as Experimental so we can change the API if we don't get it right the first time?

[AAverin]

Why not do the same for on?
Then it would be possible to have a bus based on promises

[sam-github]

.once() callbacks once, promises resolve once. .on() callbacks repeatedly, so how could that work with promises, which cannot resolve multiple times?

[mcollina]

Why not do the same for on?

We could expose a on(ee, 'event') method that returns an async iterator. However there would be no "exit" condition from the loop.

for await (let event of on(ee, 'event')) {
  // do something with event
  // this loop will never end
}

Would you mind opening a new issue to discuss? Do you think this would be worthwhile even if the loop would never end?

[AAverin]

Well, problem I have so far, if I have an active subscription to events that can come forever until I unsubscribe, there is no way to have a nice handling of that with promises.
Basically I am looking for a "hot observable" kind of thing.

[mcollina]

I would recommend using a stream instead. It’s async iterable already.

[jasnell]

We could support an on function that returns an async iterator so long as we introduced a concept of a loop terminator.... That is, an object that would cause the loop to terminate...

E.g. something roughly like...

const foo = events.on(emitter, 'foo', term);
for await (const f of foo) {
  // ...
  term.exit()
}

[benjamingr]

@jasnell @mcollina actually break and return would work:

const foo = events.on(emitter, 'foo');
for await (const f of foo) {
  break; // this calls `.return` on the async iterable and can unsubscribe.
}

This is also true for regular generators. I'm +1 on exploring this.

[jasnell]

@benjamingr ... yes, those work within the actual for await but we would need to make sure we're able to clean up inside the async iterator itself (e.g. removing the event listener). It may be rather trivial to do, I just haven't tried it yet :-) ... Perhaps I'll take a break from QUIC implementation and give it a try :-)

[jasnell]

Ok, quick investigation... it should be possible to implement an EventEmitter.on(emitter, event) function that is essentially a simplification of the code used for stream.Readable with a few notable differences:

1. Unlike stream.Readable, there is no natural end to the event stream, and there is no destroy() method, so cleanup and termination of the for await() {} would be entirely dependent on using break and return as suggested.

2. With stream.Readable, when there are multiple synchronous push()'s, there's still just a single 'readable', and just in case there are multiple read() operations performed. For an arbitrary EventEmitter, we would need a way of reliably handling synchronous emits.

Everything else looks like it should be fairly reasonable to implement if someone wants to take it on ;-)

[benjamingr]

moving to a separate issue