Refactor to support Async consumers #67
Conversation
Update to support streaming MQ connections
Add `ready` property to check IrisConnector status Update dependencies Resolve unit test failure Add test coverage for `IrisConnector`
NeonDaniel
changed the title
neon_iris/mq_connector.py
Outdated
| def wait_for_connection(self): | ||
| LOG.info("Waiting for connection") | ||
| while not self._ready.is_set(): | ||
| sleep(0.5) |
There was a problem hiding this comment.
Sleep isn't thread safe, is it?
There was a problem hiding this comment.
Ah, probably not. I think the issue I ran into with run(self._ready.wait()) is that somehow the event would be set, but wait would not return. This was really more of a temporary workaround, so I will look into how to fix this
There was a problem hiding this comment.
Upon further investigation, the issue appears related to the SelectConnection using its own IOLoop. I tried passing a specific EventLoop, but that is apparently incompatible with pika. I settled on a working solution that still uses sleep, but does use the connection's ioloop which I believe makes that thread safe..
Looking back, I'm not sure this is functionally different than what was originally there and sleep isn't threadsafe, but it probably doesn't have to be in the current implementation since the even't we're checking is threadsafe..
…her than periodic checking Add type annotations to new methods
Description
Refactors MQ handling to use a connector class, similar to other services
Adds unit test coverage for new connector
Update dependencies
Issues
Other Notes
Currently deployed to https://iris.neonaialpha.com/