Timer

Timer is Durable Timer persisted on the Dex Server. While waiting on the Timer, there is no resource consumed at the Worker side.

class TimerExampleStep(Step[int]):
    def wait_for(self, _context: Context, input: int) -> Wait:
        return Wait.until(Timer.by_duration(timedelta(seconds=input)))

    def execute(self, context: Context, _input: int) -> StepDecision:
        print(context.has_timer_fired())
        return graceful_complete("timer-fired")

Example: examples/python/dex_examples/primitives/timer/timer_flow.py

How to skip a Timer

You can call SkipTimer with the Flow ID and Timer selector to skip a Timer. A skipped timer acts like timer has fired. This is helpful to manually trigger a timer for test simulation for production operation.

client.skip_timer(
    flow_id,
    StepExecutionId("TimerExampleStep"),
    TimerId.by_condition_index(0),
)

How to skip a Timer​

How to skip a Timer