4f4037b4ae
- nats_consumer/grpc_server: annotate nc/sub/server attrs (were inferred as None-only from __init__, breaking on the real assignment later). - active_read: narrow on 'if names:' instead of the derived 'persisted' bool so mypy sees names as non-Optional inside the publish_reading call. - pyproject: exclude ingestor/proto/ (protoc/grpcio-tools output) from mypy's file walk and skip following it as an import - it's generated, not hand-maintained, and mypy can't see its runtime-built message classes.
56 lines
1.8 KiB
Python
56 lines
1.8 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. No reconnect logic — nats-py handles reconnection internally.
|
|
"""
|
|
|
|
import json
|
|
|
|
import nats
|
|
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',
|
|
):
|
|
self.nats_url = nats_url
|
|
self.write_engine = write_engine
|
|
self.logger = logger
|
|
self.subject = subject
|
|
self.queue = queue
|
|
self.nc: Client | None = None
|
|
self.sub: Subscription | None = None
|
|
|
|
async def start(self) -> None:
|
|
self.nc = await nats.connect(servers=self.nats_url)
|
|
self.sub = await self.nc.subscribe(self.subject, queue=self.queue, cb=self._handle)
|
|
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
|
|
await self.write_engine.write(payload)
|
|
|
|
async def stop(self) -> None:
|
|
if self.nc is None:
|
|
return
|
|
if self.sub is not None:
|
|
await self.sub.unsubscribe()
|
|
await self.nc.drain()
|