@btravstack/时空-工人

这个 半个工人 时间 @btravstack/core 开始 工作流程是客户的工作,属于 @temporal-contract/client一个模块 提供运行时间、配置和连接; 每集一个单位 和释放内核的排出物 内核 截止时间,而不是时间 shutdownForceTime。 。 。 .

📖 文档 • 妇女 参考文献 • 妇女 API 参考

pnpm add @btravstack/temporal-worker @btravstack/core @btravstack/config @btravstack/di unthrown \
  @temporalio/worker @temporalio/activity @temporalio/common \
  @temporal-contract/worker@^8.0.0-beta.7 @temporal-contract/contract@^8.0.0-beta.7

所有9个都是同级依赖——安装. 节点 >=22.22。本套装仅限无害环境管理.

实例

import { runMain, Logger } from "@btravstack/core";
import { Provider } from "@btravstack/di";
import {
  ActivityInput,
  TemporalActivities,
  TemporalModule,
} from "@btravstack/temporal-worker";
import { P } from "unthrown";

// The module forked around every activity attempt, SEEDED with the validated
// input on `ActivityInput(contract)`. The tenant the contract carries is
// claimed here once — not at each activity — and the application's vertical
// is composed over it, so `PlaceOrder` is bound before any activity runs.
const ActivityUnit = Module("ActivityUnit")({
  needs: [ActivityInput(contract)],
  imports: [OrderApplicationModule],
  provides: [
    Provider(Tenant)({
      inject: { input: ActivityInput(contract) },
      sync: ({ input }) => TenantId(input.tenantId),
    }),
  ],
  exports: [Tenant, PlaceOrder],
});

// The application's half: its activities, on the starter's own activities
// port, typed by the contract (a worker serves one activities record, so
// there is nothing to name). What the leaves read off `context.unit` is
// declared once, beside `inject`.
const orderActivities = TemporalActivities(contract)({
  unit: { place: PlaceOrder },
  sync: () => ({
    placeOrder: {
      place: ({ errors, context, input }) =>
        context.unit.place
          .execute(input.orderId, input.quantity)
          .mapErrCases((matcher) =>
            matcher
              .with(P.tag("DuplicateOrder"), (error) =>
                errors.OrderAlreadyPlaced({ id: error.id }),
              )
              .with(P.tag("InvalidOrderId"), (error) =>
                errors.InvalidOrderId({ id: error.id }),
              )
              .with(P.tag("InvalidQuantity"), (error) =>
                errors.InvalidQuantity({ id: error.id }),
              ),
          ),
    },
  }),
});

// The composition root: a di module, plus the contract, the activities
// provider and the workflow source — and nothing else to know.
const OrderWorker = TemporalModule("OrderWorker")({
  contract,
  activities: orderActivities,
  workflows: {
    workflowsPath: workflowsPathFromURL(import.meta.url, "./workflows.js"),
  },
  imports: [AppModule],
  unit: { activity: ActivityUnit },
});

await runMain(OrderWorker);

TemporalModule 导入启动器 - TemporalRuntime, (中文(简体) ), TemporalConfig 约束从 TEMPORAL_ADDRESS (默认) 127.0.0.1:7233和 TEMPORAL_NAMESPACE (默认) default, (中文(简体) ), TemporalConnection 打开为 a 图表资源,在每个退出路径上关闭——提供活动 并导出运行时端口。 无法回答的服务是一个模型 TemporalUnreachable关闭 1; 启动器呼叫 declareActivitiesHandler 因此它不能履行的合同是 启动 Err而不是缺陷. runtimeInfo() 读取 { taskQueue, namespace } 一旦工人投票就回来.

活动片头 DI 身份包括合同任务队列. 独立工人 合同可以在不共享端口的情况下重新使用密钥;电线活动名称保持不变.

几个工作流程的工人投票可以是几个片子而不是一个 记录 : TemporalWorkflowActivities(contract, key)({ inject: { name: Dep }, unit, fork, sync }) 薄荷a 提供One工作流程活动(或一项合同-全球活动)的提供者, 由键单独键入,以及 TemporalActivities(contract)([...]) 组成一个 组合到同一活动提供方 TemporalModule 采取—— 数组必须覆盖合同宣布的每个密钥, 每个部件自己的端口 仍然必须退伍provides),因为编组提供者的道具是: 不是他们关闭的港口 一个工作流程的活动完全属于 一个切片,编译器只持有的一半: 覆盖被检查, 注射不是,所以两个切片 声称一个关键类型检查在一起。 没错 di的重复提供器缺陷一次建立 两者 块被卸载为 一个图中的供应商——只有一个线,另一个是执行 默默地从未注册过.

一块 unit 命名端口 context.unit, (中文(简体) ), 已解析出装入的模块的每分叉 unit.activity —— 说 以有效输入为种子的 ActivityInput(contract)单位 租户来自援引,而不是环境记录. 上面的记录臂也一样 unit: 记录每个条目 所以一个没有超过一个功能的工人 context.unit 无 先切开. 根与所申报的: 不导出其中一个端口的模块 TemporalModule 拒绝, 命名端口.

活动范围不同于其他名称的作品 模块 fork:它的尝试叉出这个模块,以同样的方式播种,而不是 unit.activity - 列入清单的活动,其中除下列活动外没有租户: 粉丝 - 每一个活动都为一人而行动。 这个是 unit 然后检查 相对于它自己的模块 在作品的呼唤, 模块未满足的需求成为 根的( 由 start),并在模块名下被覆盖: overridden(root, providers, { unit: { TenantUnit: [...] } })。 。 。 .

