Added support for Async events (#643)

* Added support for async events callbacks

* Don't fire IDLE event if the first callback hasn't completed execution.

ui/media/js/engine.js

@@ -196,7 +196,7 @@
     const eventSource = new GenericEventSource(EVENTS_TYPES)
 
     function setServerStatus(msgType, msg) {
-        eventSource.fireEvent(EVENT_STATUS_CHANGED, {type: msgType, message: msg})
+        return eventSource.fireEvent(EVENT_STATUS_CHANGED, {type: msgType, message: msg})
     }
 
     const ServerStates = {
@@ -625,7 +625,7 @@
             }
             this._setStatus(TaskStatus.pending)
             task_queue.set(this, promiseGenerator)
-            eventSource.fireEvent(EVENT_TASK_QUEUED, {task:this})
+            await eventSource.fireEvent(EVENT_TASK_QUEUED, {task:this})
             await Task.enqueue(promiseGenerator, ...args)
             await this.waitUntil({status: TaskStatus.completed})
             if (this.exception) {
@@ -843,7 +843,7 @@
             if (typeof jsonResponse?.task !== 'number') {
                 console.warn('Endpoint error response: ', jsonResponse)
                 const event = Object.assign({task:this}, jsonResponse)
-                eventSource.fireEvent(EVENT_UNEXPECTED_RESPONSE, event)
+                await eventSource.fireEvent(EVENT_UNEXPECTED_RESPONSE, event)
                 if ('continueWith' in event) {
                     jsonResponse = await Promise.resolve(event.continueWith)
                 }
@@ -1087,6 +1087,7 @@
         return activeDevicesCount
     }
 
+    let idleEventPromise = undefined
     function continueTasks() {
         if (typeof navigator?.scheduling?.isInputPending === 'function') {
             const inputPendingOptions = {
@@ -1101,14 +1102,18 @@
         }
         const serverCapacity = getServerCapacity()
         if (task_queue.size <= 0 && concurrent_generators.size <= 0) {
-            eventSource.fireEvent(EVENT_IDLE, {capacity: serverCapacity, idle: true})
+            if (!idleEventPromise?.isPending) {
+                idleEventPromise = makeQuerablePromise(eventSource.fireEvent(EVENT_IDLE, {capacity: serverCapacity, idle: true}))
+            }
             // Calling idle could result in task being added to queue.
             if (task_queue.size <= 0 && concurrent_generators.size <= 0) {
-                return asyncDelay(IDLE_COOLDOWN)
+                return idleEventPromise.then(() => asyncDelay(IDLE_COOLDOWN))
             }
         }
         if (task_queue.size < serverCapacity) {
-            eventSource.fireEvent(EVENT_IDLE, {capacity: serverCapacity - task_queue.size})
+            if (!idleEventPromise?.isPending) {
+                idleEventPromise = makeQuerablePromise(eventSource.fireEvent(EVENT_IDLE, {capacity: serverCapacity - task_queue.size}))
+            }
         }
         const completedTasks = []
         for (let [generator, promise] of concurrent_generators.entries()) {
@@ -1175,8 +1180,8 @@
                 continue
             }
             const event = {task, generator};
-            eventSource.fireEvent(EVENT_TASK_START, event) // optional beforeStart promise to wait on before starting task.
-            const promise = makeQuerablePromise(Promise.resolve(event.beforeStart))
+            const beforeStart = eventSource.fireEvent(EVENT_TASK_START, event) // optional beforeStart promise to wait on before starting task.
+            const promise = makeQuerablePromise(beforeStart.then(() => Promise.resolve(event.beforeStart)))
             concurrent_generators.set(event.generator, promise)
             task_queue.set(task, event.generator)
         }
@@ -1201,7 +1206,7 @@
             }
             const continuePromise = continueTasks().catch(async function(err) {
                 console.error(err)
-                eventSource.fireEvent(EVENT_UNHANDLED_REJECTION, {reason: err})
+                await eventSource.fireEvent(EVENT_UNHANDLED_REJECTION, {reason: err})
                 await asyncDelay(RETRY_DELAY_ON_ERROR)
             })
             taskPromise = makeQuerablePromise(continuePromise)

ui/media/js/main.js

@@ -473,7 +473,7 @@ function makeImage() {
     initialText.style.display = 'none'
 }
 
-function onIdle() {
+async function onIdle() {
     const serverCapacity = SD.serverCapacity
     for (const taskEntry of getUncompletedTaskEntries()) {
         if (SD.activeTasks.size >= serverCapacity) {
@@ -485,7 +485,7 @@ function onIdle() {
             taskStatusLabel.style.display = 'none'
             continue
         }
-        onTaskStart(task)
+        await onTaskStart(task)
     }
 }
 
@@ -676,7 +676,7 @@ function onTaskCompleted(task, reqBody, instance, outputContainer, stepUpdate) {
 }
 
 
-function onTaskStart(task) {
+async function onTaskStart(task) {
     if (!task.isProcessing || task.batchesDone >= task.batchCount) {
         return
     }
@@ -714,22 +714,24 @@ function onTaskStart(task) {
     task.outputContainer.insertBefore(outputContainer, task.outputContainer.firstChild)
 
     const eventInfo = {reqBody:newTaskReqBody}
-    PLUGINS['TASK_CREATE'].forEach((hook) => {
+    const callbacksPromises = PLUGINS['TASK_CREATE'].map((hook) => {
         if (typeof hook !== 'function') {
             console.error('The provided TASK_CREATE hook is not a function. Hook: %o', hook)
-            return
+            return Promise.reject(new Error('hook is not a function.'))
         }
         try {
-            hook.call(task, eventInfo)
+            return Promise.resolve(hook.call(task, eventInfo))
         } catch (err) {
             console.error(err)
+            return Promise.reject(err)
         }
     })
+    await Promise.allSettled(callbacksPromises)
     let instance = eventInfo.instance
     if (!instance) {
         const factory = PLUGINS.OUTPUTS_FORMATS.get(eventInfo.reqBody?.output_format || newTaskReqBody.output_format)
         if (factory) {
-            instance = factory(eventInfo.reqBody || newTaskReqBody)
+            instance = await Promise.resolve(factory(eventInfo.reqBody || newTaskReqBody))
         }
         if (!instance) {
             console.error(`${factory ? "Factory " + String(factory) : 'No factory defined'} for output format ${eventInfo.reqBody?.output_format || newTaskReqBody.output_format}. Instance is ${instance || 'undefined'}. Using default renderer.`)

ui/media/js/utils.js

@@ -561,16 +561,22 @@ class GenericEventSource {
             throw new Error(`Event ${String(name)} missing from Events.types`)
         }
         if (!this.#events.hasOwnProperty(name)) {
-            return
+            return Promise.resolve()
         }
         if (!args || !args.length) {
             args = []
         }
         const evs = this.#events[name]
-        const len = evs.length
-        for (let i = 0; i < len; ++i) {
-            evs[i].apply(SD, args)
+        if (evs.length <= 0) {
+            return Promise.resolve()
         }
+        return Promise.allSettled(evs.map((callback) => {
+            try {
+                return Promise.resolve(callback.apply(SD, args))
+            } catch (ex) {
+                return Promise.reject(ex)
+            }
+        }))
     }
 }