跳到主要内容

Flow 基础

Flow 是你向 Dex 注册的顶层组件。

它定义在后台运行的 Step、持久化状态的 Attribute 与 Channel，以及提供 durable 通信的 RPC。

实现一个 Flow​

你通过实现 Flow interface 定义一个 Flow。

大多数 SDK 会从 struct 或 class 名推导 FlowType，除非你显式覆盖。

一个 Flow 实现主要包含：

  1. Steps — start Step 与 Flow 运行的其它 Step。
  2. Persistence schema — Flow 读写的 Attribute 与 Channel。
  3. RPC handlers — Flow 可能调用的方法。
status = Attribute("status", str)
notify = Channel("notify", None)


class ExampleFlow(Flow[int]):
    def __init__(self) -> None:
        self.finish = Finish()
        self.example = Example(self.finish)

    def get_steps(self) -> StepList[int]:
        return StepList.start_step(self.example).other_steps(self.finish)

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(status, notify)

    @rpc
    def describe(self, context: Context) -> RPCResult[str]:
        return RPCResult.of(status.get(context))

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

标识符​

Dex 在其 API 和文档中使用这些标识符。

首先，说到 Flow 时，它通常依上下文指由 FlowType 标识的 Flow 定义，或由 Flow ID 标识的一次 Flow execution。

标识符谁设置含义
FlowType应用（Flow 实现）Dex 路由 Worker 调用与搜索结果使用的稳定类型名。
Flow ID应用在 start 时一个 Flow execution 的业务身份。ID reuse policy 控制是否可以使用同一个 Flow ID 启动另一个 Flow execution。
Run IDDex Server这个 Flow ID 的一次 run 的身份。Time travel 保留它，并创建一个新的 run attempt。请用 Flow ID 定位 Flow；Run ID 只用于诊断。
StepType应用（Step 实现）Flow 内 Step 定义的 stable 名称。
StepExecutionIDDex Server某个 Step type 的一次执行实例，格式为 StepType-Number。

Steps​

GetSteps / get_steps / steps 注册一个 start Step，以及通过 StepDecision movement 可能到达的其它 Step type。start Step 的 input type 也是 Flow 的 start input type。

Step 列表为空，或没有 start Step，在 Flow 完全由 RPC 驱动、稍后再进入 Step 时是合法的。

WaitFor、Execute 与默认 Step options 见 Step 基础。

Persistence schema​

Persistence schema 列出 Step 或 RPC 使用的每个 Attribute 与 Channel。未声明的名称会被 Dex 拒绝读写。

RPC handlers​

每个 Flow 都显式声明 RPC handler 及其 execution options。Go 通过 GetRPCs 和 DefineRPC 返回 direct bound method。Dex 将 method name 作为 durable RPC name，Rust 也可以提供显式名称。Client 使用 InvokeRPC 调用；Dex 随后将调用派发到 Worker。