它自己的测试套房需要一个 Docker 守护进程: 与共享相对应的规格 temporalio/temporal 开发服务器容器 internal/test-infra 每台机器开始一次 工作空间再利用。 这里没有别的事.

诊断性披露

withTemporalDiagnostics(options, work) temperal 的诊断默认值 而不急于创建其本土运行时间。 在回电室里启动工人 并保持开放,直到拥有的工人和连接关闭。 启动拒绝或 首个本地连接之前的缺陷, 然后不留下诊断 让过程保持活力 帮助者返回一个 AsyncResult 保存 callback 的结果, 在成功时恢复自有默认值、 建模出错或缺陷, 同时保留 活动运行时实例和替换配置.

import { createLogger, diagnosticPolicy, jsonSink } from "@btravstack/observability";
import { withTemporalDiagnostics } from "@btravstack/temporal-worker";
import { fromSafePromise } from "unthrown";

await withTemporalDiagnostics(
  { logger: createLogger(jsonSink()), diagnostics: diagnosticPolicy() },
  () => fromSafePromise(runMain(OrderWorker)),
).get();

TemporalDiagnosticsOptions 持有 { logger, diagnostics, nativeLogLevel? }。 。 。 . SDK、工作流程、活动和原生核心日志到达共享 DiagnosticPolicy· ; 本地动词默认为 WARN,而内核拥有进程信号。 一个 现有运行时间或重叠范围是一个缺陷。 不要打电话 Runtime.install 在回调内:明确的 SDK 安装会超越被范围化的默认值. Temperal 拥有连接拆解和日志冲洗; 帮助者永远不会关闭一个 活动运行时间或修复本地资源在运行时间创建后泄露.

installTemporalDiagnostics(options) 仍然是热心全过程的安装器 返回时 Result<void, never>延迟或重复安装缺陷。 不能这样 连接启动失败后释放闲置的运行时间; 偏好范围化 应用程序启动的助手.

temporalDiagnosticLogger(logger, diagnostics) 返回临时 SDK 日志 配置应用程序 Runtime.install 自己来 本地日志转发 仍然需要运行时配置。 两个帮手都接受申报的保单 内 @btravstack/core· ; @btravstack/observability 提供 diagnosticPolicy()。 。 。 . 见 诊断性披露 用于构成和单独的有效载荷/故障边界.

时间表

@btravstack/temporal-worker/schedule 出口 ensureSchedule - 键入 调度客户端 create,使一能, 所以部署再次运行 更新计划而不是失败 ScheduleAlreadyExistsError。 。 。 . @temporal-contract/client 只能通过此方式达到的可选对等点 子路径。 为什么排定的工作要花一个组,当它不应该是 此堆栈 : 运行进度表。 。 。 .

测试

@btravstack/temporal-testing 运行一个工人的规格 相对于每个Docker主机一个真实的Timeoral服务器,通过测试共享 进程,每个 spec 文件都有命名空间。 这是它本身的依赖, 所以这个包裹没有试验密码 见 试用临时工。 。 。 .

选项

• taskQueue: 针头 TEMPORAL_TASK_QUEUE,默认为合同队列;参见 参考文献。 。 。 .

TemporalModule(name)({...}) 选择启动者的选项+ activities 和模块列表imports, (中文(简体) ), provides, (中文(简体) ), exports, (中文(简体) ), needs) (中文(简体) ). 这是 唯一不导出创建者的方法 :

完整的表格——需要/可选的、默认的和推理的——继续存在 参考页, (中文(简体) ), 这张名单是详细的.

确保连接

TLS 、 mTLS 和一个 API 密钥从 Tempolar 下的环境配置 拥有自己的名字。 对于mTLS:

export TEMPORAL_TLS=true
export TEMPORAL_TLS_SERVER_CA_CERT_PATH=/var/run/secrets/temporal/ca.crt
export TEMPORAL_TLS_SERVER_NAME=temporal.example.internal
export TEMPORAL_TLS_CLIENT_CERT_PATH=/var/run/secrets/temporal/tls.crt
export TEMPORAL_TLS_CLIENT_KEY_PATH=/var/run/secrets/temporal/tls.key

或者说 TEMPORAL_API_KEY_PATH (或 TEMPORAL_API_KEYTLS 打开,除非 TEMPORAL_TLS 说不然. 重读已挂载的文件 TEMPORAL_CREDENTIALS_RELOAD_MS (默认) 60000: 更新的证书将工人移动到新的连接上,而不 在打开的键上设置了更新的键。 组合它不能 连接为 a ConfigInvalid 在靴子上。 这个 参考文献 有规则和限制.

申请可通过 apiKey 端口 执行 TemporalApiKeySource.read()来源拥有象征性的收购, 缓存和过期; 启动人投票 它与证书 重新加载间隔. 源需要 TLS 并排除内置/文件 API 密钥和 mTLS. 见 参考文献 和失败的行为.

下水道

worker.shutdown() 立刻停止投票,但是 run() 只在 最后一次飞行活动已经完成——在Temperal自己完成 shutdownForceTime。 。 。 . Serving.drain(signal) 等待内核最后期限的比赛和 返回,以先来者为准:内核恢复其线程 报告工作 abandoned工人们不停地在Temperal的店里 时钟直到进程退出. @temporalio/worker 没有公开强迫 关闭,所以"停止等"就是升级. Result 活性衰竭是 declareActivitiesHandler这里刻意不重复 剩下的钱 现在在 文档网站。 。 。 .

TemporalModule 自动公开导出 Logger 财务报告和审定财务报表 HealthChecks 端口,包括整个模块再出口和在根上申报的供应商. 它使私人基础设施隐蔽,商业出口明确.

许可证

麻省理工学院 贝诺伊特旅行社