Add subscription mechanism using async iterators

index.js

@@ -2,6 +2,8 @@
 
 const anyMap = new WeakMap();
 const eventsMap = new WeakMap();
+const producersMap = new WeakMap();
+const anyProducer = Symbol('anyProducer');
 const resolvedPromise = Promise.resolve();
 
 function assertEventName(eventName) {
@@ -25,6 +27,82 @@ function getListeners(instance, eventName) {
 	return events.get(eventName);
 }
 
+function getEventProducers(instance, eventName) {
+	const key = typeof eventName === 'string' ? eventName : anyProducer;
+	const producers = producersMap.get(instance);
+	if (!producers.has(key)) {
+		producers.set(key, new Set());
+	}
+
+	return producers.get(key);
+}
+
+function enqueueProducers(instance, eventName, eventData) {
+	const producers = producersMap.get(instance);
+	if (producers.has(eventName)) {
+		for (const producer of producers.get(eventName)) {
+			producer.enqueue(eventData);
+		}
+	}
+
+	if (producers.has(anyProducer)) {
+		const item = Promise.all([eventName, eventData]);
+		for (const producer of producers.get(anyProducer)) {
+			producer.enqueue(item);
+		}
+	}
+}
+
+function iterator(instance, eventName) {
+	let finished = false;
+	let flush = () => {};
+	let queue = [];
+	const producer = {
+		enqueue(item) {
+			queue.push(item);
+			flush();
+		},
+		finish() {
+			finished = true;
+			flush();
+		}
+	};
+
+	getEventProducers(instance, eventName).add(producer);
+	return {
+		async next() {
+			if (!queue) {
+				return {done: true};
+			}
+
+			if (queue.length === 0) {
+				if (finished) {
+					queue = null;
+					return this.next();
+				}
+
+				await new Promise(resolve => {
+					flush = resolve;
+				});
+				return this.next();
+			}
+
+			return {done: false, value: await queue.shift()};
+		},
+		async return(value) {
+			queue = null;
+			getEventProducers(instance, eventName).delete(producer);
+			flush();
+			return arguments.length > 0 ?
+				{done: true, value: await value} :
+				{done: true};
+		},
+		[Symbol.asyncIterator]() {
+			return this;
+		}
+	};
+}
+
 function defaultMethodNamesOrAssert(methodNames) {
 	if (methodNames === undefined) {
 		return allEmitteryMethods;
@@ -92,6 +170,7 @@ class Emittery {
 	constructor() {
 		anyMap.set(this, new Set());
 		eventsMap.set(this, new Map());
+		producersMap.set(this, new Map());
 	}
 
 	on(eventName, listener) {
@@ -117,9 +196,16 @@ class Emittery {
 		});
 	}
 
+	events(eventName) {
+		assertEventName(eventName);
+		return iterator(this, eventName);
+	}
+
 	async emit(eventName, eventData) {
 		assertEventName(eventName);
 
+		enqueueProducers(this, eventName, eventData);
+
 		const listeners = getListeners(this, eventName);
 		const anyListeners = anyMap.get(this);
 		const staticListeners = [...listeners];
@@ -170,6 +256,10 @@ class Emittery {
 		return this.offAny.bind(this, listener);
 	}
 
+	anyEvent() {
+		return iterator(this);
+	}
+
 	offAny(listener) {
 		assertListener(listener);
 		anyMap.get(this).delete(listener);
@@ -178,17 +268,34 @@ class Emittery {
 	clearListeners(eventName) {
 		if (typeof eventName === 'string') {
 			getListeners(this, eventName).clear();
+			const producers = getEventProducers(this, eventName);
+
+			for (const producer of producers) {
+				producer.finish();
+			}
+
+			producers.clear();
 		} else {
 			anyMap.get(this).clear();
+
 			for (const listeners of eventsMap.get(this).values()) {
 				listeners.clear();
 			}
+
+			for (const producers of producersMap.get(this).values()) {
+				for (const producer of producers) {
+					producer.finish();
+				}
+
+				producers.clear();
+			}
 		}
 	}
 
 	listenerCount(eventName) {
 		if (typeof eventName === 'string') {
-			return anyMap.get(this).size + getListeners(this, eventName).size;
+			return anyMap.get(this).size + getListeners(this, eventName).size +
+				getEventProducers(this, eventName).size + getEventProducers(this).size;
 		}
 
 		if (typeof eventName !== 'undefined') {
@@ -201,6 +308,10 @@ class Emittery {
 			count += value.size;
 		}
 
+		for (const value of producersMap.get(this).values()) {
+			count += value.size;
+		}
+
 		return count;
 	}
 

readme.md

@@ -68,8 +68,13 @@ emitter.once('🦄').then(data => {
 
 emitter.emit('🦄', '🌈');
 ```
+#### events(eventName)
 
-#### emit(eventName, data?)
+Get an asynchronous iterator which buffers data each time an event is emitted.
+
+Call `return()` on the iterator to remove the subscription.
+
+#### emit(eventName, [data])
 
 Trigger an event asynchronously, optionally with some data. Listeners are called in the order they were added, but execute concurrently.
 
@@ -93,6 +98,12 @@ Returns a method to unsubscribe.
 
 Remove an `onAny` subscription.
 
+#### anyEvent()
+
+Get an asynchronous iterator which buffers a tuple of an event name and data each time an event is emitted.
+
+Call `return()` on the iterator to remove the subscription.
+
 #### clearListeners()
 
 Clear all event listeners on the instance.