Unverified Commit 4096bb70 authored by Hong Minhee's avatar Hong Minhee
Browse files

Refactor inbox listener dispatcher

parent 8d28f99f

federation/handler.ts

+14 −23
Original line number Diff line number Diff line
@@ -313,10 +313,9 @@ export interface InboxHandlerParameters<TContextData> { 
  kv: KvStore;

  kvPrefix: KvKey;

  actorDispatcher?: ActorDispatcher<TContextData>;

  inboxListeners: Map<

    new (...args: unknown[]) => Activity,

    InboxListener<TContextData, Activity>

  >;

  inboxListenerDispatcher: (

    activity: Activity,

  ) => InboxListener<TContextData, Activity> | null;

  inboxErrorHandler?: InboxErrorHandler<TContextData>;

  onNotFound(request: Request): Response | Promise<Response>;

  signatureTimeWindow: Temporal.DurationLike;
@@ -330,7 +329,7 @@ export async function handleInbox<TContextData>( 
    kv,

    kvPrefix,

    actorDispatcher,

    inboxListeners,

    inboxListenerDispatcher,

    inboxErrorHandler,

    onNotFound,

    signatureTimeWindow,
@@ -430,13 +429,8 @@ export async function handleInbox<TContextData>( 
    });

    return response;

  }

  // deno-lint-ignore no-explicit-any

  let cls: new (...args: any[]) => Activity = activity

    // deno-lint-ignore no-explicit-any

    .constructor as unknown as new (...args: any[]) => Activity;

  while (true) {

    if (inboxListeners.has(cls)) break;

    if (cls === Activity) {

  const listener = inboxListenerDispatcher(activity);

  if (listener == null) {

    logger.error(

      "Unsupported activity type:\n{activity}",

      { activity: json },
@@ -446,9 +440,6 @@ export async function handleInbox<TContextData>( 
      headers: { "Content-Type": "text/plain; charset=utf-8" },

    });

  }

    cls = globalThis.Object.getPrototypeOf(cls);

  }

  const listener = inboxListeners.get(cls)!;

  try {

    await listener(context, activity);

  } catch (error) {

federation/middleware.ts

+21 −1
Original line number Diff line number Diff line
@@ -1389,6 +1389,26 @@ export class Federation<TContextData> { 
    return setters;

  }



  #dispatchInboxListener(

    activity: Activity,

  ): InboxListener<TContextData, Activity> | null {

    // deno-lint-ignore no-explicit-any

    let cls: new (...args: any[]) => Activity = activity

      // deno-lint-ignore no-explicit-any

      .constructor as unknown as new (...args: any[]) => Activity;

    const inboxListeners = this.#inboxListeners;

    if (inboxListeners == null) {

      return null;

    }

    while (true) {

      if (inboxListeners.has(cls)) break;

      if (cls === Activity) return null;

      cls = globalThis.Object.getPrototypeOf(cls);

    }

    const listener = inboxListeners.get(cls)!;

    return listener;

  }



  /**

   * Sends an activity to recipients' inboxes.  You would typically use

   * {@link Context.sendActivity} instead of this method.
@@ -1646,7 +1666,7 @@ export class Federation<TContextData> { 
          kv: this.#kv,

          kvPrefix: this.#kvPrefixes.activityIdempotence,

          actorDispatcher: this.#actorCallbacks?.dispatcher,

          inboxListeners: this.#inboxListeners ?? new Map(),

          inboxListenerDispatcher: this.#dispatchInboxListener.bind(this),

          inboxErrorHandler: this.#inboxErrorHandler,

          onNotFound,

          signatureTimeWindow: this.#signatureTimeWindow,