k1nq
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
import asyncio
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
|
|
from faststream.rabbit import RabbitBroker
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
from app.messaging import PAYMENTS_EXCHANGE, PAYMENTS_ROUTING_KEY
|
|
from app.models import OutboxEvent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class OutboxPublisher:
|
|
def __init__(
|
|
self,
|
|
session_factory: async_sessionmaker,
|
|
broker: RabbitBroker,
|
|
*,
|
|
batch_size: int,
|
|
poll_interval: float,
|
|
publish_timeout: float,
|
|
) -> None:
|
|
self._session_factory = session_factory
|
|
self._broker = broker
|
|
self._batch_size = batch_size
|
|
self._poll_interval = poll_interval
|
|
self._publish_timeout = publish_timeout
|
|
|
|
async def publish_batch(self, stop_event: asyncio.Event | None = None) -> int:
|
|
async with self._session_factory() as session:
|
|
event_ids = list(
|
|
await session.scalars(
|
|
select(OutboxEvent.id)
|
|
.where(OutboxEvent.published_at.is_(None))
|
|
.order_by(OutboxEvent.created_at)
|
|
.limit(self._batch_size)
|
|
)
|
|
)
|
|
|
|
published = 0
|
|
for event_id in event_ids:
|
|
if stop_event is not None and stop_event.is_set():
|
|
break
|
|
|
|
async with self._session_factory() as session:
|
|
event = await session.scalar(
|
|
select(OutboxEvent)
|
|
.where(
|
|
OutboxEvent.id == event_id,
|
|
OutboxEvent.published_at.is_(None),
|
|
)
|
|
.with_for_update(skip_locked=True)
|
|
)
|
|
if event is None:
|
|
continue
|
|
|
|
try:
|
|
await self._broker.publish(
|
|
event.payload,
|
|
queue=PAYMENTS_ROUTING_KEY,
|
|
exchange=PAYMENTS_EXCHANGE,
|
|
persist=True,
|
|
message_id=str(event.id),
|
|
message_type=event.event_type,
|
|
timeout=self._publish_timeout,
|
|
)
|
|
except Exception as exc:
|
|
event.attempts += 1
|
|
event.last_error = str(exc)[:2000]
|
|
logger.exception("Failed to publish outbox event %s", event.id)
|
|
else:
|
|
event.published_at = datetime.now(UTC)
|
|
event.attempts += 1
|
|
event.last_error = None
|
|
published += 1
|
|
|
|
await session.commit()
|
|
|
|
return published
|
|
|
|
async def run(self, stop_event: asyncio.Event) -> None:
|
|
while not stop_event.is_set():
|
|
try:
|
|
await self.publish_batch(stop_event)
|
|
except Exception:
|
|
logger.exception("Outbox polling iteration failed")
|
|
|
|
try:
|
|
await asyncio.wait_for(stop_event.wait(), timeout=self._poll_interval)
|
|
except TimeoutError:
|
|
pass
|