Advanced: Long-lived Parent
Keep a bounded pool of parent Steps alive while the Flow accepts more requests.
Init publishes the initial requests to RequestChannel, sets Stopped to false, and starts N HandleRequest executions. Each worker takes one FIFO message and moves to HandleSubFlow. That Step starts and waits for ExampleSubFlow. When the child completes, it either loops to HandleRequest or gracefully completes after Stop has set Stopped to true.
SendRequest checks the Channel size before publishing. It returns false when the buffer threshold is reached, so callers can apply back pressure instead of overloading the parent.
Core implementation
DEFAULT_CONCURRENCY = 10
MAX_BUFFERED_REQUESTS = 100
class LongLiveInit(Step[ParentInput]):
def __init__(
self,
request_channel: Channel[str],
stopped: Attribute[bool],
) -> None:
self.request_channel = request_channel
self.stopped = stopped
def get_step_type(self) -> str:
return "Init"
def execute(self, context: Context, input: ParentInput) -> StepDecision:
for request in input.requests:
self.request_channel.publish(context, request)
self.stopped.set(context, False)
concurrency = input.concurrency if input.concurrency > 0 else DEFAULT_CONCURRENCY
return go_to_many(
*(StepMovement.of(LongLiveHandleRequest, None) for _ in range(concurrency))
)
class LongLiveHandleRequest(Step[None]):
def __init__(self, request_channel: Channel[str]) -> None:
self.request_channel = request_channel
def get_step_type(self) -> str:
return "HandleRequest"
def wait_for(self, context: Context, input: None) -> Wait:
return Wait.until(self.request_channel.for_one())
def execute(self, context: Context, input: None) -> StepDecision:
return go_to(LongLiveHandleSubFlow, self.request_channel.results(context)[0])
class LongLiveHandleSubFlow(Step[str]):
def __init__(
self,
example_subflow: ExampleSubFlow,
stopped: Attribute[bool],
) -> None:
self.example_subflow = example_subflow
self.stopped = stopped
def get_step_type(self) -> str:
return "HandleSubFlow"
def wait_for(self, context: Context, request: str) -> Wait:
return Wait.until(SubFlow.run(self.example_subflow, request))
def execute(self, context: Context, request: str) -> StepDecision:
if self.stopped.get(context):
return graceful_complete()
return go_to(LongLiveHandleRequest, None)
class AdvancedLongLiveParentFlow(Flow[ParentInput]):
request_channel = Channel("RequestChannel", str)
stopped = Attribute("Stopped", bool)
def __init__(self, example_subflow: ExampleSubFlow) -> None:
self.init = LongLiveInit(self.request_channel, self.stopped)
self.handle_request = LongLiveHandleRequest(self.request_channel)
self.handle_subflow = LongLiveHandleSubFlow(example_subflow, self.stopped)
def get_steps(self) -> StepList[ParentInput]:
return StepList.start_step(self.init).other_steps(
self.handle_request, self.handle_subflow
)
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.of(self.request_channel, self.stopped)
@rpc
def send_request(self, context: Context, request: str) -> RPCResult[bool]:
if self.request_channel.size(context) >= MAX_BUFFERED_REQUESTS:
return RPCResult(False)
self.request_channel.publish(context, request)
return RPCResult(True)
@rpc
def stop(self, context: Context) -> None:
self.stopped.set(context, True)
Example: examples/python/dex_examples/patterns/parallel-subflows/advanced_long_live_parent_flow.py