feat: async pubsub #381
feat: async pubsub #381
Conversation
canvas_sdk/utils/http.py
Outdated
| @@ -26,7 +26,7 @@ def wrapper(self: "Http", *args: Any, **kwargs: Any) -> Any: | |||
| result = fn(self, *args, **kwargs) | |||
| end_time = time.time() | |||
| timing = int((end_time - start_time) * 1000) | |||
| self.statsd_client.timing(f"http_{fn.__name__}", timing) | |||
| self.statsd_client.timing(f"plugins.http_{fn.__name__}", timing) | |||
There was a problem hiding this comment.
This change is now in main
There was a problem hiding this comment.
jamagalhaes
force-pushed
the
feat/async-logger
branch
from
2cc23a2 to
fb946d3
Compare
| @@ -10,12 +12,18 @@ class PubSubLogHandler(logging.Handler): | |||
|
|
|||
| def __init__(self) -> None: | |||
| self.publisher = Publisher() | |||
| self.async_publisher = AsyncPublisher() | |||
There was a problem hiding this comment.
Should we also use this in the synchronizer? Massively lower volume of messages to be published there, but would be a win for consistency perhaps.
There was a problem hiding this comment.
async_publisher will be used within the synchronizer each time you do log.<level> because synchronizer runs in an async context
There was a problem hiding this comment.
I mean when publishing the message to pubsub for other containers to reload
| return redis.Redis.from_url(self.redis_endpoint, decode_responses=True) | ||
|
|
||
| return None | ||
| def _create_client(self) -> redis.Redis | redis.asyncio.Redis | None: ... |
There was a problem hiding this comment.
If we want PubSubBase to be an abstract base class, we should have it inherit from ABC and mark this method as an abstractmethod.
| def _create_client(self) -> redis.Redis | redis.asyncio.Redis | None: ... | |
| @abstractmethod | |
| def _create_client(self) -> redis.Redis | redis.asyncio.Redis | None: ... |
| try: | ||
| loop = asyncio.get_running_loop() | ||
| loop.create_task(self.async_publisher.publish(message)) | ||
| except RuntimeError: |
There was a problem hiding this comment.
The lack of specificity of this exception makes me mildly concerned since there are two actions in the try. I don't know if this is a valid concern or not since I don't know what could cause create_task to raise, but what would you think about this small refactor?
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = None
if loop:
loop.create_task(self.async_publisher.publish(message))
else:
self.publisher.publish(message)
in response to #333 (comment)