Camunda 8 爪哇脚本 SDK

SDK API 文档。 。 。 . 组合 API 客户端全 API 文档。 。 。 .

这是Camunda 8 JavaScript SDK的官方版本. 它被写入TypeScript并运行于Node.js上. 看为什么 此 SDK 不在网页浏览器中运行。 。 。 .

如果您需要在网页浏览器中运行应用程序,请查看是否使用 @camunda8/orchestration-cluster-api 直接包.

见 获得开始示例 在Camunda Docs的快速开始.

我该用哪个包裹?

这个SDK为Camunda 8.的各种版本提供API客户端. 如果你在Camunda8.8或以后做一个绿地项目,那么你应该考虑直接使用 @camunda8/orchestration-cluster-api 软件包。 该套件为Camunda 8 Orchestration Cluster API提供了一个客户端,一个具有全功能的REST API.

这个SDK包括那个客户端,但依靠这个SDK包拉出其他API客户端,如果你不使用这些客户端,就会暴露出几个因素:

• 与您的应用程序无关的依赖关系的依赖大小增加
• 只有节点J 另一个软件包在浏览器中运行.

如何选择?

使用 @camunda8/sdk 若为 :

• 你需要GRPC API来做工作流.
• 你的服务器目标在8.7或更早.
• 您正在逐步迁移一个已有的应用程序, 使用8. 8 管弦乐团 API .

使用 @camunda8/orchestration-cluster-api 直接软件包,如果:

• 您正在开发一个绿地应用程序.
• 你的服务器目标在8.8左右.
• 您不需要使用 gRPC API .

我应该用哪个API?

SDK为几个Camunda 8 API提供客户端.

如果你在Camunda8.8.0或以后做一个绿地项目,那么你应该使用Camunda管弦乐团 API getOrchestrationClusterApiClient。这是一个 REST API,在一个API 表面提供完整的集群功能。 此客户端被强烈地打入,为请求和响应字段提供了强烈的类型(例如: ProcessInstanceKey, (中文(简体) ), ProcessDefinitionId(见e/cn.4/sub.2/2002/l.48等).

要在现有项目中与以前的客户一起逐步采用该客户端,您可以拨打 getOrchestrationClusterApiClientLoose 以获取一个不强烈输入请求和响应字段的客户端( 它们仍然是原始 scalar) string 类型).

"支持"是什么意思?

这是Camunda Nodejs为Camunda平台8提供的官方支持.

节点(Node.js SDK)在Camunda平台8发布后不会始终支持其所有特性. 平台发布的完整API覆盖将落后于平台发布.

执行特征的优先次序受到客户需求的影响.

语义版本

SDK包跟踪了Camunda平台8的小版本. 支持当前平台小版本特性的特性发布导致SDK的补丁发布.

在您的工程中使用 SDK

安装 SDK 作为依赖 :

npm i @camunda8/sdk

使用量

Camunda 8的功能通过组件API的专用客户端被曝光. Camunda 8.8推荐的API客户端是管弦乐团 API,使用 CamundaClient 返回时间 getOrchestrationClusterApiClient()。 。 。 .

import { Camunda8 } from '@camunda8/sdk'

const c8 = new Camunda8()

// Camunda 8 Orchestration Cluster API (STRICT client) - recommended from 8.8.0
// Provides strongly typed request & response IDs (e.g. ProcessInstanceKey, ProcessDefinitionId)
const orchestration = c8.getOrchestrationClusterApiClient()

// Camunda 8 Orchestration Cluster API (LOOSE client) - progressive adoption variant
// Same methods, but all IDs are plain string. Start here in existing codebases, then
// migrate to the strict client when convenient.
const orchestrationLoose = c8.getOrchestrationClusterApiClientLoose()

配置

SDK的配置可以通过环境变量和明确配置的组合来实现 Camunda8 建筑师.

建议的配置方法是使用零相通构件,将配置完全从环境中水分化:

const camunda = new Camunda8()

这使得您可以将配置的顾虑与您的代码清洁地区分开来,并将其完全定位于环境变量管理中.

可找到所有配置参数(环境变量)的完整文档 这里。 。 。 .

任何配置传递到 Camunda8 构造器在环境的任何配置上被合并. 配置对象字段和环境变量的名称完全相同.

管弦乐组 API: 严格 vs Loose 客户端

从Camunda 8.8起,统一的Orchestration Cluster API是首选的界面. 这个SDK通过两种工厂方法将其曝光,以便实现平稳而递增地迁移:

为什么有两个客户?

品牌(名义)类型能显著地减少不相干ID的意外混合(例如,在进程实例密钥预期的情况下通过决定定义id). 然而,在现有的代码库中采用这些模块可能需要在许多模块中进行重构。 松散的客户端让你立即开始拨打新的端点,而不触及这些模块. 然后,在更新代码时,可以递增地迁移到严格的客户端.

