diff --git a/faststream_redis_timers/broker.py b/faststream_redis_timers/broker.py index 8ec9e5a..2a82f71 100644 --- a/faststream_redis_timers/broker.py +++ b/faststream_redis_timers/broker.py @@ -1,4 +1,5 @@ import logging +import sys import typing from collections.abc import Iterable, Sequence from datetime import datetime, timedelta @@ -66,7 +67,6 @@ def _spec_url(client: "RedisClient | None", timeline_key: str) -> list[str]: class TimersParamsStorage(DefaultLoggerStorage): - __max_msg_id_ln = -1 _max_channel_name = 7 def register_subscriber(self, params: dict[str, typing.Any]) -> None: @@ -78,13 +78,8 @@ def get_logger(self, *, context: "ContextRepo") -> LoggerProto: logger = get_broker_logger( name="timers", default_context={"channel": "", "message_id": ""}, - message_id_ln=self.__max_msg_id_ln, - fmt=( - "%(asctime)s %(levelname)-8s - " - f"%(channel)-{self._max_channel_name}s | " - f"%(message_id)-{self.__max_msg_id_ln}s " - "- %(message)s" - ), + message_id_ln=sys.maxsize, + fmt=(f"%(asctime)s %(levelname)-8s - %(channel)-{self._max_channel_name}s | %(message_id)s - %(message)s"), context=context, log_level=self.logger_log_level, ) diff --git a/tests/test_log_context.py b/tests/test_log_context.py new file mode 100644 index 0000000..e775680 --- /dev/null +++ b/tests/test_log_context.py @@ -0,0 +1,52 @@ +import asyncio +import logging +import uuid +from collections import Counter + +from redis.asyncio import Redis + +from faststream_redis_timers import TimersBroker + + +class _RecordCollector(logging.Handler): + def __init__(self) -> None: + super().__init__() + self.records: list[logging.LogRecord] = [] + + def emit(self, record: logging.LogRecord) -> None: + self.records.append(record) + + +async def test_received_and_processed_logs_name_each_timer(redis_client: Redis) -> None: + suffix = uuid.uuid4().hex + broker = TimersBroker( + redis_client, + timeline_key=f"logctx_tl_{suffix}", + payloads_key=f"logctx_pl_{suffix}", + ) + timer_ids = [f"t-{n}" for n in range(12)] + handled: list[str] = [] + all_done = asyncio.Event() + + @broker.subscriber("topic", max_concurrent=4) + async def handler(body: str) -> None: + handled.append(body) + await asyncio.sleep(0.01) + if len(handled) == len(timer_ids): + all_done.set() + + collector = _RecordCollector() + access_logger = logging.getLogger("faststream.access.timers") + access_logger.addHandler(collector) + try: + async with broker: + for timer_id in timer_ids: + await broker.publish(timer_id, topic="topic", timer_id=timer_id) + await asyncio.wait_for(all_done.wait(), timeout=5.0) + finally: + access_logger.removeHandler(collector) + + assert sorted(handled) == sorted(timer_ids) + for event in ("Received", "Processed"): + logged = Counter(r.message_id for r in collector.records if r.getMessage() == event) # ty: ignore[unresolved-attribute] + assert logged == Counter(timer_ids), event