95b6560ffd
Updated comment to specify that the subscriber starts in the background, enhancing code readability and understanding of the asynchronous behavior. Signed-off-by: zhenyus <zhenyus@mathmast.com>
36 lines
1.5 KiB
Python
36 lines
1.5 KiB
Python
import asyncio
|
|
from app.backend.infra.rabbitmq.async_subscriber import AsyncMQSubscriber
|
|
from app.backend.services.deployment_status_update_service import DeploymentStatusUpdateService
|
|
|
|
|
|
def register(app):
|
|
# Initialize the message subscriber and status update service
|
|
app.deployment_heartbeat_subscriber = AsyncMQSubscriber("devops_reconcile_heartbeat")
|
|
app.deployment_status_service = DeploymentStatusUpdateService()
|
|
|
|
@app.on_event("startup")
|
|
async def start_message_consumers():
|
|
print("Starting message consumers")
|
|
# Register the heartbeat processor
|
|
await app.deployment_heartbeat_subscriber.register_consumer(
|
|
registry_key="deployment_heartbeat_processor",
|
|
callback_method=app.deployment_status_service.process_heartbeat_message,
|
|
args={}
|
|
)
|
|
print("Registered deployment heartbeat processor")
|
|
|
|
# Start the subscriber in the background
|
|
loop = asyncio.get_running_loop()
|
|
loop.create_task(
|
|
app.deployment_heartbeat_subscriber.subscribe(max_retries=5, event_loop=loop)
|
|
)
|
|
print("Started deployment heartbeat subscriber")
|
|
|
|
@app.on_event("shutdown")
|
|
async def stop_message_consumers():
|
|
# Clear consumers and close connection
|
|
print("Stopping message consumers")
|
|
await app.deployment_heartbeat_subscriber.clear_all_consumers()
|
|
print("Cleared all consumers")
|
|
await app.deployment_heartbeat_subscriber.close()
|
|
print("Closed deployment heartbeat subscriber") |