@abinnovision/nestjs-hatchet (英语)

雀巢JS整合 标签 - 一个分布式工作流程管弦引擎。 提供类型安全的装饰,服务和事件处理,以构建强大的工作流程.

安装

yarn add @abinnovision/nestjs-hatchet @hatchet-dev/typescript-sdk

@hatchet-dev/typescript-sdk 是一个对等依赖,所以您的应用程序控制 SDK 版本。 导入 SDK 等类型 Context 直接从它.

快速启动

1. 配置模块

import { HatchetModule } from "@abinnovision/nestjs-hatchet";
import { Module } from "@nestjs/common";

@Module({
  imports: [
    HatchetModule.forRoot({
      config: {
        token: process.env.HATCHET_CLIENT_TOKEN,
        tls_config: { tls_strategy: "none" },
      },
      worker: { name: "my-worker" },
    }),
  ],
})
export class AppModule {}

工人通过Nest的停机钩被拦住 等待工作 结束。 打电话 app.enableShutdownHooks() 所以这也发生在 SIGTERM编号 :SIGINT。 。 。 . 让Hatchet SDK自己处理这些信号 设定 handleKill: true 内 工人的选择.

2. 界定任务

import { Host, Task, TaskCtx, taskHost } from "@abinnovision/nestjs-hatchet";
import { z } from "zod";

@Host({ name: "process-data" })
export class ProcessDataTask extends taskHost(z.object({ data: z.string() })) {
  @Task({})
  public async task(ctx: TaskCtx<typeof this>) {
    return { result: ctx.input.data.toUpperCase() };
  }
}

3. 登记和举报

import { Client, HatchetModule, taskRef } from "@abinnovision/nestjs-hatchet";
import { Injectable, Module } from "@nestjs/common";

import { ProcessDataTask } from "./process-data.task";

@Injectable()
export class MyService {
  constructor(private readonly client: Client) {}

  async process(data: string) {
    return this.client.run(taskRef(ProcessDataTask), { data });
  }
}

@Module({
  imports: [HatchetModule.forFeature(taskRef(ProcessDataTask))],
  providers: [ProcessDataTask, MyService],
})
export class ProcessDataModule {}

forFeature() 只向工人登记宿主。 每个主机也必须列出 输入 providers 模块,否则工人无法解决这个问题 应用程序无法启动 .

核心概念

任务

采用单一可执行方法的独立工作单位.

@Host({ name: "cleanup" })
export class CleanupTask extends taskHost() {
  @Task({})
  public async task(ctx: TaskCtx<typeof this>) {
    return { cleaned: true };
  }
}

工作流量

具有任务依赖性的多步骤进程.

@Host({ name: "order-workflow" })
export class OrderWorkflow extends workflowHost(
  z.object({ orderId: z.string() }),
) {
  @WorkflowTask<typeof OrderWorkflow>({ parents: [] })
  public async validate(ctx: WorkflowCtx<typeof this>) {
    return { valid: true };
  }

  @WorkflowTask<typeof OrderWorkflow>({ parents: ["validate"] })
  public async process(ctx: WorkflowCtx<typeof this>) {
    const validation = await ctx.parent(this.validate);
    return { processed: validation.valid };
  }
}

活动

具有schema验证的类型安全事件定义.

// Define event
export const UserCreatedEvent = defineEvent(
  "user:created",
  z.object({ userId: z.string(), email: z.string().email() }),
);

// Workflow triggered by event
@Host({ name: "user-handler", onEvents: ["user:created"] })
export class UserHandlerWorkflow extends workflowHost() {
  @WorkflowTask<typeof UserHandlerWorkflow>({ parents: [] })
  public async handle(ctx: WorkflowCtx<typeof this>) {
    if (UserCreatedEvent.isCtx(ctx)) {
      return { userId: ctx.input.userId };
    }
  }
}

// Emit event
await client.emit(UserCreatedEvent, {
  userId: "123",
  email: "[email protected]",
});

高级使用

非锁定执行

const runRef = await ctx.run(taskRef(SomeTask), input, { wait: false });
// Continue without waiting...
const result = await runRef.output; // Await when needed

散装事件排放

await client.emit(OrderEvent, [
  { orderId: "1", total: 100 },
  { orderId: "2", total: 200 },
]);

任务截取器

任务截取器将每个任务执行包在工人身上,包括每个工作流程的任务,例如: 记录或追踪。 传过来 forRoot()编号 :forRootAsync() 并把他们列为供养者,因为他们 由应用程序而不是模块即时解决。 生产方 电话,看 客户截取器。 。 。 .

import {
  BaseCtx,
  HatchetModule,
  TaskInterceptor,
} from "@abinnovision/nestjs-hatchet";
import { Injectable, Logger, Module } from "@nestjs/common";

@Injectable()
export class LoggingInterceptor extends TaskInterceptor {
  private readonly logger = new Logger(LoggingInterceptor.name);

  public async intercept<T>(ctx: BaseCtx<any>, next: () => Promise<T>) {
    this.logger.log(`Running ${ctx.fromSDK.taskName()}`);
    return next();
  }
}

