Client Basics

Client 调用来自 application，不来自 Step。下面的例子使用 order-123 这个 Flow ID。

启动 Flow

StartFlow 创建 Flow execution，并在 Dex Server 接受请求后立即返回。它接收已注册的 Flow、业务 Flow ID、start Step input 和 StartFlowOptions。

Flow ID 是 application 选择的身份。Dex Server 创建 Run ID。关于 timeout、request idempotency、ID reuse、retry、initial Attributes 和 config override，见 Flow options。

run_id = await app_state.client.start_flow(
    app_state.client_apis, "order-123", "first order", start_options()
)

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

停止 Flow

StopFlow 关闭一个 active Flow。默认是请求它 cooperative cancellation。你也可以立刻 terminate，或者带 reason 把它标记为 failed。

await client.stop_flow("order-123", StopFlowOptions(reason="customer cancelled"))

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

调用 RPC

InvokeRPC 调用已注册的 RPC handler，并返回结果。只走 query 的 RPC 可以读取仍在 retention 内的 terminal execution。需要 commit effect 的 RPC 则要求 execution 仍处于 active 状态。

result = await client.invoke_rpc(flow.trigger, "order-123", "approved")

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

读写 Flow 状态

修改 Attribute、AttributeMap、Channel 和 ChannelMap 状态时，应用应以 Flow RPC 作为边界。为每个 command 定义 typed RPC，只选择它需要的状态，再用 InvokeRPC 调用。这样 authorization、validation、lock、transaction 和状态变更都会留在 Flow 内。

读取已存储的 Attribute 值时，调用 GetAttributes。它直接从 Dex Server 读取已提交的状态。如果读取需要 Flow 内的 validation 或 authorization、Channel 消息、派生视图或 Attribute lock，改用 RPC。已有的只读 RPC 继续可用，应用可以按自己的节奏迁移到 GetAttributes。

Client 仍提供 WaitForAttributeMatch 作为阻塞式 observation API。当应用需要等待已知的 scalar Attribute 条件，而不是读取任意状态快照时，使用这个接口。

读取 Attribute

GetAttributes 从 Flow 外读取已注册的 Attribute 和 AttributeMap instance 的值。每个值传一个 read：一个 singleton Attribute，或一个 AttributeMap instance。这个 API 只有批量形式。只读一个值时，传一个 read。

它在 Go 中叫 GetAttributes，在 Python 和 Rust 中叫 get_attributes，在 Java 和 TypeScript 中叫 getAttributes。

Rust 调用会阻塞调用线程，与其他 Rust Client 方法一样。它接受由 1 到 12 个 read 组成的 tuple，并返回同样大小的 tuple。要读取更多值，请分多次调用。不同调用读到的值可能来自不同的提交。

Attribute primitive 示例读取一个 status Attribute、progress AttributeMap 的 payment instance，以及一个 Flow 从不设置的 email Attribute。

flow = app_state.attribute
payment_progress = flow.progress.instance("payment")
try:
    values = await app_state.client.get_attributes(
        required_query("workflowId"),
        flow.status,
        payment_progress,
        flow.email,
    )
except FlowNotFoundError as error:
    response = jsonify({"error": str(error)})
    response.status_code = 404
    return response
return jsonify(
    {
        "status": values.get(flow.status, "not started"),
        "paymentProgress": values.get(payment_progress, "not started"),
        # This Flow never sets email, so get returns None.
        "email": values.get(flow.email),
    }
)

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

可以在 examples playground 中运行这个 HTTP 示例。每个示例服务器都提供 Read Attributes endpoint。

读取有以下语义：

• 一次提交。 Client 在一个服务器请求中发送所有 read。所有值都来自同一个已提交状态。
• 当前或最新 run。 调用读取当前 run；没有打开的 run 时，读取最新 run。已关闭的 run 在 Dex 清除它之前仍可读取。
• 不经过 Worker。 Dex Server 自己响应读取。它不向 Worker 派发任何工作，不占用 Worker 容量，没有 Worker 运行时也能读取。
• 只读已提交状态。 调用看不到正在运行的 Step 或 RPC 已写入但尚未提交的值。相比之下，RPC handler 还能看到它在同一次调用中先前写入的值。GetAttributes 不获取 Attribute lock，因此 read-modify-write 应使用 RPC。
• 大值和 hard-delete 值。 Dex Server 在响应前解析 hard-delete 值，已删除的值读作未设置。Client 从 blob 存储加载其他大值，再用定义解码每个值。只有所有值都解码成功后，调用才返回。

没有存储值的 Attribute 或 AttributeMap instance 是未设置的。各 SDK 用自己的方式表示未设置的 read：

SDK	未设置的 read
Go	destination 保留原值。AttributeReadResult.IsSet 和 IsMapInstanceSet 返回 false。
Python	返回的 AttributeValues 中没有这个 read。values.get(read, default) 返回 default，read in values 为 false，values[read] 抛出 KeyError。
Java	AttributeValues.get(read) 返回 Optional.empty()。
TypeScript	tuple 中对应的元素是 undefined。
Rust	tuple 中对应的元素是 None。每个元素都是 read 的 value type 的 Option。

在 Python 和 TypeScript 中，解码为 None 或 null 的存储值仍算已设置。在 Java 中，解码为 null 的存储值读作 Optional.empty()，与未设置的值相同，因为 Optional 不能保存 null。

每个 read 使用的定义都必须由 Client Registry 中的某个 Flow 声明。任何已注册 Flow 的定义都可以使用。Dex 按名称读取 key，不检查目标 Flow 的类型是否声明了它，所以同名但 value type 不同的 Attribute 会解码失败。

调用会报告以下错误：

• Flow ID 不存在，或其最新 run 正在被清除时，返回 Flow-not-found 错误。Go、Python 和 TypeScript 中是 FlowNotFoundError，Java 中是 FlowNotFoundException，Rust 中是 SdkError::FlowNotFound。
• 定义没有在 Client Registry 中注册，或与已注册的定义不匹配时，在发出请求前失败。Go、Python 和 TypeScript 中的错误是 FlowDefinitionError，Java 中是 FlowDefinitionException，Rust 中是 SdkError::FlowDefinition。Go、Python 和 Rust 会把定义与已注册的定义比较，包括 value type。Java 和 TypeScript 要求传入注册时的同一个定义对象。
• Flow ID 为空或只含空白、没有 read，或两个 read 读取同一个值时，在发出请求前被拒绝。AttributeMap instance key 为空或包含 / 时，同样在发出请求前被拒绝。Java 抛出 IllegalArgumentException，Rust 返回 SdkError::InvalidArgument。在 Rust 中，空的 read tuple 无法通过编译。
• 在 Python 和 TypeScript 中，传入没有调用 instance 的 AttributeMap 时，在发出请求前抛出 TypeError。在 Go 中，ReadAttribute 只接受 Attribute，ReadAttributeMapInstance 只接受 AttributeMap，所以编译器会拒绝这种错误。在 Java 和 Rust 中，AttributeMap 不是 read，所以编译器同样会拒绝它。
• 存储的值无法解码时，返回 value mapping 错误。Go、Python 和 TypeScript 中是 ValueMappingError，Java 中是 ValueMappingException，Rust 中是 SdkError::ValueMapping。