Make emit() wait for all listeners to settle

index.js

@@ -25,6 +25,10 @@ function getListeners(instance, eventName) {
 	return events.get(eventName);
 }
 
+function hostReportError(err) {
+	Promise.reject(err);
+}
+
 class Emittery {
 	constructor() {
 		anyMap.set(this, new Set());
@@ -63,16 +67,17 @@ class Emittery {
 		const staticAnyListeners = [...anyListeners];
 
 		await resolvedPromise;
-		return Promise.all([
-			...staticListeners.map(async listener => {
-				if (listeners.has(listener)) {
-					return listener(eventData);
-				}
+
+		await Promise.all([
+			...staticListeners.map(listener => {
+				return listeners.has(listener) && new Promise(resolve => {
+					resolve(listener(eventData));
+				}).catch(hostReportError);
 			}),
-			...staticAnyListeners.map(async listener => {
-				if (anyListeners.has(listener)) {
-					return listener(eventName, eventData);
-				}
+			...staticAnyListeners.map(listener => {
+				return anyListeners.has(listener) && new Promise(resolve => {
+					resolve(listener(eventName, eventData));
+				}).catch(hostReportError);
 			})
 		]);
 	}

test/_run.js

@@ -2,6 +2,58 @@ import test from 'ava';
 import delay from 'delay';
 
 module.exports = Emittery => {
+	let rejectionCollector;
+	let onUnhandledRejection;
+
+	test.before(() => {
+		onUnhandledRejection = function (err, promise) {
+			if (rejectionCollector) {
+				rejectionCollector.next(err);
+				promise.catch(() => {});
+			}
+		};
+
+		process.on('unhandledRejection', onUnhandledRejection);
+	});
+
+	test.after(() => {
+		process.removeListener('unhandledRejection', onUnhandledRejection);
+		onUnhandledRejection = undefined;
+	});
+
+	test.afterEach.always(() => {
+		if (rejectionCollector) {
+			rejectionCollector.throw();
+		}
+	});
+
+	async function catchUnhandledRejection(limit = 1) {
+		if (rejectionCollector) {
+			throw new Error('Already collecting unhandled rejections');
+		}
+
+		return new Promise((resolve, reject) => {
+			rejectionCollector = wait();
+			rejectionCollector.next();
+
+			function * wait() {
+				try {
+					const reasons = [];
+
+					while (reasons.length < limit) {
+						reasons.push(yield);
+					}
+
+					resolve(reasons);
+				} catch (err) {
+					reject(err);
+				} finally {
+					rejectionCollector = undefined;
+				}
+			}
+		});
+	}
+
 	test('on()', async t => {
 		const emitter = new Emittery();
 		const calls = [];
@@ -127,21 +179,52 @@ module.exports = Emittery => {
 		await t.throws(emitter.emit(42), TypeError);
 	});
 
-	test.cb('emit() - is async', t => {
-		t.plan(2);
-
+	test('emit() - is async', async t => {
 		const emitter = new Emittery();
 		let unicorn = false;
-
 		emitter.on('🦄', () => {
 			unicorn = true;
-			t.pass();
-			t.end();
 		});
+		const emitted = emitter.emit('🦄');
+
+		t.false(unicorn);
+		t.is(await emitted, undefined);
+		t.true(unicorn);
+	});
+
+	test.serial('emit() - settles once all handlers settle', async t => {
+		const rejectionsPromise = catchUnhandledRejection(1);
+		const emitter = new Emittery();
+		let settled = false;
+		emitter.on('🦄', () => Promise.reject(new Error()));
+		emitter.on('🦄', () => delay(10).then(() => {
+			settled = true;
+		}));
+
+		await emitter.emit('🦄');
+		t.true(settled);
+
+		await rejectionsPromise;
+	});
+
+	test.serial('emit() - let host handle listeners rejections', async t => {
+		const rejectionsPromise = catchUnhandledRejection(3);
+		const emitter = new Emittery();
+		const first = new Error('first rejection');
+		const second = new Error('second rejection');
+		const third = new Error('third rejection');
+
+		emitter.on('🦄', () => Promise.reject(first));
+		emitter.on('🦄', () => Promise.reject(second));
+		emitter.onAny(() => Promise.reject(third));
 
 		emitter.emit('🦄');
 
-		t.false(unicorn);
+		const errors = new Set(await rejectionsPromise);
+
+		t.true(errors.has(first));
+		t.true(errors.has(second));
+		t.true(errors.has(third));
 	});
 
 	test('emit() - calls listeners subscribed when emit() was invoked', async t => {