62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
"""worker-notify entrypoint: connects to RabbitMQ and runs the notify consumer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import signal
|
|
|
|
from src.contract_check.core.config import get_settings
|
|
from src.contract_check.core.logging import bind_context, configure_logging, get_logger
|
|
from src.contract_check.core.metrics import start_metrics_server
|
|
from src.contract_check.core.sentry import init_sentry
|
|
from src.contract_check.core.telemetry import setup_telemetry, shutdown_telemetry
|
|
from src.contract_check.worker_notify.consumer import NotifyConsumer
|
|
|
|
log = get_logger(__name__)
|
|
|
|
|
|
async def main() -> None:
|
|
settings = get_settings()
|
|
configure_logging(
|
|
settings.log_level,
|
|
json_output=settings.json_logs,
|
|
service="worker-notify",
|
|
env=settings.env,
|
|
)
|
|
bind_context(service="worker-notify", env=settings.env)
|
|
init_sentry("worker-notify")
|
|
setup_telemetry("worker-notify")
|
|
|
|
start_metrics_server(9103)
|
|
|
|
consumer = NotifyConsumer(
|
|
url=settings.rabbitmq_url,
|
|
origin="worker-notify",
|
|
prefetch=settings.mq_prefetch_notify,
|
|
max_attempts=settings.mq_max_attempts,
|
|
retry_base_ms=settings.mq_retry_base_ms,
|
|
)
|
|
await consumer.connect()
|
|
|
|
loop = asyncio.get_running_loop()
|
|
stop_event = asyncio.Event()
|
|
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
loop.add_signal_handler(sig, stop_event.set)
|
|
|
|
consumer_task = asyncio.create_task(consumer.run())
|
|
stop_task = asyncio.create_task(stop_event.wait())
|
|
|
|
log.info("worker_notify_started", prefetch=settings.mq_prefetch_notify)
|
|
try:
|
|
await asyncio.wait(
|
|
{consumer_task, stop_task},
|
|
return_when=asyncio.FIRST_COMPLETED,
|
|
)
|
|
finally:
|
|
await consumer.stop()
|
|
shutdown_telemetry()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|