@Module({
  imports: [
    HatchetModule.forRoot({
      config: {},
      worker: { name: "my-worker" },
      taskInterceptors: [LoggingInterceptor],
    }),
  ],
  providers: [LoggingInterceptor],
})
export class AppModule {}

客户截取器

客户端截取器包装制片方呼叫 Client数字 : client.emit (和 已贬值 emitBulk和 client.run扩展 ClientInterceptor 执行 interceptEmit, (中文(简体) ), interceptRun或两者兼有。 以 clientInterceptors 选项 forRoot()编号 :forRootAsync() 并列出在 providers。 。 。 .

以下例子宣传了一种关联性标识。 客户截取器写到 additionalMetadata并有一个任务拦截器 工人从 运行中任务的元数据 .

import { AsyncLocalStorage } from "node:async_hooks";
import { randomUUID } from "node:crypto";

import {
  BaseCtx,
  ClientInterceptor,
  EmitCtx,
  EmitOptions,
  HatchetModule,
  TaskInterceptor,
  RunCtx,
  RunOptions,
} from "@abinnovision/nestjs-hatchet";
import { Injectable, Module } from "@nestjs/common";

export const correlation = new AsyncLocalStorage<{ id: string }>();

@Injectable()
export class CorrelationClientInterceptor extends ClientInterceptor {
  public override async interceptEmit(
    ctx: EmitCtx,
    next: (options?: EmitOptions) => Promise<unknown>,
  ) {
    const id = correlation.getStore()?.id;
    if (!id) return next();

    return next({
      ...ctx.options,
      additionalMetadata: {
        correlationId: id,
        ...ctx.options?.additionalMetadata,
      },
    });
  }

  public override async interceptRun(
    ctx: RunCtx,
    next: (options?: RunOptions) => Promise<unknown>,
  ) {
    const id = correlation.getStore()?.id;
    if (!id) return next();

    return next({
      ...ctx.options,
      additionalMetadata: {
        correlationId: id,
        ...ctx.options?.additionalMetadata,
      },
    });
  }
}

@Injectable()
export class CorrelationInterceptor extends TaskInterceptor {
  public async intercept<T>(ctx: BaseCtx<any>, next: () => Promise<T>) {
    const id = ctx.fromSDK.additionalMetadata()?.correlationId ?? randomUUID();
    return correlation.run({ id }, next);
  }
}

@Module({
  imports: [
    HatchetModule.forRootAsync({
      useFactory: () => ({ config: {}, worker: { name: "my-worker" } }),
      clientInterceptors: [CorrelationClientInterceptor],
      taskInterceptors: [CorrelationInterceptor],
    }),
  ],
  providers: [CorrelationClientInterceptor, CorrelationInterceptor],
})
export class AppModule {}

行为:

• 每个拦截器每次呼叫一次 散装弹出(数列有效载荷)和 run 与 数组输入每个计为一个调用 .
• 截取器只改变选项。 有效载荷或输入不可改变.
• next() 保留当前选项. next(options) 替换如下: 截取器和SDK呼叫,所以散开 ctx.options 以保持呼叫者值.
• next(undefined) 行为就像 next()。要放弃所有选项,请通过 {}。 。 。 .
• wait 和返回形状 client.run 不能被推翻. RunOptions 不包括 wait。 。 。 .
• 与 wait: true (默认), interceptRun 覆盖全部运行,包括 等待输出。 与 wait: false 它只包着花序.
• childKey 没有任何影响 Client.run。 。 。 .
• 拦截器按阵列顺序运行,第一个是最外层. 拦截器可能 跳过 next() 把电话传短路.
• 客户端截取类必须列出于 providers失败 应用拖带.
• 儿童运行始于 ctx.run 继承父方的 additionalMetadata 打开 弹夹引擎 >= v0.104.0所以他们不需要客户拦截器.

洞穴:

• 通话 client.emit 或者说 client.run 在一个拦截器内部 重新进入链. 保护自己避免重犯.
• 不支持请求范围拦截器提供者, 与任务相同 拦截器.

帮助方法

@Host({ name: "my-task" })
export class MyTask extends taskHost(schema) {
  @Task({})
  public async task(ctx: TaskCtx<typeof this>) {
    return this.helper(ctx);
  }

  private async helper(ctx: HelperCtx<typeof this>) {
    return { result: ctx.input.value };
  }
}

API 参考

模块

模块选项( extras to forRoot编号 :forRootAsync:

装饰者

主机厂

参考资料

客户端

呼吁 emit, (中文(简体) ), emitBulk 财务报告和审定财务报表 run 通过注册 客户端截取器。 。 。 .

上下文类型

上下文属性

ctx.input       // Validated input data
ctx.fromSDK     // Underlying Hatchet SDK context
ctx.run(ref, input, opts ?)  // Invoke other tasks/workflows
ctx.parent(method)          // Get parent task output (WorkflowCtx only)

许可证

阿帕奇 - 2.0