__ __ __ __ _____ _ _ _____ _ _ _ | \/ | \ \ / / | __ \ (_) | | / ____| | | | | | \ / |_ __\ V / | |__) | __ ___ ____ _| |_ ___ | (___ | |__ ___| | | | |\/| | '__|> < | ___/ '__| \ \ / / _` | __/ _ \ \___ \| '_ \ / _ \ | | | | | | |_ / . \ | | | | | |\ V / (_| | || __/ ____) | | | | __/ | | |_| |_|_(_)_/ \_\ |_| |_| |_| \_/ \__,_|\__\___| |_____/|_| |_|\___V 2.1 if you need WebShell for Seo everyday contact me on Telegram Telegram Address : @jackleetFor_More_Tools:
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")
| Name | Type | Size | Permission | Actions |
|---|---|---|---|---|
| __pycache__ | Folder | 0755 |
|
|
| api | Folder | 0755 |
|
|
| application | Folder | 0755 |
|
|
| contracts | Folder | 0755 |
|
|
| files | Folder | 0755 |
|
|
| internals | Folder | 0755 |
|
|
| ioc | Folder | 0755 |
|
|
| migrations | Folder | 0755 |
|
|
| model | Folder | 0755 |
|
|
| plugins | Folder | 0755 |
|
|
| simple_rpc | Folder | 0755 |
|
|
| subsys | Folder | 0755 |
|
|
| utils | Folder | 0755 |
|
|
| __init__.py | File | 98 B | 0644 |
|
| __main__.py | File | 33 B | 0644 |
|
| _version.py | File | 83 B | 0644 |
|
| aibolit_job.py | File | 3.85 KB | 0644 |
|
| cli.py | File | 344 B | 0644 |
|
| nats_gateway.py | File | 2.42 KB | 0644 |
|
| rpc_handlers.py | File | 622 B | 0644 |
|
| run.py | File | 5.63 KB | 0644 |
|