feat: Support for async subscriber callbacks (#132)

* Update event_hub.ts Add support for async subscriber callbacks while retaining backwards compatibility * Made changes to precommits. * Global fix for precommits. * Update source/event_hub.ts Co-authored-by: Lars Johansson <gismya@gmail.com> * Update test/event_hub.test.js Co-authored-by: Lars Johansson <gismya@gmail.com> * Update source/event_hub.ts Co-authored-by: Lars Johansson <gismya@gmail.com> * Update source/event_hub.ts Co-authored-by: Lars Johansson <gismya@gmail.com> * Update event_hub.ts Incorrectly copied the fix when suggesting line by line. * Removed changes. --------- Co-authored-by: Lars Johansson <gismya@gmail.com>

Author: ffMathy

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();
+ });
 });