77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
from app.clients.mci_server_client import MciServerClient
|
|
from app.models.event import EventStatus, OutboundEvent
|
|
from app.repositories.event_store import EventStore
|
|
|
|
|
|
class EventPublisher:
|
|
def __init__(
|
|
self,
|
|
store: EventStore,
|
|
client: MciServerClient,
|
|
max_retry: int = 5,
|
|
base_delay_seconds: int = 5,
|
|
interval_seconds: float = 5.0,
|
|
batch_size: int = 100,
|
|
):
|
|
self.store = store
|
|
self.client = client
|
|
self.max_retry = min(max_retry, 10)
|
|
self.base_delay_seconds = base_delay_seconds
|
|
self.interval_seconds = interval_seconds
|
|
self.batch_size = batch_size
|
|
self._task: Optional[asyncio.Task] = None
|
|
self._stop_event = asyncio.Event()
|
|
|
|
def start(self) -> None:
|
|
self._task = asyncio.create_task(self._loop())
|
|
|
|
async def stop(self) -> None:
|
|
self._stop_event.set()
|
|
if self._task:
|
|
try:
|
|
await asyncio.wait_for(self._task, timeout=5.0)
|
|
except asyncio.TimeoutError:
|
|
self._task.cancel()
|
|
|
|
async def _loop(self) -> None:
|
|
cleanup_counter = 0
|
|
cleanup_interval = max(1, int(24 * 3600 / self.interval_seconds))
|
|
while not self._stop_event.is_set():
|
|
try:
|
|
await self._tick()
|
|
except Exception:
|
|
pass
|
|
cleanup_counter += 1
|
|
if cleanup_counter >= cleanup_interval:
|
|
try:
|
|
await self.cleanup_old_events(10)
|
|
except Exception:
|
|
pass
|
|
cleanup_counter = 0
|
|
try:
|
|
await asyncio.wait_for(self._stop_event.wait(), timeout=self.interval_seconds)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
async def _tick(self) -> None:
|
|
events = self.store.fetch_pending(limit=self.batch_size)
|
|
for event in events:
|
|
try:
|
|
await self.client.post_event(event)
|
|
self.store.mark_delivered(event.event_id)
|
|
except Exception as e:
|
|
next_attempt = event.attempt_count + 1
|
|
delay = self.base_delay_seconds * (2 ** (next_attempt - 1))
|
|
next_retry_at = datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(seconds=delay)
|
|
status = EventStatus.DEAD_LETTER if next_attempt >= self.max_retry else EventStatus.RETRYING
|
|
self.store.mark_retry(event.event_id, status, next_attempt, next_retry_at, str(e)[:500])
|
|
|
|
async def cleanup_old_events(self, retention_days: int) -> int:
|
|
return self.store.cleanup(retention_days)
|