Step decisions
View as MarkdownExecute returns a StepDecision. It moves to next Steps, closes the Flow, or both stops this branch and cancels other Steps.
Multiple next Steps
GoTo starts one next Step. GoToMany starts multiple next Steps. Each Step movement has its own input and can also have its own Step options.
impl Step for Route {
type Input = String;
fn execute(&self, _context: &mut Context, mode: Self::Input) -> HandlerResult<StepDecision> {
match mode.as_str() {
"graceful" => Ok(StepDecision::graceful_complete(String::from("done"))),
"dead-end" => Ok(StepDecision::go_to_many([
StepMovement::to(&BranchWorker, "left".to_string()),
StepMovement::to(&BranchWorker, "right".to_string()),
])),
_ => {
let quote = Quote {
carrier: "winner".to_string(),
price: 9,
};
Ok(StepDecision::go_to_many([
StepMovement::to(
&CarrierA,
Quote {
carrier: "A".to_string(),
price: 10,
},
),
StepMovement::to(
&CarrierB,
Quote {
carrier: "B".to_string(),
price: 12,
},
),
StepMovement::to(&Winner, quote),
]))
}
}
}
}
Example: examples/rust/src/primitives/step_decision/flow.rs
Close decisions
| Decision | What it does |
|---|---|
| GracefulComplete | Stops this branch and triggers the Flow to prepare for completion. When all branches and Steps are stopped, the Flow completes. |
| DeadEnd | Stops this branch only. |
| ForceComplete | Completes the Flow immediately. |
| ForceFail | Fails the Flow immediately. |
GracefulComplete and ForceComplete can return an output. ForceFail returns a failure reason. The Flow result returns each output with the Step type and Step execution ID. When multiple Steps complete gracefully, all their outputs return together.
A Flow can have no active Steps and still keep running. DeadEnd stops this branch without making a decision that triggers the Flow to complete.
impl Step for Route {
type Input = String;
fn execute(&self, _context: &mut Context, mode: Self::Input) -> HandlerResult<StepDecision> {
match mode.as_str() {
"graceful" => Ok(StepDecision::graceful_complete(String::from("done"))),
"dead-end" => Ok(StepDecision::go_to_many([
StepMovement::to(&BranchWorker, "left".to_string()),
StepMovement::to(&BranchWorker, "right".to_string()),
])),
_ => {
let quote = Quote {
carrier: "winner".to_string(),
price: 9,
};
Ok(StepDecision::go_to_many([
StepMovement::to(
&CarrierA,
Quote {
carrier: "A".to_string(),
price: 10,
},
),
StepMovement::to(
&CarrierB,
Quote {
carrier: "B".to_string(),
price: 12,
},
),
StepMovement::to(&Winner, quote),
]))
}
}
}
}
impl Step for BranchWorker {
type Input = String;
fn execute(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult<StepDecision> {
Ok(StepDecision::dead_end())
}
}
Example: examples/rust/src/primitives/step_decision/flow.rs
impl FlowGracefulTimeout {
pub(crate) fn timeout_handler_options(&self) -> FlowTimeoutHandlerOptions {
FlowTimeoutHandlerOptions::new()
.method_timeout(Duration::from_secs(30))
.retry(RetryPolicy::new().maximum_attempts(3))
}
fn handle_timeout(&self, _context: &mut Context) -> HandlerResult<StepDecision> {
Ok(StepDecision::force_fail("task exceeded graceful timeout"))
}
}
impl Step for LongWait {
type Input = bool;
fn wait_for(&self, _context: &mut Context, successful: bool) -> HandlerResult<Wait> {
let duration = if successful { 1 } else { 65 };
Ok(Wait::until(Timer::by_duration(Duration::from_secs(
duration,
))))
}
fn execute(&self, _context: &mut Context, _successful: bool) -> HandlerResult<StepDecision> {
Ok(StepDecision::force_complete("task-completed".to_string()))
}
}
Example: examples/rust/src/patterns/timeout/flow.rs
Cancel other Steps
A StepDecision can cancel other active Steps.
- CancelSteps selects active Steps by Step type.
- CancelSiblingSteps selects active Steps that were triggered by the same parent Step execution.
An RPC can also cancel Steps by type. See Cancel running Steps. For how a running attempt observes cancellation, see Cancellation.
Ok(StepDecision::go_to(&RecordQuote, quote)
.cancel_step(&CarrierA)
.cancel_step(&CarrierB))
Example: examples/rust/src/primitives/step_decision/flow.rs
ForceCompleteIfChannelsEmpty
ForceCompleteIfChannelsEmpty atomically checks whether all named Channels are empty. A Channel with queued messages is non-empty. After a message is consumed and committed to a Step, that Channel remains non-empty until the Step finishes processing Execute. If every named Channel is empty, it completes the Flow. Otherwise, it goes to other Steps.
Use it to drain Channels and complete the Flow, so the Flow can stay short. See Drain External Channel for the full example.
impl Step for DrainChannel {
type Input = String;
fn wait_for(&self, _context: &mut Context, input: String) -> HandlerResult<Wait> {
if input.is_empty() {
return Ok(Wait::until(EXTERNAL_QUEUE.for_one()));
}
Ok(Wait::skip_immediately())
}
fn execute(&self, context: &mut Context, input: String) -> HandlerResult<StepDecision> {
let item = if input.is_empty() {
EXTERNAL_QUEUE
.condition_results(context)?
.into_iter()
.next()
.unwrap_or_default()
} else {
input
};
context.record_event("drained-channel", item)?;
Ok(StepDecision::force_complete_if_channels_empty(
String::new(),
StepMovement::to(&DrainChannel, String::new()),
[EXTERNAL_QUEUE.when_empty()],
))
}
}