Method executeTask

packages/piscina/src/framework.ts:53–113

* 根据 handler 名称执行任务

(
    handler: string,
    payload?: T
  )

   * 根据 handler 名称执行任务
   */
  async executeTask<T = any, R = any>(
    handler: string,
    payload?: T
  ): Promise<R> {
    const TaskClass = this.taskHandlers.get(handler);

    if (!TaskClass) {
      throw new MidwayCommonError(
        `Task handler "${handler}" not found. Did you forget to use @PiscinaTask('${handler}') decorator?`
      );
    }

    const ctx = this.app.createAnonymousContext();
    const traceService = this.applicationContext.get(MidwayTraceService);
    const traceMetaResolver = (this.configurationOptions as any)?.tracing?.meta;
    const traceEnabled =
      (this.configurationOptions as any)?.tracing?.enable !== false;
    const traceExtractor = (this.configurationOptions as any)?.tracing
      ?.extractor;
    const entryCarrier =
      typeof traceExtractor === 'function'
        ? traceExtractor({
            ctx,
            request: payload,
            custom: {
              handler,
            },
          })
        : {};
    return await traceService.runWithEntrySpan(
      `piscina ${handler}`,
      {
        enable: traceEnabled,
        carrier: entryCarrier,
        attributes: {
          'midway.protocol': 'piscina',
          'midway.piscina.handler': handler,
        },
        meta: traceMetaResolver,
        metaArgs: {
          ctx,
          carrier: entryCarrier,
          request: payload,
          custom: {
            handler,
          },
        },
      },
      async () => {
        const taskInstance = await ctx.requestContext.getAsync<any>(TaskClass);

        if (!taskInstance || typeof taskInstance.execute !== 'function') {
          throw new MidwayCommonError(
            `Task "${handler}" must implement execute method`
          );
        }

        return await taskInstance.execute(payload);