迁移路径

1. 在现有 v1 组件客户端的同时引入松散客户端 : const ocaLoose = c8.getOrchestrationClusterApiClientLoose()。 。 。 .
2. 将旧客户端的呼叫(Operate / Tasklist / 优化读取操作,Zeebe REST用于工作流程互动)替换为使用松散客户端的等效 Orchestration API呼叫.
3. 在方便时, 切换一个模块到严格的客户端: 更改导入到使用 getOrchestrationClusterApiClient() 并地址 TypeScript 编译出错,方法是采用有品牌的ID类型(通常是通过宣传该类型代替) string) (中文(简体) ).
4. 一旦所有模块都与严格的客户端进行干净的编译后,就移除宽松的客户端使用.

REST 地址正常化(/v2)

设定 ZEEBE_REST_ADDRESS 没有版本后缀; SDK 将附加 /v2 准确一次:

export ZEEBE_REST_ADDRESS='http://localhost:8888'

有效基址结果 : http://localhost:8888/v2。 。 。 .

如果包含 /v2 已经 :

export ZEEBE_REST_ADDRESS='http://localhost:8888/v2'

保留(不重复)。 在正常化前将剥去被划伤的鞭子(http://localhost:8888/ - > (中文) http://localhost:8888/v2) (中文(简体) ).

在客户端中选择

使用严格的客户端用于新代码,库,或类型安全能提高清晰度的例子. 当 :

• 您需要快速的滴入替换而不改变已有的函数签名 .
• 您正在逐渐迁移许多模块.
• 你想推迟采用品牌的ID类型,直到以后.

两个客户端都暴露出相同的方法名称和行为;只有TypeScript表面不同. 运行时的语义和响应在其他方面是等同的.

示例( Strit vs Loose)

import { Camunda8 } from '@camunda8/sdk'
const c8 = new Camunda8()

// Strict client (preferred once migrated)
const oca = c8.getOrchestrationClusterApiClient()
const proc = await oca.createProcessInstance({
  processDefinitionId: ProcessDefinitionId.assumeExists('order-process'),
  variables: { orderId: 'A123' },
})
// proc.processInstanceKey is a branded type (not a plain string)

// Loose client (progressive adoption)
const ocaLoose = c8.getOrchestrationClusterApiClientLoose()
const procLoose = await ocaLoose.createProcessInstance({
  processDefinitionId: 'order-process',
  variables: { orderId: 'B456' },
})
// procLoose.processInstanceKey is a plain string

将代码从松动转换为严格的代码时,删除不必要的铸件,例如 (key as string) 并允许编译器指导品牌类型表面的校正.

提升遗产字符串标识

要在不立即将每个模块切换到严格的客户端的情况下将使用普通字符串进行标识的现有代码迁移,可以"将"这些字符串递增地"切入品牌的关键类型.

所有有品牌的密钥起重机命名空间归在下面 OrchestrationLifters 所以根 SDK 导出表面保持整洁 :

import { OrchestrationLifters } from '@camunda8/sdk'

// Suppose you previously passed raw strings:
const rawProcessInstanceKey = '2251799813685249'

// Lift it to the branded type (runtime value is still a string, but type safety now applies):
const processInstanceKey = OrchestrationLifters.ProcessInstanceKey.assumeExists(rawProcessInstanceKey)

// Use with strict client
const c8 = new Camunda8()
const oca = c8.getOrchestrationClusterApiClient()
await oca.cancelProcessInstance({ processInstanceKey })

常见助手 : assumeExists(value: string) 将已知有效遗留的标识符变成其品牌等同. 如果您确定 ID 指的是一个已有的实体( 抬起器不执行网络验证; 它纯粹是名义的打字) .

建议的渐进办法:

1. 首先要替换输出值(例如来自 createProcessInstance),严格客户端已经返回了品牌类型.
2. 在您仍然收到遗留字符串的边界(配置、环境、持久性层、遗留的SDK模块)引入升起器.
3. 一旦这些边界本身就产生了品牌类型,就删除升起器.

所有可用的起重机(例): ProcessInstanceKey, (中文(简体) ), ProcessDefinitionId, (中文(简体) ), JobKey, (中文(简体) ), ElementInstanceKey, (中文(简体) ), DecisionDefinitionKey, (中文(简体) ), FormKey, (中文(简体) ), UserTaskKey, (中文(简体) ), VariableKey等。见 OrchestrationLifters 导出完整列表.

关于Int64如何处理 JavaScript SDK 的说明

Camunda 8.6 和早期的实体密钥作为 int64 数字。 范围 int64 延伸至无法由 JavaScript 代表的数字 number 类型。 为了对付这个, int64 密钥由 SDK 序列化为 JavaScript string 类型。 见 这个问题 更多细节.

对于 int64 类型在时间之前未知的值,例如工作变量,您可以通过附加注释的数据传输对象(DTO)来可靠地解码它们。 见关于 进程变量打字如果未指定 DTO, SDK 的默认行为是将所有数字序列化为 JavaScript number,如果在运行时间检测到一个无法准确表示为 JavaScript 的数字值,则丢弃一个例外 number 类型(即大于2^53-1的值).

授权

对API的呼叫可以通过一些策略授权. 最常见的是OAuth——通过客户端ID/秘密对对交换获得的标志.

• 默认的 Camunda SaaS 和 Self-Managed 被配置为使用 OAuth 并带有符号交换. 在大多数情况下,你会使用 OAUTH 并仅提供配置。 令牌交易所及其生命周期由SDK在这个策略中进行管理. 这个传出标志 authorization 请求头.
• 如果您在Nginx倒置代理服务器的后端安放有基本认证的网关,您将使用 BASIC 战略。 这将登录证书添加为 authorization 请求头.
• 对于C8Run 8.7,您需要使用 COOKIE 战略。 这管理从登录端点获得的会话 cookie, 并将其作为 cookie 标题添加到请求中 .
• 更多定制时,您可以使用 BEARER 战略。 这个策略允许你动态提供 Bearer 直接标出。 当前设置的符号作为 authorization 请求头.
• 如果您有更先进的使用大小写(例如,需要添加带有特定值的特定头来认证代理网关),您可以构建自己的AuthProvider并将其传递给构建器.
• 您也可以完全禁用页眉认证并使用 mTLS (客户证书) —— 或者完全没有认证—— 使用 NONE 战略.

关于这些设想的详情,见下文有关章节.

禁用自动

要禁用 Auth,请设置环境变量 CAMUNDA_AUTH_STRATEGY=NONE。例如,在开发环境中运行一个最小的 Zeebe 经纪人时,您可以使用它。 您也可以在使用x509 mTLS证书进行认证时使用此功能(参见mTLS上一节).

奥奥特

如果您的平台是使用 OAuth 令牌交换( Camunda SaaS 或 具有身份的自我管理) 进行保护, 则至少提供以下配置字段, 或者通过 Camunda8 构造器或环境变量:

CAMUNDA_AUTH_STRATEGY=OAUTH
ZEEBE_GRPC_ADDRESS=...
ZEEBE_REST_ADDRESS=...
ZEEBE_CLIENT_ID=...
ZEEBE_CLIENT_SECRET=...
CAMUNDA_OAUTH_URL=...

为获得Camunda SaaS管理API或Camunda SaaS建模 API的代号,设置如下:

CAMUNDA_AUTH_STRATEGY=OAUTH
CAMUNDA_CONSOLE_CLIENT_ID=...
CAMUNDA_CONSOLE_CLIENT_SECRET=...

切换

OAuth 令牌被缓存入膜和上盘. 例如,磁盘缓存在重新启动或滚动工人时有助于防止符号终点饱和。 他们都可以点击缓存,而不是要求新的代币.

您可以设定关闭磁盘缓存 CAMUNDA_TOKEN_DISK_CACHE_DISABLE 真实的。 这将只缓存记忆中的符号 .

默认情况下,令牌缓存目录是 $HOME/.camunda。您可以通过提供完整文件路径值来指定不同的目录 CAMUNDA_TOKEN_CACHE_DIR。 。 。 .

以下是通过构建器指定不同缓存目录的例子:

import { Camunda8 } from '@camunda8/sdk'

const c8 = new Camunda8({
  CAMUNDA_TOKEN_CACHE_DIR: '/tmp/cache',
})

如果缓存目录不存在,SDK会试图创建它(重复地). 如果 SDK 无法创建它,或者目录存在但无法被您的应用程序写入, SDK 将丢出一个例外 .

基本认证

要使用基本的认证,请通过环境或通过构建器在代码中明确设定下列值:

CAMUNDA_AUTH_STRATEGY=BASIC
CAMUNDA_BASIC_AUTH_USERNAME=....
CAMUNDA_BASIC_AUTH_PASSWORD=...

饼干 Auth

对于C8Run和8.7,您需要使用 Cookie 认证。 。 。 .

要使用 cookie 认证, 请设置以下值 :

CAMUNDA_AUTH_STRATEGY=COOKIE

# Optional configurable values - these are the defaults
CAMUNDA_COOKIE_AUTH_URL=http://localhost:8080/api/login
CAMUNDA_COOKIE_AUTH_USERNAME=demo
CAMUNDA_COOKIE_AUTH_PASSWORD=demo

熊熊托肯奥特

BEARER认证策略作为低等原始提供,用于其他不涵盖的高级用案. 在这种情况下,你需要通过一个 authorization 标题为 Bearer 在请求时表示,但它不是由支持的 OAuth 符号交换端点所发行。 在这种情况下,您可以实施一个符号交换机制,并管理该符号的生命周期,并且通过一个 BearerAuthProvider。 。 。 .

要使用您已经获得的且在应用程序存在期间不需要动态更新的熊鼠标志,只需设置以下值:

CAMUNDA_AUTH_STRATEGY=BEARER
CAMUNDA_OAUTH_TOKEN=....

为了在运行时动态地刷新无记名标志(例如,当它过期并获得新的标志时),您可以在 BearerAuthProvider 由您控制 :

import { Camunda8, Auth } from '@camunda8/sdk'

const bearerAuth = new Auth.BearerAuthProvider()
const c8 = new Camunda8({ oauthProvider: bearerAuth }) // All clients and workers will use bearerAuth
// ... after obtaining a new token
bearerAuth.setToken('SOMETOKENVALUE....') // Dynamically update the bearer token value

高级自定义信头

您可以通过执行在所有请求中添加任意信头 IOAuthProvider数字 :

import { Camunda8, Auth } from '@camunda8/sdk'

class MyCustomAuthProvider implements Auth.IOAuthProvider {
  async getToken(audience: string) {
    // here we give a static example, but this class may read configuration,
    // exchange credentials with an endpoint, manage token lifecycles, and so forth...
    // Return an object which will be merged with the headers on the request
  return {
    'x-custom-auth-header': 'someCustomValue',
    }
  }
}

const customAuthProvider = new MyCustomAuthProvider()
const c8 = new Camunda8({ oauthProvider: customAuthProvider })

Camunda SaaS 持续401个停机坪

Camunda SaaS继续返回 401 Unauthorized 用于错误配置的证书和观众配对,但后续请求延迟30秒(服务器侧冷却)。 SDK在第一个SaaS上 积极创建了持久的"坑"标记 401 防止网络呼叫延迟.

行为:

• 第一届 401 页:1 (clientId, clientSecret, audienceType) 创建 oauth-401-tarpit-<clientId>-<audience>-<hash>.json 缓存目录中 ($HOME/.camunda (默认情况下). <hash> 是一个已切换的 PBKDF2( 100K 迭代) 密钥散列 .
• 后续 getHeaders() 呼唤那个拖拉机 立即扔出一个拖拉机错误 而不击中符号端点.
• 卧铺不会自动过期.

隔离旋转( R) :

• 按键于客户端Id + 秘密散列 + 受众;不同的受众是独立的.
• 旋转密钥会产生新的散列(如果需要,可以手动去除旧的停机坪文件).

清除停机坪 :

import { Auth } from '@camunda8/sdk'
Auth.OAuthProvider.clear401Tarpit({
  clientId: 'myClientId',
  clientSecret: 'currentSecret',
  audienceType: 'ZEEBE',
})

清除后, 下一个呼叫会尝试真正的象征性请求 .

后退交互 :

• Token端点后起倒出/故障计数器被压制为停机坪401s以快速地解决地表配置问题.
• 其他错误类型保留了正常的后退行为.

观察性:

• DEBUG=camunda:oauth 日志创建 : Created persistent 401 tarpit file ...
• 检查缓存目录 oauth-401-tarpit-* 文档.

补救措施:

1. 修复证书/观众配置.
2. 通过帮助清除寄存器文件(或手动删除) .
3. 重试令牌收购.

这种持续的停机坪行为对于这个SDK版本的SaaS环境来说是自动的.

电话

如果您正在使用自签名证书, 您可以使用配置参数/ 环境变量提供自签名证书路径 : CAMUNDA_CUSTOM_ROOT_CERT_PATH。 。 。 .

毫升数

Zeebe gRPC客户端支持mTLS. 您可以提供 mTLS 证书和密钥 :

CAMUNDA_CUSTOM_CERT_CHAIN_PATH # path to mTLS certificate
CAMUNDA_CUSTOM_PRIVATE_KEY_PATH # path to mTLS (client-side) key

连接配置示例

卡蒙达8 跑8.8

这是配置 Camunda 8 运行 - 当地Camunda 8例,这是最好的开始方式.

export ZEEBE_REST_ADDRESS='http://localhost:8080'
export ZEEBE_GRPC_ADDRESS='grpc://localhost:26500'
export CAMUNDA_AUTH_STRATEGY=NONE

自行管理

这是运行在 Dockerized 自控堆栈上所需的完整环境配置 docker 子目录 :

# Self-Managed
export ZEEBE_GRPC_ADDRESS='grpc://localhost:26500'
export ZEEBE_REST_ADDRESS='http://localhost:8080'
export ZEEBE_CLIENT_ID='zeebe'
export ZEEBE_CLIENT_SECRET='zecret'
export CAMUNDA_OAUTH_STRATEGY='OAUTH'
export CAMUNDA_OAUTH_URL='http://localhost:18080/auth/realms/camunda-platform/protocol/openid-connect/token'
export CAMUNDA_TASKLIST_BASE_URL='http://localhost:8082'
export CAMUNDA_OPERATE_BASE_URL='http://localhost:8081'
export CAMUNDA_OPTIMIZE_BASE_URL='http://localhost:8083'
export CAMUNDA_MODELER_BASE_URL='http://localhost:8070/api'

# Turn off the tenant ID, which may have been set by multi-tenant tests
# You can set this in a constructor config, or in the environment if running multi-tenant
export CAMUNDA_TENANT_ID=''

如果您使用OIDC,则需要 scope 参数随符号请求相接,设置以下变量:

CAMUNDA_TOKEN_SCOPE

以下是通过建筑师而不是通过环境来做到这一点的例子:

import { Camunda8 } from '@camunda8/sdk'

const c8 = new Camunda8({
  ZEEBE_GRPC_ADDRESS: 'grpc://localhost:26500',
  ZEEBE_REST_ADDRESS: 'http://localhost:8080',
  ZEEBE_CLIENT_ID: 'zeebe',
  ZEEBE_CLIENT_SECRET: 'zecret',
  CAMUNDA_OAUTH_STRATEGY: 'OAUTH',
  CAMUNDA_OAUTH_URL:
    'http://localhost:18080/auth/realms/camunda-platform/protocol/openid-connect/token',
  CAMUNDA_TASKLIST_BASE_URL: 'http://localhost:8082',
  CAMUNDA_OPERATE_BASE_URL: 'http://localhost:8081',
  CAMUNDA_OPTIMIZE_BASE_URL: 'http://localhost:8083',
  CAMUNDA_MODELER_BASE_URL: 'http://localhost:8070/api',
  CAMUNDA_TENANT_ID: '', // We can override values in the env by passing an empty string value
  CAMUNDA_SECURE_CONNECTION: false,
})

卡蒙达·萨阿斯

以下是连接到 Camunda SaaS 的完整配置示例 :

export ZEEBE_GRPC_ADDRESS='grpcs://5c34c0a7-7f29-4424-8414-125615f7a9b9.syd-1.zeebe.camunda.io:443'
export ZEEBE_REST_ADDRESS='https://syd-1.zeebe.camunda.io/5c34c0a7-7f29-4424-8414-125615f7a9b9'
export ZEEBE_CLIENT_ID='yvvURO9TmBnP3zx4Xd8Ho6apgeiZTjn6'
export ZEEBE_CLIENT_SECRET='iJJu-SHgUtuJTTAMnMLdcb8WGF8s2mHfXhXutEwe8eSbLXn98vUpoxtuLk5uG0en'

export CAMUNDA_TASKLIST_BASE_URL='https://syd-1.tasklist.camunda.io/5c34c0a7-7f29-4424-8414-125615f7a9b9'
export CAMUNDA_OPTIMIZE_BASE_URL='https://syd-1.optimize.camunda.io/5c34c0a7-7f29-4424-8414-125615f7a9b9'
export CAMUNDA_OPERATE_BASE_URL='https://syd-1.operate.camunda.io/5c34c0a7-7f29-4424-8414-125615f7a9b9'
export CAMUNDA_OAUTH_URL='https://login.cloud.camunda.io/oauth/token'
export CAMUNDA_AUTH_STRATEGY='OAUTH'

# Admin Console and Modeler API Client
export CAMUNDA_CONSOLE_CLIENT_ID='e-JdgKfJy9hHSXzi'
export CAMUNDA_CONSOLE_CLIENT_SECRET='DT8Pe-ANC6e3Je_ptLyzZvBNS0aFwaIV'
export CAMUNDA_CONSOLE_BASE_URL='https://api.cloud.camunda.io'
export CAMUNDA_CONSOLE_OAUTH_AUDIENCE='api.cloud.camunda.io'

日志

SDK使用温斯顿/皮诺兼容的日志设置. 默认它使用温斯顿.

在使用默认的日志库时,您可以通过环境变量(或构造器配置属性)设置 SDK 的日志级别 CAMUNDA_LOG_LEVEL。此默认为“ info” 。 数值(按优先顺序排列): error, (中文(简体) ), warn, (中文(简体) ), info, (中文(简体) ), http, (中文(简体) ), verbose, (中文(简体) ), debug, (中文(简体) ), silly。 。 。 .

自定义日志

您可以通过构建器提供自定义日志 。 例如,使用 皮诺 日志库 :

import pino from 'pino'

import { Camunda8 } from '@camunda8/sdk'

const level = process.env.CAMUNDA_LOG_LEVEL ?? 'trace'
const logger = pino({ level }) // Logging level controlled via the logging library

logger.info('Pino logger created')
const c8 = new Camunda8({
  logger,
})
c8.log.info('Using pino logger')

等待同步查询数据

Camunda 8采用了最终一致的数据架构. 当您启动进程实例时, 数据存储中无法立即获取与此进程实例相关的数据 。 这会导致您在应用程序中需要管理的数据同步问题. 为了帮助你,SDK提供了一个功能: PollingOperation您可以通过一个查询 API 操作到此工具, 并带有投票间隔和超时.

这个 PollingOperation 将反复执行查询,直到数据存储中可以获得预期数据,或者达到超时.

下列实例将在元素实例数据获得后立即返回新建进程实例的查询回复:

import { Camunda8, PollingOperation } from '@camunda8/sdk'

const c8 = new Camunda8()

const elementInstances = await PollingOperation({
  operation: () =>
  c8.searchElementInstances({
    sort: [{ field: 'processInstanceKey' }],
    filter: {
      processInstanceKey: processInstance.processInstanceKey,
      type: 'SERVICE_TASK',
    },
  }),
  interval: 500,
  timeout: 10000,
})

默认情况下 PollingOperation 等待 Orchestration 集群 API 的查询响应,该响应在 items 阵列。 如果有一个更具体的上游函数,或者正在使用其中一款v1组件API,则可以在自定义的上游函数中通过.

以下示例等待操作 API 上的查询可获取进程实例 :

import { Camunda8, PollingOperation } from '@camunda8/sdk'

const c8 = new Camunda8()
const c = c8.getOperateApiClient()

const process = await PollingOperation({
  operation: () => c.getProcessInstance(p.processInstanceKey),
  predicate: (res) => res.key === p.processInstanceKey,
  interval: 500,
  timeout: 15000,
})

订阅查询

您可以使用 QuerySubscription。这是一个事件发射器 data 当有新数据时。 您通过一个上游函数,它需要 previous 财务报告和审定财务报表 current 查询结果设置 。 在此上游函数中,您可以检查两个状态,看看是否释放出数据事件,并进行数据处理——例如删除上次更新中排放的项目.

请注意,这是一个实验性特征,在未来的释放中可能会有变化. 请在GitHub Repo中报告问题.

以下是使用 QuerySubscription数字 :

import { Camunda8, QuerySubscription } from '@camunda8/sdk'

const c8 = new Camunda8()

const query = () =>
 c8.searchProcessInstances({
  filter: {
    processDefinitionKey: key,
    state: 'ACTIVE',
  },
  sort: [{ field: 'startDate', order: 'ASC' }],
})

const subscription = QuerySubscription({
  query,
  predicate: (previous, current) => {
    // This is the default predicate, provided here as an example
    const previousItems = (previous?.items ?? []) as Array<unknown>
    const currentItems = current.items.filter(
        (item) =>
            !previousItems.some((prevItem) => isDeepStrictEqual(prevItem, item))
    )
    if (currentItems.length > 0) {
        return {
            ...current,
            items: currentItems,
            page: { ...current.page, totalItems: currentItems.length },
        }
    }
    return false // No new items, do not emit
  },
  interval: 5000,
  trackingWindow: 5, // Remember emitted items from last 5 poll cycles (default)
})

subscription.on('data', data => {
    // new process instances
})
//...
subscription.cancel() // close subscription and free resources
// You can also use subscription.pause() and subscription.resume() to pause and resume the subscription

说明数字 :

• QuerySubscription 使用滚动窗口方法来防止在跟踪被排放的项目时发生内存泄漏. 默认情况下,它会记住过去5个民意测验周期的项目来防止重复. 你可以用 trackingWindow 参数 :

const subscription = QuerySubscription({
  query,
  predicate: myCustomPredicate,
  interval: 5000,
  trackingWindow: 10, // Remember items from the last 10 polling cycles
})

如果实施自定义上游,则上游确定应排放哪些数据,然后滚动窗口机制防止多个民意测验周期的重复排放.

重要内容: 滚动窗口机制的运作方式是将每个被排放的物品序列化到一个字符串,并跟踪其特定的投票周期。 这意味着,如果一个实体改变状态,然后在跟踪窗口时间范围内返回到以前的状态,那么在原始状态中的第二次出现可能不会触发排放. 例如,如果一个进程实例从"READY"向"NOT READY"过渡并回到跟踪窗口内的"READY",那么第二个"READY"状态可能不会被释放出. 如果您需要跟踪所有州的过渡, 不论之前的州, 请考虑:

1. 设置一个较小 trackingWindow 用于缩短跟踪期的值
2. 在您的应用程序中执行自定义状态跟踪机制
3. 在项目识别逻辑中包含时间戳或版本字段

调试

SDK会使用 debug 程序库来帮助您调试特定的问题。 这会产生动词,从特定组件到控制台的低等输出.

要启用调试输出,请为 DEBUG 环境变量。 该值是一个以逗号分隔的调试命名空间列表. SDK有以下命名空间:

以下是 OAuth 和 Operate 组件启动调试的例子:

DEBUG=camunda:oauth,camunda:operate node app.js

这样做是为了进行开发调试,生产时不应允许。 调试追踪可以包括机密等敏感信息.

诊断 跟踪文件

您可以通过设置环境变量输出一个诊断痕量文件, 与 Camunda 技术支持使用 CAMUNDA_SUPPORT_LOG_ENABLED 真实的。 这将输出一个文件 camunda-support.log 载有诊断信息和追踪电话。 此信息可用于调试 。 注意这个日志文件将包含无记名符号等秘密,以及曝光urls.

进程变量打字

进程变量 - e variables zeebe 信件、 工作、 进程实例创建请求和响应 - 存储在经纪人中, 作为密钥: 值对 。 它们是作为JSON绳来运输的. SDK将JSON字符串分解为JavaScript对象.

各种Zeebe方法接受DTO类用于可变输入和输出. 这些DTO类用于提供设计时间类型信息 variables 对象。 它们也被用来安全地解码无法被 JavaScript 准确表示的64位整数 number 类型.

创建 DTO 以代表预期形状和类型 variables 对象,扩展 LosslessDto 类 :

class myVariableDTO extends LosslessDto {
    firstName!: string
    lastName!: string
    age!: number
    optionalValue?: number
    @Int64String
    veryBigInteger?: string
    constructor(data: Partial<myVariableDTO>) {
        super(data)
    }
}

Zeebe 工人变量的打字

Zeebe 工人任务处理器中的可变有效载荷可作为物体使用 job.variables。默认情况下,这是类型 any 用于GRPC API,以及 unknown 倒计时.

这个 ZBClient.createWorker() 方法接受一个 inputVariableDto 来控制数字值的解析,并提供设计时的类型信息。 通过一个 inputVariableDto 给Zeebe工人的课是可选的 如果一个DTO类被传递给Zeebe工人,则用于两个目的:

• 提供设计时间类型信息 job.variables 对象.
• 指定 JSON 数字字段的解析 。 这些可能代表 int64 无法用 JavaScript 准确表示值 number 类型。 使用 DTO ,您可以指定一个特定的 JSON 数字字段会被无损失地解析到一个 string 或者说 BigInt。 。 。 .

没有指定 DTO 时, 没有设计时的安全性 。 在运行时, JSON 数字全部转换为 JavaScript number 类型。 如果一个可变字段的数值不能使用 JavaScript 数字类型(一个大于 2^53 -1 的值) 安全表示,则丢出一个例外 .

要提供DTO,请扩展 LosslessDto 这样的课:

class MyVariableDto extends LosslessDto {
    name!: string
    maybeAge?: number
    @Int64String
    veryBigNumber!: string
    @BigIntValue
    veryBigInteger!: bigint
}

在这种情况下, veryBigNumber 是一个 int64 数值。 它作为JSON的号码在电线上转移, 但解析器会把它分析成一个 string 以免丧失精度。 同样, veryBigInteger 是一个非常大的整数值。 在这种情况下,我们指示解析器将这个可变字段分析为 bigint。 。 。 .

你可以像这样去DTO:

class MyLargerDto extends LosslessDto {
    id!: string
    @ChildDto(MyVariableDto)
    entry!: MyVariableDto
}

自定义信头的打字

Zeebe工人收到定制头条: job.customHeaders编辑 ZBClient.createWorker() 方法接受一个 customHeadersDto 控制自定义标题分析数字值的行为,并提供设计-时间类型信息.

如前所述,这遵循与职务变量相同的策略.

Zeebe 用户任务

从8.5可以使用Zeebe用户任务. 见以下文件: 如何迁移到 Zeebe 用户任务。 。 。 .

SDK支持"Zeebe REST API". 确定设置 ZEEBE_REST_ADDRESS 通过环境变量或配置字段.

任务流

Zeebe gRPC API支持流出现有工作,而不是为他们投票.

ZeebeGrpc 快捷法 streamJobs 允许您使用此 API.

何时 streamJobs ,它首先打开一条溪来接收任何新建的工作,然后进行初步民调(通过 activateJobs)来取出任何在溪流建立前已经排队的工作. 这种"订阅再回填"的方法确保不漏出工作——溪流在开通后捕获所创造的任何东西,民意调查捕获之前存在的任何东西,而经纪人的单活性保证可以防止重复.

旁车投票

在最初的回填后,低频侧车民意调查与溪道并列作为安全网. 仅靠这条溪流无法处理的边缘情况——例如,在工人不能在超时完成后重新排队的工作,或在一条死溪被重新打开时到达的工作(见 连续故障和回收) (中文(简体) ).

每次民意测验都是对经纪人的指令,因此默认间隔是故意长(30秒)以尽量减少服务器负载. 溪流仍然是主要的提供机制.

配置侧车民调 :

• pollInterval ——侧车票周期相隔数毫秒. 默认为 30000 (30 s) 设定为 0 或者说 -1 完全禁用重复投票(最初的回填投票仍在进行).
• pollMaxJobsToActivate ——每个民意调查周期(初始和重复)的最大工作. 默认为 32。 。 。 .

连续故障和回收

工作流可以在工人过程持续运行时死亡:连接下降,网关重新启动或拒绝流,或服务器结束. 没有康复,一个工人的溪流已经死亡,只能得到低频侧车民意调查找到的工作.

streamJobs 检测已死流( a) error, (中文(简体) ), end 或者说 close 事件,或试图打开它失败),默认情况下,以指数回放(1 s,双倍为30 s)重新打开. 一旦溪流重新建立,它将进行一次回填民意调查,以收集溪流下行时创造的工作。 一旦重新打开的流提供工作或至少停留最长时间的延迟,备份即重新开始。 通话 close() 在返回的手柄(或客户端)上停止所有重联尝试.

以 :

• onStreamError(error) ——当溪流出错或无法打开时调用.
• onStreamEnd() ——当服务器结束或关闭流没有出错时调用.
• onStreamReconnect(attempt) ——每次尝试重开溪流前都打电话. attempt 是连续失败的次数,从 1 开始.
• streamReconnect —— 说 true (默认), false 以禁用再连接(上面的回调仍然起火),或 { initialDelayMs, maxDelayMs } 以调和后退.

回电可能就是 async被这些召回和拒绝任何承诺的例外情况被抓住并记录下来;它们不影响重新连接。 如果第一次尝试时无法打开溪流 streamReconnect 实值 false承诺的回报 streamJobs 以基本错误拒绝.

streamJobs 返回一个承诺,所以 await 用它来获取手柄 close() 方法.

你可以选择通过一个 jitter 参数(以毫秒计)可以将起始时间随机延迟至该值。 这在同时启动多条流工时是有用的,以避免在应用程序启动时饱和网关.

示例

const streamWorker = await client.streamJobs({
    type: 'payment-service',
    worker: 'my-worker',
    timeout: 30000,
    tenantIds: ['<default>'],
    taskHandler: async (job) => {
        // business logic
        return job.complete()
    },
    // Optional: stagger startup across workers
    jitter: 2000,
    // Optional: control poll batch size (default: 32)
    pollMaxJobsToActivate: 64,
    // Optional: sidecar poll every 60 s instead of the default 30 s
    pollInterval: 60_000,
    // Optional: be told when the job stream dies and when it is re-opened
    onStreamError: (error) => console.error('Job stream failed', error),
    onStreamReconnect: (attempt) => console.log(`Re-opening stream (${attempt})`),
    // Optional: tune the reconnect backoff (default: 1 s doubling to 30 s)
    streamReconnect: { initialDelayMs: 1000, maxDelayMs: 30_000 },
})

// Stop the worker, including any pending reconnect
streamWorker.close()

多租户工人

工人,无论是投票还是流动,都可以是多租户,要求一个以上的租户提供工作.

示例

client.createWorker({
    taskHandler: (job) => {
        console.log(job.tenantId) // '<default>' | 'green'
        return job.complete()
    },
    taskType: 'multi-tenant-work',
    tenantIds: ['<default>', 'green'],
})

client.streamJobs({
    taskHandler: async (job) => {
        console.log(job.tenantId) // '<default>' | 'green'
        return job.complete()
    },
    type: 'multi-tenant-stream-work',
    tenantIds: ['<default>', 'green'],
    worker: 'stream-worker',
    timeout: 2000,
})

投票工人在错误条件下退站

当一名投票员遇到错误,包括没有得到认证时,该名投票员将因每次失败而以+2秒的时速后退,最多不超过 CAMUNDA_JOB_WORKER_MAX_BACKOFF_MS,默认为15000(15秒). 如果失败是由于无效的资质,并在象征性请求中发生,那么工人的后退会与符号端点后退相加,每次后退最多可达15秒为+1000ms.

这意味着,如果开始一个工人的证书无效,那么投票回放会是这样,默认情况下(以秒计):3,6,9,12,15,18,21,23,24,25,26,27,28,29,30,30,30...

如果工人退场的原因不是无效证书——例如来自网关的回压信号——将是:2,4,6,8,10,12,14,15,15,15.