Drain Internal Channels
Finish a producer/consumer Flow after its internally published Channel messages have been processed.
Problem and approach
Use this pattern when one branch of a Flow publishes work for another branch. Init starts MainStep and SideStep in parallel. MainStep publishes each document command to SideStepData. SideStep waits for one Channel value, writes it, and waits again.
The key is the sentinel published by Finalize. After MainStep has published every ordinary message, Finalize publishes a final command to SideStepData. Because a Channel is a FIFO queue, SideStep cannot receive that sentinel until it has received every earlier message. It completes only when it receives the sentinel, so the Flow does not finish before every published document is handled. With several producer branches, make Finalize wait for all producers before it publishes the sentinel.
Code examples
class MainStep(Step[str]):
def execute(self, context: Context, input: str) -> StepDecision:
execution_count = self.execution_counter.get(context) + 1
self.execution_counter.set(context, execution_count)
statuses = {1: "RECEIVED", 2: "ACCEPTED", 3: "PASSED"}
self.side_step_data.publish(
context,
MongoDocument(input, statuses.get(execution_count, "ERROR"), False),
)
if execution_count <= 3:
return go_to(MainStep, input)
return go_to(Finalize, None)
class SideStep(Step[None]):
def wait_for(self, context: Context, input: None) -> Wait:
return Wait.until(self.side_step_data.for_one())
def execute(self, context: Context, input: None) -> StepDecision:
document = self.side_step_data.results(context)[0]
self.mongo_collection.upsert(document)
if document.final_command:
return graceful_complete()
return go_to(SideStep, None)
class Finalize(Step[None]):
def execute(self, context: Context, input: None) -> StepDecision:
self.side_step_data.publish(
context, MongoDocument("documentId-1", "FINALIZED", True)
)
return graceful_complete()
例子: examples/python/dex_examples/patterns/drain-channels/internal/drain_internal_channels_flow.py