birbnetes/input-service
birbnetes/input-service
4
0
Fork 0
Code Pull Requests Releases Activity

Fixed routing keys
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Pünkösd Marcell 2021-08-18 15:56:24 +02:00
parent 17bf93f8fd
commit eecb65894d
1 changed files with 4 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/magic_amqp.py
View File

@ -75,7 +75,7 @@ class MagicAMQP:
if total_time > 1:
self.app.logger.warning(f"Loop: Total loop took {total_time:5f} sec")
def _publish(self, exchange: str, payload=None):
def _publish(self, exchange: str, routing_key: str, payload=None):
"""
Publish a simple json serialized message to the configured queue.
If the connection is broken, then this call will block until the connection is restored
@ -94,7 +94,7 @@ class MagicAMQP:
try:
self._pika_channel.basic_publish(
exchange=exchange,
routing_key='feature',
routing_key=routing_key,
body=json.dumps(payload).encode('UTF-8')
)
self.app.logger.debug(f"Published: {payload}")
@ -125,10 +125,10 @@ class MagicAMQP:
self.app.logger.warning(f"Publish: Total publish took {total_time:5f} sec")
def publish_cache(self, payload=None):
return self._publish(self.app.config['EXCHANGE_NAME_CACHE'], payload)
return self._publish(self.app.config['EXCHANGE_NAME_CACHE'], "cache", payload)
def publish_meta(self, payload=None):
return self._publish(self.app.config['EXCHANGE_NAME_META'], payload)
return self._publish(self.app.config['EXCHANGE_NAME_META'], "meta", payload)
def is_healthy(self) -> bool:
with self._lock: