如何用PyTransition取消异步转换

Question

我有一个排队的PyTransition状态机，有些状态在on_enter中工作。但是，我希望用户能够在任何时候停止机器而不需要等待。为此，我需要一种取消过渡的方法。

这是我现在所发现的。然而，访问_transition_queue_dict似乎是一次黑客攻击。有合适的方法吗？

#!/usr/bin/env python3

import asyncio
import logging

from transitions.extensions.asyncio import AsyncMachine

logging.getLogger('transitions').setLevel(logging.DEBUG)


class Model:
    async def do_long_work(self):
        print("Working...")
        await asyncio.sleep(10)
        print("Worked!")

    async def print_stop(self):
        print("Stopped!")

    async def interrupt(self):
        global machine
        await asyncio.sleep(1)
        for task in machine.async_tasks[id(self)]:
            task.cancel()
        machine._transition_queue_dict[id(model)].clear()

        await self.stop()

model = Model()
machine = AsyncMachine(model=model, queued=True)

machine.add_states('running', on_enter=[model.do_long_work])
machine.add_states('stopped', on_enter=[model.print_stop])
machine.add_transition('run', 'initial', 'running')
machine.add_transition('stop', 'running', 'stopped')


async def run():
    await asyncio.gather(machine.dispatch('run'), model.interrupt())

asyncio.run(run())

我使用最后一次提交主(3836dc4)。

Answer 1

这里的问题是您传递了queued=True，它指示机器将新事件放入模型队列并依次处理事件。由于您希望能够中断事件，所以当您转换掉/退出状态时，省略queued=True或设置queued=False (默认值)将取消事件。在这种情况下，不需要修改内部队列。

import asyncio
import logging

from transitions.extensions.asyncio import AsyncMachine

logging.getLogger('transitions').setLevel(logging.DEBUG)


class Model:
    async def do_long_work(self):
        print("Working...")
        await asyncio.sleep(10)
        print("Worked!")

    async def print_stop(self):
        print("Stopped!")

    async def interrupt(self):
        await asyncio.sleep(1)
        await self.stop()


model = Model()
machine = AsyncMachine(model=model, queued=False)

machine.add_states('running', on_enter=[model.do_long_work])
machine.add_states('stopped', on_enter=[model.print_stop])
machine.add_transition('run', 'initial', 'running')
machine.add_transition('stop', 'running', 'stopped')


async def run():
    await asyncio.gather(machine.dispatch('run'), model.interrupt())

asyncio.run(run())
# Working...
# Stopped!