wip

src/Utils/event-buffer.ts

@@ -39,16 +39,15 @@ type BaileysBufferableEventEmitter = BaileysEventEmitter & {
 	process(handler: (events: BaileysEventData) => void | Promise<void>): (() => void)
 	/**
 	 * starts buffering events, call flush() to release them
+	 * @returns true if buffering just started, false if it was already buffering
 	 * */
-	buffer(): void
+	buffer(): boolean
 	/** buffers all events till the promise completes */
 	createBufferedFunction<A extends any[], T>(work: (...args: A) => Promise<T>): ((...args: A) => Promise<T>)
-	/**
-	 * flushes all buffered events
-	 * @param force if true, will flush all data regardless of any pending buffers
-	 * @returns returns true if the flush actually happened, otherwise false
-	 */
-	flush(force?: boolean): boolean
+	/** flushes all buffered events */
+	flush(): Promise<void>
+	/** waits for the task to complete, before releasing the buffer */
+	processInBuffer(task: Promise<any>)
 	/** is there an ongoing buffer */
 	isBuffering(): boolean
 }
@@ -63,7 +62,9 @@ export const makeEventBuffer = (logger: Logger): BaileysBufferableEventEmitter =
 	const historyCache = new Set<string>()
 
 	let data = makeBufferData()
-	let buffersInProgress = 0
+	let isBuffering = false
+	let preBufferTask: Promise<any> = Promise.resolve()
+	let preBufferTraces: string[] = []
 
 	// take the generic event and fire it as a baileys event
 	ev.on('event', (map: BaileysEventData) => {
@@ -73,24 +74,25 @@ export const makeEventBuffer = (logger: Logger): BaileysBufferableEventEmitter =
 	})
 
 	function buffer() {
-		buffersInProgress += 1
+		if(!isBuffering) {
+			logger.trace('buffering events')
+			isBuffering = true
+			return true
+		}
+
+		return false
 	}
 
-	function flush(force = false) {
-		// no buffer going on
-		if(!buffersInProgress) {
-			return false
+	async function flush() {
+		if(!isBuffering) {
+			return
 		}
 
-		if(!force) {
-			// reduce the number of buffers in progress
-			buffersInProgress -= 1
-			// if there are still some buffers going on
-			// then we don't flush now
-			if(buffersInProgress) {
-				return false
-			}
-		}
+		logger.trace({ preBufferTraces }, 'releasing buffered events...')
+		await preBufferTask
+
+		preBufferTraces = []
+		isBuffering = false
 
 		const newData = makeBufferData()
 		const chatUpdates = Object.values(data.chatUpdates)
@@ -115,8 +117,6 @@ export const makeEventBuffer = (logger: Logger): BaileysBufferableEventEmitter =
 			{ conditionalChatUpdatesLeft },
 			'released buffered events'
 		)
-
-		return true
 	}
 
 	return {
@@ -131,26 +131,34 @@ export const makeEventBuffer = (logger: Logger): BaileysBufferableEventEmitter =
 			}
 		},
 		emit<T extends BaileysEvent>(event: BaileysEvent, evData: BaileysEventMap[T]) {
-			if(buffersInProgress && BUFFERABLE_EVENT_SET.has(event)) {
+			if(isBuffering && BUFFERABLE_EVENT_SET.has(event)) {
 				append(data, historyCache, event as any, evData, logger)
 				return true
 			}
 
 			return ev.emit('event', { [event]: evData })
 		},
+		processInBuffer(task) {
+			if(isBuffering) {
+				preBufferTask = Promise.allSettled([ preBufferTask, task ])
+				preBufferTraces.push(new Error('').stack!)
+			}
+		},
 		isBuffering() {
-			return buffersInProgress > 0
+			return isBuffering
 		},
 		buffer,
 		flush,
 		createBufferedFunction(work) {
 			return async(...args) => {
-				buffer()
+				const started = buffer()
 				try {
 					const result = await work(...args)
 					return result
 				} finally {
-					flush()
+					if(started) {
+						await flush()
+					}
 				}
 			}
 		},