e94b489196
- One call per pytest.raises block (S5778): build the guarded start() before the block and nest the patch outside it. - The NATS callbacks stay async on purpose: nats-py rejects callbacks that are not coroutine functions. Comment it and assert it in the callback test, so nobody drops the async keyword to please S7503. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
125 lines
4.7 KiB
Python
125 lines
4.7 KiB
Python
"""NATS input adapter for OPC write commands (contract C2).
|
|
|
|
Thin adapter: decode the message JSON and hand the dict to WriteEngine,
|
|
which is already raise-safe against malformed commands. No metrics of its
|
|
own — the engine records write metrics; a malformed-JSON drop is logged
|
|
here. Once connected, nats-py reconnects by itself, forever.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import math
|
|
import traceback
|
|
|
|
from nats.aio.client import Client
|
|
from nats.aio.subscription import Subscription
|
|
from sientia_do.observability.logger import Logger
|
|
|
|
from ingestor.engines.write import WriteEngine
|
|
|
|
|
|
class NatsWriteConsumer:
|
|
def __init__(
|
|
self,
|
|
nats_url: str,
|
|
write_engine: WriteEngine,
|
|
logger: Logger,
|
|
subject: str = 'opc.write',
|
|
queue: str = 'gateway-writers',
|
|
connect_timeout_s: float = 5.0,
|
|
):
|
|
self.nats_url = nats_url
|
|
self.write_engine = write_engine
|
|
self.logger = logger
|
|
self.subject = subject
|
|
self.queue = queue
|
|
self.connect_timeout_s = connect_timeout_s
|
|
self.nc: Client | None = None
|
|
self.sub: Subscription | None = None
|
|
|
|
@property
|
|
def closed(self) -> bool:
|
|
"""True when there is no connection, or it was closed for good."""
|
|
return self.nc is None or self.nc.is_closed
|
|
|
|
async def start(self) -> None:
|
|
"""Connect and subscribe, giving up after ``connect_timeout_s``.
|
|
|
|
It runs from the main loop (startup and ``GatewayRuntime.idle_sweep``), so it
|
|
must never block it: nats-py's defaults retry the first connection 60 times,
|
|
about 2 minutes, longer than the heartbeat TTL, and with
|
|
``max_reconnect_attempts=-1`` they would retry it forever. Once connected,
|
|
reconnection is unbounded, so an outage never closes the connection for good.
|
|
|
|
Raises:
|
|
Exception: When the connection or the subscription failed; nothing is left
|
|
open in that case.
|
|
"""
|
|
nc = Client()
|
|
try:
|
|
await asyncio.wait_for(
|
|
nc.connect(
|
|
servers=self.nats_url,
|
|
connect_timeout=math.ceil(self.connect_timeout_s),
|
|
max_reconnect_attempts=-1,
|
|
name='opc-gateway',
|
|
# One log line per event instead of nats-py's traceback per attempt.
|
|
error_cb=self._on_error,
|
|
disconnected_cb=self._on_disconnected,
|
|
reconnected_cb=self._on_reconnected,
|
|
),
|
|
timeout=self.connect_timeout_s,
|
|
)
|
|
self.sub = await nc.subscribe(self.subject, queue=self.queue, cb=self._handle)
|
|
except (Exception, asyncio.CancelledError):
|
|
await self._discard(nc)
|
|
raise
|
|
self.nc = nc
|
|
self.logger.info(f'NatsWriteConsumer started: subject={self.subject} queue={self.queue}')
|
|
|
|
async def _handle(self, msg) -> None:
|
|
try:
|
|
payload = json.loads(msg.data)
|
|
except ValueError as e: # includes JSONDecodeError and UnicodeDecodeError
|
|
preview = msg.data[:200]
|
|
self.logger.error(f'NatsWriteConsumer dropped malformed JSON ({e}): {preview!r}')
|
|
return
|
|
try:
|
|
await self.write_engine.write(payload)
|
|
except Exception as e:
|
|
# The engine is raise-safe; this only keeps an unexpected bug from going to
|
|
# nats-py's error callback without context.
|
|
self.logger.error(
|
|
f'NatsWriteConsumer: unexpected error handling a write command ({e}):\n'
|
|
f'{traceback.format_exc()}'
|
|
)
|
|
|
|
# These stay async with nothing to await: nats-py rejects callbacks that are not
|
|
# coroutine functions (InvalidCallbackTypeError on connect).
|
|
async def _on_error(self, e: Exception) -> None:
|
|
self.logger.warning(f'NATS error: {type(e).__name__} {e}')
|
|
|
|
async def _on_disconnected(self) -> None:
|
|
self.logger.warning('Disconnected from NATS; reconnecting')
|
|
|
|
async def _on_reconnected(self) -> None:
|
|
self.logger.info('Reconnected to NATS')
|
|
|
|
async def _discard(self, nc: Client) -> None:
|
|
try:
|
|
await nc.close()
|
|
except Exception as e:
|
|
self.logger.warning(f'Error closing a failed NATS connection: {e}')
|
|
|
|
async def stop(self) -> None:
|
|
if self.nc is None:
|
|
return
|
|
if self.nc.is_closed:
|
|
# Closed for good already: unsubscribe/drain would raise ConnectionClosedError
|
|
# and abort the shutdown sequence.
|
|
self.nc = None
|
|
return
|
|
if self.sub is not None:
|
|
await self.sub.unsubscribe()
|
|
await self.nc.drain()
|