Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
248469fe1e
|
@@ -1,8 +1,7 @@
|
||||
import logging
|
||||
|
||||
import aio_pika
|
||||
from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange
|
||||
from pika.exchange_type import ExchangeType
|
||||
from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange, ExchangeType
|
||||
|
||||
from amqp.adapter.data_message_factory import DataMessageFactory
|
||||
from amqp.adapter.logging_utils import (
|
||||
@@ -55,7 +54,7 @@ class RabbitMQClient:
|
||||
name=cm_reply_to_queue, durable=True, exclusive=False, auto_delete=False
|
||||
)
|
||||
self.reply_to_exchange = await self.channel.declare_exchange(
|
||||
AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.direct
|
||||
AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.DIRECT
|
||||
)
|
||||
await self.reply_to_queue.bind(
|
||||
exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue
|
||||
@@ -114,7 +113,7 @@ class RabbitMQClient:
|
||||
)
|
||||
exchange: AbstractRobustExchange = (
|
||||
await self.channel.declare_exchange(
|
||||
name=_exchange_name, type=ExchangeType.topic
|
||||
name=_exchange_name, type=ExchangeType.TOPIC
|
||||
)
|
||||
)
|
||||
await queue.bind(exchange, route.key)
|
||||
|
||||
@@ -15,9 +15,8 @@ from aio_pika.abc import (
|
||||
AbstractRobustQueue,
|
||||
AbstractRobustConnection,
|
||||
AbstractRobustExchange,
|
||||
AbstractIncomingMessage,
|
||||
AbstractIncomingMessage, ExchangeType,
|
||||
)
|
||||
from pika.exchange_type import ExchangeType
|
||||
|
||||
from amqp.adapter.logging_utils import (
|
||||
logging_error,
|
||||
@@ -82,13 +81,13 @@ class RouterProducer(RouterBase):
|
||||
# **** set up the service discovery internal exchanges ****
|
||||
# 1. declare RequestExchange to where to publish the RoutingData requests
|
||||
self.cm_request_exchange = await self.channel.declare_exchange(
|
||||
name=CM_REQUEST_EXCHANGE, type=ExchangeType.fanout
|
||||
name=CM_REQUEST_EXCHANGE, type=ExchangeType.FANOUT
|
||||
)
|
||||
#
|
||||
# 2. declare ResponseExchange where to publish current RoutingData (response to
|
||||
# a RoutingData request)
|
||||
self.cm_response_exchange = await self.channel.declare_exchange(
|
||||
name=CM_RESPONSE_EXCHANGE, type=ExchangeType.fanout
|
||||
name=CM_RESPONSE_EXCHANGE, type=ExchangeType.FANOUT
|
||||
)
|
||||
# 2a. declare a Response queue for Routing Data replies from Services
|
||||
_cm_response_queue: str = self.get_response_queue_name()
|
||||
@@ -145,9 +144,9 @@ class RouterProducer(RouterBase):
|
||||
cm_message.reply_to,
|
||||
)
|
||||
if (
|
||||
cm_message.is_valid()
|
||||
and not self.amq_configuration.amq_adapter.service_name
|
||||
== cm_message.reply_to
|
||||
cm_message.is_valid()
|
||||
and not self.amq_configuration.amq_adapter.service_name
|
||||
== cm_message.reply_to
|
||||
):
|
||||
if cm_message.message_type == ServiceMessageType.ROUTING_DATA:
|
||||
self.add_routes(cm_message.message)
|
||||
@@ -167,7 +166,7 @@ class RouterProducer(RouterBase):
|
||||
ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT
|
||||
)
|
||||
if not await self.publish_service_message(
|
||||
serviceMessage, self.cm_request_exchange
|
||||
serviceMessage, self.cm_request_exchange
|
||||
):
|
||||
logging_warning(
|
||||
"RouterProducer could not request route data: %s channel is not open.",
|
||||
@@ -178,7 +177,7 @@ class RouterProducer(RouterBase):
|
||||
logging_error("RouterProducer failed request routing data: %s", ioe)
|
||||
|
||||
async def publish_service_message(
|
||||
self, message: ServiceMessage, exchange: AbstractRobustExchange
|
||||
self, message: ServiceMessage, exchange: AbstractRobustExchange
|
||||
) -> bool:
|
||||
"""
|
||||
* Publishes a service message to the service queue and logs success log entry.
|
||||
|
||||
+4
-4
@@ -19,10 +19,10 @@ classifiers = [
|
||||
]
|
||||
requires-python = ">=3.11"
|
||||
dependencies = [
|
||||
"aio-pika~=9.5.5",
|
||||
"aiohttp~=3.11.11",
|
||||
"aio_pika~=9.5.5",
|
||||
"fastapi==0.114.2",
|
||||
"aio-pika~=9.5",
|
||||
"aiohttp~=3.11",
|
||||
"aio_pika~=9.5",
|
||||
"fastapi~=0.114",
|
||||
"opentelemetry-api",
|
||||
"opentelemetry-sdk",
|
||||
"opentelemetry-exporter-otlp",
|
||||
|
||||
Reference in New Issue
Block a user