Compare commits

...

1 Commits

Author SHA1 Message Date
hurui200320 248469fe1e fix(General): Fix deps version so cleverswarm can install it
Unit test coverage / pytest (push) Failing after 1m0s
2025-04-30 21:37:26 +08:00
3 changed files with 15 additions and 17 deletions
+3 -4
View File
@@ -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)
+8 -9
View File
@@ -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
View File
@@ -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",