Skip to main content

Inactiveness Tracker Timer

Use this pattern when an action should happen only after a period without activity. InactivenessTrackerFlow starts with TrackerStep. It waits for either the Timer or the Active Channel. Activity arrives first, so TrackerStep moves to itself and starts a fresh Timer. If the Timer fires first, the Flow moves to ProcessInactivenessStep.

Definition graph

InactivenessTrackerFlow

Valid

python · examples/python/dex_examples/patterns/inactiveness-tracker-timer/inactiveness_tracker_flow.py

Core Step​

class TrackerStep(Step[None]):
    def __init__(self, process_inactiveness: ProcessInactivenessStep, active_channel: Channel[None]) -> None:
        self.process_inactiveness = process_inactiveness
        self.active_channel = active_channel

    def wait_for(self, context: Context, input: None) -> Wait:
        del context, input
        return Wait.any_of(Timer.by_duration(TRACKER_DURATION), self.active_channel.for_one())

    def execute(self, context: Context, input: None) -> StepDecision:
        del input
        if context.has_timer_fired():
            return go_to(ProcessInactivenessStep, None)
        return go_to(TrackerStep, None)

Example: examples/python/dex_examples/patterns/inactiveness-tracker-timer/inactiveness_tracker_flow.py