How to implement an async message bus

[Nagy László Zsolt]

>>>     try:
>>>         return await waiter
>>>     finally:
>>>         # TODO: Use a context manager to add and remove the keys.
>>>         for key in keys:
>>>             self._waiters[key].discard(waiter)
>>>         if handle:
>>>             handle.cancel()
>>>
>>> def notify(self, key, message):
>>>     if key in self._waiters and self._waiters[key]:
>>>         waiter = next(iter(self._waiters[key]))
>>>         waiter.set_result((key, message))
>> I think this is what I needed. I'm going to try this tomorrow.
> Yes, putting aside the asyncio/tornado distinction, I think a Future
> will still solve the problem for you.
No, it won't. :-( 

Finally, I got it working, by replacing asyncio.Future with
tornado.concurrent.Future.

At least it is consistent. But it is also missing key features. For
example, there is no tornado.concurrent.Condition.