73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
import json
|
|
import logging
|
|
from typing import Awaitable, Callable
|
|
|
|
from websockets.asyncio.client import connect
|
|
from websockets.exceptions import WebSocketException
|
|
|
|
from .config import PropertyApiConfig
|
|
from .listings import Listing, listing_from_event
|
|
|
|
logger = logging.getLogger(__name__)
|
|
ListingSink = Callable[[Listing], Awaitable[None]]
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ConnectionStatus:
|
|
connected: bool = False
|
|
last_connected_at: datetime | None = None
|
|
last_event_at: datetime | None = None
|
|
last_error: str | None = None
|
|
|
|
|
|
class ListingEventConsumer:
|
|
def __init__(self, config: PropertyApiConfig, on_listing: ListingSink) -> None:
|
|
self._config = config
|
|
self._on_listing = on_listing
|
|
self.status = ConnectionStatus()
|
|
|
|
async def handle_message(self, raw_message: str | bytes) -> bool:
|
|
try:
|
|
payload = json.loads(raw_message)
|
|
except (TypeError, json.JSONDecodeError):
|
|
logger.warning("Ignoring invalid WebSocket JSON event")
|
|
return False
|
|
listing = listing_from_event(payload)
|
|
if listing is None:
|
|
return False
|
|
await self._on_listing(listing)
|
|
self.status.last_event_at = datetime.now(UTC)
|
|
return True
|
|
|
|
async def run(self) -> None:
|
|
delay = self._config.reconnect_initial_delay_seconds
|
|
while True:
|
|
try:
|
|
async with connect(
|
|
self._config.websocket_url,
|
|
open_timeout=self._config.connect_timeout_seconds,
|
|
ping_interval=20,
|
|
ping_timeout=20,
|
|
) as websocket:
|
|
self.status.connected = True
|
|
self.status.last_connected_at = datetime.now(UTC)
|
|
self.status.last_error = None
|
|
delay = self._config.reconnect_initial_delay_seconds
|
|
logger.info("Connected to property API: %s", self._config.websocket_url)
|
|
async for message in websocket:
|
|
await self.handle_message(message)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except (OSError, WebSocketException, TimeoutError) as error:
|
|
self.status.last_error = str(error)
|
|
logger.warning("Property API connection failed; retrying in %.1fs: %s", delay, error)
|
|
finally:
|
|
self.status.connected = False
|
|
await asyncio.sleep(delay)
|
|
delay = min(delay * 2, self._config.reconnect_max_delay_seconds)
|