AsyncEmitter

ae.js

'use strict';

const AsyncEmitter = () => ({
  _events: {},
  on(name, f) {
    !Object.keys(this._events).includes(name) &&
      (this._events[name] = []);
    if (this._events[name].find(fn => fn.source === f || fn === f)) return;
    this._events[name].push(f);
  },
  once(name, f) {
    if (f === undefined) return new Promise(r => this.once(name, r));
    !Object.keys(this._events).includes(name) &&
      (this._events[name] = []);
    if (this._events[name].find(fn => fn.source === f || fn === f)) return;
    const hocfn = async (...args) => {
      this.remove(name, f);
      f(...args);
    }
    hocfn.source = f;
    this._events[name].push(hocfn);
  },
  async emit(name, ...args) {
    if (!this._events[name]) return;
    return Promise.all(this._events[name].map(f => f(...args)));
  },
  remove(name, f) {
    if (!this._events[name] || !this._events[name].length) return;
    const remIdx = this._events[name]
      .findIndex(fn => fn.source === f || fn === f)
    remIdx >= 0 && this._events[name].splice(remIdx, 1);
    !this._events[name].length && delete this._events[name];
  },
  clear(name) {
    if (!name) {
      this._events = {};
      return;
    }
    delete this._events[name];
  },
  count(name) { return this._events[name] ? this._events[name].length : 0 },
  listeners(name) { return this._events[name] },
  names() { return Object.keys(this._events) }
});

module.exports = AsyncEmitter;

ae.test.js

const Ae = require('./ae');

const ee = Ae();

(async () => {
  const listeners = Array(4096 * 2)
    .fill(null)
    .map((_, i) => async () => console.log(`Listener #${i}`));
  listeners.forEach((fn, idx) => {
    ee.on(idx.toString(), fn);
  });

  console.time('exec')
  Promise.all(listeners.map(async (_, idx) => {
    await ee.emit(idx.toString())
  }))
  console.timeEnd('exec');
})();

metatests.js

'use strict';

const AsyncEmitter = require('./ae');
const metatests = require('metatests');

metatests.test('AsyncEmitter on/emit', async test => {
  const ae = AsyncEmitter();

  const fn = test.mustCall(async (a, b, c, d) => {
    test.strictSame(a, 1);
    test.strictSame(b, 2);
    test.strictSame(c, 3);
    test.strictSame(d, 4);
  });

  ae.on('e1', fn);

  test.strictSame(Object.keys(ae._events).length, 1);

  await ae.emit('e1', 1, 2, 3, 4);

  test.strictSame(Object.keys(ae._events).length, 1);

  test.end();
});

metatests.test('AsyncEmitter once', async test => {
  const ae = AsyncEmitter();

  ae.once('e1', test.mustCall());
  ae.once('e1', test.mustCall());

  test.strictSame(Object.keys(ae._events).length, 1);

  ae.emit('e1');
  ae.emit('e1');

  test.strictSame(Object.keys(ae._events).length, 0);

  test.end();
});

metatests.test('AsyncEmitter await once', async test => {
  const ae = AsyncEmitter();

  const fn = test.mustCall(() => {
    test.strictSame(Object.keys(ae._events).length, 1);
    ae.emit('e1');
  });

  setTimeout(fn, 0);
  await ae.once('e1');

  test.strictSame(Object.keys(ae._events).length, 0);

  test.end();
});

metatests.test('AsyncEmitter remove', async test => {
  const ae = AsyncEmitter();

  const fn = test.mustCall();

  ae.on('e1', fn);
  ae.emit('e1');

  ae.remove('e1', () => {});
  ae.remove('e1', fn);
  ae.emit('e1');

  test.strictSame(Object.keys(ae._events).length, 0);

  ae.remove('e1', fn);
  ae.emit('e1');

  test.end();
});

metatests.test('AsyncEmitter remove once', async test => {
  const ae = AsyncEmitter();

  const fn = test.mustNotCall();

  ae.on('e1', fn);
  ae.once('e1', fn);

  test.strictSame(Object.keys(ae._events).length, 1);

  ae.remove('e1', fn);

  test.strictSame(Object.keys(ae._events).length, 0);

  ae.emit('e1');

  test.strictSame(Object.keys(ae._events).length, 0);

  test.end();
});

metatests.test('AsyncEmitter count', async test => {
  const ae = AsyncEmitter();

  ae.on('e1', async () => {});
  ae.on('e1', async () => {});

  test.strictSame(ae.count('e1'), 2);
  test.strictSame(ae.count('e2'), 0);

  test.end();
});

metatests.test('AsyncEmitter clear all', async test => {
  const ae = AsyncEmitter();

  ae.on('e1', test.mustNotCall());
  ae.clear();
  ae.emit('e1');

  test.strictSame(Object.keys(ae._events).length, 0);

  test.end();
});

metatests.test('AsyncEmitter clear by name', async test => {
  const ae = AsyncEmitter();

  ae.on('e1', test.mustNotCall());
  ae.clear('e1');
  ae.clear('e2');
  ae.emit('e1');

  test.strictSame(Object.keys(ae._events).length, 0);

  test.end();
});

metatests.test('AsyncEmitter clear once', async test => {
  const ae = AsyncEmitter();

  ae.once('e1', test.mustNotCall());
  ae.once('e2', test.mustNotCall());

  test.strictSame(Object.keys(ae._events).length, 2);

  ae.clear('e1');

  test.strictSame(Object.keys(ae._events).length, 1);

  ae.emit('e1');

  test.strictSame(Object.keys(ae._events).length, 1);

  test.end();
});

metatests.test('AsyncEmitter names', async test => {
  const ae = AsyncEmitter();

  ae.on('e1', () => {});
  ae.on('e1', () => {});
  ae.on('e2', () => {});
  ae.on('e3', () => {});

  test.strictSame(ae.names(), ['e1', 'e2', 'e3']);

  test.strictSame(Object.keys(ae._events).length, 3);

  test.end();
});

metatests.test('AsyncEmitter listeners', async test => {
  const ae = AsyncEmitter();

  ae.on('e1', () => {});
  ae.on('e1', () => {});
  ae.on('e2', () => {});
  ae.on('e3', () => {});

  test.strictSame(ae.listeners('e1').length, 2);
  test.strictSame(ae.listeners('e2').length, 1);

  test.end();
});

metatests.test('AsyncEmitter await', async test => {
  const ae = AsyncEmitter();

  ae.on('e1', test.mustCall());
  await ae.emit('e1');

  test.end();
});

metatests.test('AsyncEmitter await multiple listeners', async test => {
  const ae = AsyncEmitter();

  ae.on('e1', test.mustCall());
  ae.on('e1', test.mustCall());
  ae.on('e1', test.mustCall());

  await ae.emit('e1');

  test.strictSame(Object.keys(ae._events).length, 1);

  test.end();
});

metatests.test('AsyncEmitter await multiple events', async test => {
  const ae = AsyncEmitter();

  ae.on('e1', test.mustCall());
  ae.on('e2', test.mustCall());

  await ae.emit('e1');
  await ae.emit('e2');

  test.end();
});