Inactiveness Tracker Timer

当一个动作只应在一段时间没有活动后才发生时，使用这个模式。InactivenessTrackerFlow 从 TrackerStep 开始，等待 Timer 或 Active Channel。活动先到达时，TrackerStep 跳回自身并开始一个新的 Timer；Timer 先触发时，Flow 跳转到 ProcessInactivenessStep。

Definition graph

InactivenessTrackerFlow

Valid

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

InactivenessTrackerFlow

Active

record_activity

ProcessInactivenessStep

TrackerStepSTART

gracefulComplete

WaitFor anyOf

⏱️ TRACKER_DURATION timer📨 Active.for 1

?

goTo

🧩 ProcessInactivenessStep

goTo

🧩 TrackerStep

Press enter or space to select a node. You can then use the arrow keys to move the node around. Press delete to remove it and escape to cancel.

Press enter or space to select an edge. You can then press delete to remove it or escape to cancel.

核心 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)

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

相关内容

• Timer
• Channel