| @@ -39,17 +39,35 @@ from aiohttp import web | |||||
| from RainEagle.parse import LogDir as RELogDir, _cmaiter | from RainEagle.parse import LogDir as RELogDir, _cmaiter | ||||
| class StickyChannel(object): | class StickyChannel(object): | ||||
| '''A pub/sub style class. Any objects that are posted (via the | |||||
| post method), are received by any async iterators over the instance. | |||||
| The async iterator can be manually created by calling __aiter__ on | |||||
| an instance, and then on the returned object, awaiting on the coro | |||||
| __anext__. | |||||
| This is sticky in that the first returned value will be the last | |||||
| value that was posted, if there was one. If there was not one, | |||||
| (or the last value posted was None), it will wait and return the | |||||
| next value is posted.''' | |||||
| def __init__(self): | def __init__(self): | ||||
| self._queues = set() | self._queues = set() | ||||
| self._lastvalue = None | self._lastvalue = None | ||||
| async def post(self, value): | async def post(self, value): | ||||
| '''Post a value, and distribute it to any subscribers that | |||||
| are iterating over this class.''' | |||||
| self._lastvalue = value | self._lastvalue = value | ||||
| for i in self._queues: | for i in self._queues: | ||||
| await i.put(value) | await i.put(value) | ||||
| async def __aiter__(self): | async def __aiter__(self): | ||||
| '''An async generator that will return each object that is | |||||
| posted.''' | |||||
| q = asyncio.Queue() | q = asyncio.Queue() | ||||
| try: | try: | ||||