Drain External Channels
处理外部发布到 Channel 的消息；队列为空时完成这条短生命周期的 Flow。
问题与做法
当一条 Flow 要处理 Worker 外部发来的消息，并且消息可能突发到来时，使用这个模式。Client 调用 RPC，由 RPC 把每条消息追加到 Channel。关键要求是在 Channel 为空时完成 Flow，从而尽可能让 Flow 保持短生命周期。有新消息到来时，Flow 会重新启动。
这个 decision 会在记录下一步动作时原子地检查 Channel。Channel 为空时完成 Flow；Channel 中还有排队消息，或有消息已从该 Channel 消费并提交给尚未完成 Execute 处理的 Step 时，Dex 会走 fallback movement 回到 ProcessMessage。前一次 execution 完成后，下一个请求会使用相同的 Flow ID 启动新的 Flow execution，并将消息作为启动输入。
被 ForceCompleteIfChannelsEmpty 检查的 Channel 只能由一个 Step 消费。
Step Graph
核心实现
class ProcessMessage(Step[str]):
def __init__(self, queue_channel: Channel[str]) -> None:
self.queue_channel = queue_channel
def wait_for(self, context: Context, input: str) -> Wait:
if input is None:
return Wait.until(self.queue_channel.for_one())
return Wait.skip_immediately()
async def execute( # type: ignore[override]
self, context: AsyncContext, input: str
) -> StepDecision:
if input is not None:
print(f"DrainingExternalChannelFlow process message: {input}")
else:
values = self.queue_channel.results(context)
if not values:
raise RuntimeError("No channel message found")
value = values[0]
if value is None:
raise RuntimeError("No channel message value found")
print(f"DrainingExternalChannelFlow process message: {value}")
await asyncio.sleep(DRAIN_WINDOW_SECONDS)
return force_complete_if_channels_empty(
None,
StepMovement.of(self, None),
self.queue_channel,
)
class DrainingExternalChannelFlow(Flow[str]):
QUEUE_CHANNEL = "queueChannel"
queue_channel = Channel(QUEUE_CHANNEL, str)
def __init__(self) -> None:
self.process_message = ProcessMessage(self.queue_channel)
def get_steps(self) -> StepList[str]:
return StepList.start_step(self.process_message)
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.of(self.queue_channel)
@rpc
def publish_external_channel_message(
self, context: Context, input: str
) -> RPCResult[str]:
self.queue_channel.publish(context, input)
return RPCResult(input)
例子: examples/python/dex_examples/patterns/drain-channels/external_publishing/draining_channel_flow.py
Controller
Controller 会先调用活跃 Flow 的发布 RPC。调用表明 Flow 不活跃时，Controller 会使用该消息作为输入启动新的 Flow；其他错误会返回给调用方。
@blueprint.get("/start-or-publish")
async def start_or_publish_draining_channel() -> str:
flow_id = required_query("workflowId")
try:
await app_state.client.invoke_rpc(
app_state.drain_external.publish_external_channel_message,
flow_id,
"message from start-or-publish endpoint",
)
except FlowNotActiveError:
run_id = await app_state.client.start_flow(
app_state.drain_external,
flow_id,
"first message from start-or-publish",
start_options(),
)
return f"Started the workflow with runId {run_id}"
return "Published to the Flow"
例子: examples/python/dex_examples/patterns/drain-channels/external_publishing/controller.py