feat: Support for async subscriber callbacks

source/event_hub.ts

@@ -557,18 +557,25 @@ export class EventHub {
    */
   private _handle(eventPayload: EventPayload) {
     this.logger.debug("Event received", eventPayload);
-
+    const promises: Promise<any>[] = [];
     for (const subscriber of this._subscribers) {
       // TODO: Parse event target and check that it matches subscriber.
 
       // TODO: Support full expression format as used in Python.
       if (!this._IsSubscriberInterestedIn(subscriber, eventPayload)) {
         continue;
       }
-
-      let response = null;
       try {
-        response = subscriber.callback(eventPayload);
+        const responsePromise = Promise.resolve(
+          subscriber.callback(eventPayload)
+        );
+        promises.push(responsePromise);
+        responsePromise.then((response) => {
+          // Publish reply if response isn't null or undefined.
+          if (response != null) {
+            this.publishReply(eventPayload, response, subscriber.metadata);
+          }
+        });
       } catch (error) {
         this.logger.error(
           "Error calling subscriber for event.",
@@ -577,12 +584,8 @@ export class EventHub {
           eventPayload
         );
       }
-
-      // Publish reply if response isn't null or undefined.
-      if (response != null) {
-        this.publishReply(eventPayload, response, subscriber.metadata);
-      }
     }
+    return promises;
   }
 
   /**

source/session.ts

@@ -611,7 +611,7 @@ export class Session {
         if (reason instanceof Error) {
           throw this.getErrorFromResponse({
             exception: "NetworkError",
-            content: (reason.cause as string) || reason.message,
+            content: (reason["cause"] as string) || reason.message,
           });
         }
         throw new Error("Unknown error");

test/event_hub.test.js

@@ -152,4 +152,60 @@ describe("EventHub", () => {
 
     expect(callback).not.toHaveBeenCalledWith(testEvent);
   });
+
+  test("should handle sync callback and return correct data", async () => {
+    const callback = vi.fn(() => "someData");
+    const testEvent = {
+      topic: "ftrack.test",
+      data: {},
+      id: "eventId",
+      source: { id: "sourceId" },
+    };
+
+    const publishReplySpy = vi
+      .spyOn(eventHub, "publishReply")
+      .mockImplementation((_, data) => data);
+
+    eventHub.subscribe("topic=ftrack.test", callback);
+    const promises = eventHub._handle(testEvent);
+    await Promise.all(promises);
+    expect(callback).toHaveBeenCalledWith(testEvent);
+    expect(publishReplySpy).toHaveBeenCalledWith(
+      expect.anything(),
+      "someData",
+      expect.anything()
+    );
+    publishReplySpy.mockRestore();
+  });
+
+  test("should not handle async callback with a promise", async () => {
+    const asyncCallback = vi.fn(async () => "someData");
+    const testEvent = {
+      topic: "ftrack.test",
+      data: {},
+      id: "eventId",
+      source: { id: "sourceId" },
+    };
+
+    const publishReplySpy = vi
+      .spyOn(eventHub, "publishReply")
+      .mockImplementation((_, data) => data);
+
+    eventHub.subscribe("topic=ftrack.test", asyncCallback);
+    const promises = eventHub._handle(testEvent);
+    await Promise.all(promises ?? []);
+    expect(asyncCallback).toHaveBeenCalledWith(testEvent);
+    expect(publishReplySpy).toHaveBeenCalled();
+    expect(publishReplySpy).not.toHaveBeenCalledWith(
+      expect.anything(),
+      expect.any(Promise),
+      expect.anything()
+    );
+    expect(publishReplySpy).toHaveBeenCalledWith(
+      expect.anything(),
+      "someData",
+      expect.anything()
+    );
+    publishReplySpy.mockRestore();
+  });
 });