SWAMPy

✎ CARVE FLESH

EDITING: nats_gateway.py

import contextlib
import logging
from typing import Generator

from defence360agent.api.server import send_message, NATSSendMessageException
from defence360agent.contracts.plugins import MessageSink
from defence360agent.plugins.client import SendToServerClient
from defence360agent.utils import Scope

logger = logging.getLogger(__name__)

class SendToServerNATS(SendToServerClient, MessageSink):
    SCOPE = Scope.IM360
    SHUTDOWN_PRIORITY = 900  # Shutdown late, after Accumulate has flushed

def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._nats_api = None

@contextlib.contextmanager
    def _get_api(
        self,
    ) -> Generator[send_message.NATSGatewayAPI, None, None]:
        # Reuse the same API instance to keep the NATS connection alive
        if self._nats_api is None:
            self._nats_api = send_message.NATSGatewayAPI()
        yield self._nats_api

async def _send_pending_messages(self) -> None:
        if self._pending.empty():
            return None

if self._shutting_down.is_set():
            logger.warning("Shutdown signal received, skipping NATS send")
            return None

messages = self._pending.pop_all()
        logger.info("Sending %s messages via NATS", len(messages))
        with self._get_api() as api:
            try:
                await api.send_messages(messages)
                if self._pending.qsize() > 0:
                    logger.info(
                        "Still need to send %s messages",
                        self._pending.qsize(),
                    )
            except NATSSendMessageException as e:
                unsent = messages[e.published :]
                if unsent:
                    self._pending.put_many(unsent)
                logger.warning(
                    "Failed to send messages via NATS: "
                    "%d published, %d re-queued: %s",
                    e.published,
                    len(unsent),
                    e,
                )
            except BaseException:
                # CancelledError (BaseException in 3.9+) or other fatal
                # errors — re-queue all messages to avoid data loss.
                self._pending.put_many(messages)
                raise

async def shutdown(self) -> None:
        await super().shutdown()
        if self._nats_api is not None:
            await self._nats_api.close()
            logger.info("NATS connection closed")

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📁 __pycache__/	DIR	—	2026-08-28 11:40
📁 api/	DIR	—	2026-08-28 11:40
📁 application/	DIR	—	2026-08-28 11:40
📁 contracts/	DIR	—	2026-08-28 11:40
📁 files/	DIR	—	2026-08-28 11:40
📁 internals/	DIR	—	2026-08-28 11:40
📁 ioc/	DIR	—	2026-08-28 11:40
📁 migrations/	DIR	—	2026-08-28 11:40
📁 model/	DIR	—	2026-08-28 11:40
📁 plugins/	DIR	—	2026-08-28 11:40
📁 simple_rpc/	DIR	—	2026-08-28 11:40
📁 subsys/	DIR	—	2026-08-28 11:40
📁 utils/	DIR	—	2026-08-28 11:40
📄 __init__.py	PY	98 B	2026-08-11 10:37	EDIT
📄 __main__.py	PY	33 B	2026-08-11 10:37	EDIT
📄 _version.py	PY	83 B	2026-08-11 10:37	EDIT
📄 aibolit_job.py	PY	4.5 KB	2026-08-11 10:37	EDIT
📄 cli.py	PY	344 B	2026-08-11 10:37	EDIT
📄 nats_gateway.py	PY	2.4 KB	2026-08-11 10:37	EDIT
📄 rpc_handlers.py	PY	615 B	2026-08-11 10:37	EDIT
📄 run.py	PY	5.8 KB	2026-08-11 10:37	EDIT