Drain External Channels
Process externally published Channel messages, then close the short-lived Flow when its queue is empty.
Problem and approach
Use this pattern when a Flow handles messages from outside the Worker with potential bursts. A Client invokes an RPC that appends each message to the Channel. The key requirement is to complete the Flow when the Channel is empty, so the Flow stays as short as possible. The Flow restarts when new messages arrive.
The decision atomically checks the Channel while it records the next action. An empty Channel completes the Flow. A Channel with queued messages, or one with messages committed to a Step that has not finished processing Execute, takes the fallback movement back to ProcessMessage. After completion, the next request starts a new Flow execution with the same Flow ID and passes the message as the start input.
Only one Step may consume a Channel checked by ForceCompleteIfChannelsEmpty.
Step graph
Core implementation
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)
Example: examples/python/dex_examples/patterns/drain-channels/external_publishing/draining_channel_flow.py
Controller
The controller first invokes the publishing RPC on the active Flow. When invocation reports that the Flow is inactive, it starts a new Flow with the message as its input. Other errors are returned to the caller.
@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"
Example: examples/python/dex_examples/patterns/drain-channels/external_publishing/controller.py