I'm assuming the API doesn't look like this:

await qmp.execute('stop')
await qmp.wait_for_event('STOP')

Because you need to start listening to events before executing the action that might trigger the event. Is that correct?

I think your proposal is a good low level API, but I'm worried that people could be tempted to write the code like the following, because it look deceptively simpler:

await qmp.execute('stop')
await qmp.listener('STOP').get()

Can we provide a higher level API that makes this kind of mistake less likely? e.g. create a listener that will listen to all events by default, and providing a wait_for_event() wrapper that will wait for that specific event?

That's right, the listener needs to be in place before the command is issued so that we have the ability to trap it. This is intended to be a fairly low-level API, but it's one where I'd like the usability to be the focus, so if the existing pattern feels cumbersome, it's something I'd like to improve. (I am not sure it's possible to do safely to my satisfaction.)

There is a default listener that captures all events, in qmp.events:

await qmp.execute('stop')
await qmp.events.get()

but it has the usual problems of a FIFO queue; the event you get next may not be the one you wanted; it could be stale or it might not be the STOP event. See my notes above for why I am skeptical of the tertiary filtering interface I made -- this falls into a similar category. e.g. get(type='STOP') feels dangerous because it may discard events that were important to someone else. I am wary of this. I am wary in general of anyone using the built-in all-events listener for doing much of anything except handling all events with something like a logger, in fact. We are using the all-events listener in aqmp-shell to render async events received to the screen, for instance.

wait_for_event() has problems in that it may toss certain events to the floor -- improving the design of the all-events listener would be nice, but I am at a loss for how to do it in a way that encourages safe use without also being cumbersome.

I'd be open to doing something like async def wait_for(name=None, filter=None, data={}) for more powerful tertiary filtering, but I still can't shake the feeling that this interface is dangerous and prone to abuse. What should be the default behavior if an event enters the stream that isn't the one we want? We can't just put other events back into the queue and take only the ones we want, we are necessarily committing to be the sole consumer of all prior events.

Maybe .get() can remain a simple "fetch next" interface and "wait_for" can be a more complicated interface that returns the list of discarded events, but then it'd be up to the caller to ensure that this list is empty or otherwise only contains events that we were comfortable with discarding.

We'd also want to ensure that we cleared the event stream prior to invoking it to make sure the events were time-boxed somehow, which means having the necessary foresight to call 'clear()' beforehand -- also making sure we didn't clear anything that might have been important.

However, If we are talking about a higher-level API, we could do something like this:

class SDK:
    def __init__(self, qmp: QMP):
        self.qmp = qmp

    async def stop(self):
        with qmp.listener('STOP') as listener:
            await self.qmp.execute('stop')
            await listener.get()

At a lower level, we might be able to provide something like execute_and_wait():

class QMP:
    ...

    async def execute_and_wait(self, cmd, arguments, event_name):
        with self.listener(event_name) as listener:
            await self._do_the_execute_stuff(cmd, arguments)
            await listener.get()

This provides a nice convenient wrapper for a common idiom, though the flexibility is again bad, as it relies upon merely an event name and only awaits for one such event. It'd not work well for things like jobs that expect many events related to a single execution, and we lose the listener after the function exits.

Another option if we are discussing high-level APIs is to think about a VM or Machine class that simply captures all state changes into a series of asyncio.Event() primitives, so we could do something like this:

(Keep in mind, this is just a hasty example, runstate management is a lot more complex than this with many more states. The key point is using asyncio.Event() to represent the effect/meaning of an Event, abstracting the event delivery entirely.)

class VM:
    def __init__(self, ...):
        self.qmp = QMP(...)
        self.running = asyncio.Event()
        self.stopped = asyncio.Event()

    async def state_monitor(self):
        # Pretend this is running as a concurrent Task
        with self.listener(('STOP', 'RESUME', ...)) as listener:
            async for event in listener:
                if event['event'] == 'RESUME':
                    self.running.set()
                    self.stopped.clear()
                elif event['event'] == 'STOP':
                    self.running.clear()
                    self.stopped.set()

    async def stop(self):
        await self.qmp.execute('stop')
        await self.stopped.wait()

This leverages a cooperative task that is always reading state changes and relies upon the runstate of the VM instead, providing a more semantic view of stop().

I am open to suggestions, but keep in mind that for this library I wish to avoid providing high-level semantic commands that have any awareness of the particular QMP vocabulary we are dealing with; it's a low-level API for QMP only.

(Still, providing a replacement for iotests is one of my key design goals, so having certain kind of conveniences feels important.)