跳到主要内容

Client Advanced

Flow 关闭期间的请求​

Flow 关闭期间，Dex 会以 gRPC FAILED_PRECONDITION 和子状态 FLOW_CLOSING 拒绝新发起的事务性或带锁的 RPC、WaitForStepCompletion、WaitForAttributeMatch 和 Channel 消息删除。每种 SDK 都提供有类型的 Flow closing 错误。决定是否需要进一步操作前，应重新读取保留的 Flow 终态。该错误不能证明先前的请求已经成功。

等待 Flow​

WaitForFlow 等待 Flow 到达 terminal status。bounded wait 在 Flow 仍然 open 时会返回 long-poll timeout；再次调用即可继续等待。

result = await client.wait_for_flow("order-123", timedelta(minutes=1))

例子: examples/python/dex_examples/primitives/client-apis/controller.py

等待 Step​

WaitForStepCompletion 在 Flow 继续运行时等待一个 StepExecutionID。它只确认完成，不返回 Step output。

await client.wait_for_step_completion("order-123", StepExecutionId("ChargeOrder"), timedelta(minutes=1))

例子: examples/python/dex_examples/products/order-processing/controller.py

等待 Attribute 匹配​

WaitForAttributeMatch 等待 Attribute 满足类型化的 scalar 比较，并返回当前命中的值。它定位当前 active run。对于 AttributeMap，还需要传入 instance。

String 和 Boolean 支持 equal 与 not-equal。整数和浮点数还支持 greater-than、greater-than-or-equal、less-than 与 less-than-or-equal。缺失的 Attribute 永远不命中。

当调用方需要在状态变化后刷新时，使用应用拥有的 revision Attribute。等待 revision 大于上次观察到的值，再调用 Describe 或 read RPC。Revision 是 watermark；快速变化可能合并为一个返回值。

revision = await app_state.client.wait_for_attribute_match(
    flow_id,
    app_state.job_post.update_version,
    AttributeMatch.greater_than(required_int_query("lastRevision")),
    timedelta(seconds=30),
)

例子: examples/python/dex_examples/products/job-post/controller.py

object、bytes、null、blob-backed、非有限浮点、跨类型比较和无效 ordering 比较会被拒绝。完整 revision 模式见等待 Attribute 匹配。

查看 Flow​

DescribeFlow 读取 current 或 latest run 的 IDs、Flow type、status 和 start time。Go SDK 还会返回可用的 close time。它不会等待 Flow 结束。只读 RPC 返回非终态业务状态后，可以用它做一次状态检查。ID 不存在或不属于 Flow 时，会返回有类型的 Flow-not-found 错误。

info = await client.describe_flow("order-123")

例子: examples/python/dex_examples/primitives/client-apis/controller.py

搜索 Flow​

SearchFlows 返回匹配 visibility query 的一页 Flow runs，并排除服务器内部的 execution。search 用来寻找 Flow instance；DescribeFlow 用于查看一个已知 Flow ID。要继续搜索，传回 page token。

翻页时保持查询不变。

page = await client.search_flows("FlowType = 'ClientApisFlow'", 20, "")

例子: examples/python/dex_examples/primitives/client-apis/controller.py

跳过 Timer​

SkipTimer 让一个等待中的 Timer 立刻 ready。选择 Step execution，以及 Timer 的 condition ID 或它从零开始的 condition index。

await client.skip_timer("order-123", StepExecutionId("ChargeOrder"), TimerId.by_condition_id("retry"))

例子: examples/python/dex_examples/products/order-processing/controller.py

更新 FlowConfig​

UpdateFlowConfig 修改 active Flow 的 FlowConfig。只发送要修改的字段。SDK 的这个调用不会改变 Flow 的 Worker 路由。字段说明见 用 Client 更新 FlowConfig。

触发 continue-as-new​

TriggerContinueAsNew 已弃用。Dex 不会把 Flow continue as new。这个调用总是以 UNIMPLEMENTED 失败。不要调用它。长期运行的 Flow 不需要滚动，因为 Dex Server 只保留最近的 history 窗口。

Time travel​

TimeTravel 从已有 Flow history 的某个点创建一个新的 run attempt。run 保留原来的 Run ID。这个点必须位于最近的 history 窗口内，否则调用以 FAILED_PRECONDITION 失败。Time travel 到开头总是可用。

new_run_id = await client.time_travel("order-123", TimeTravelOptions(type=TimeTravelType.BEGINNING, reason="code fix"))

例子: examples/python/dex_examples/primitives/flow/controller.py