Editing File: /opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/internals/delivery_ack.py

"""In-memory delivery acknowledgements for Reportable messages.

The send-to-server plugins queue messages rather than deliver them, so a
producer that keeps its own copy of the payload cannot tell a delivered
message from one a lost send round dropped. Every send path reports the ids
the transport accepted here, and producers register the ids they care about.

Acknowledgements are deliberately not persisted: one that never arrives
leaves the message unconfirmed and makes the producer send it again. That
costs a duplicate the server may well store twice, whereas a silently
dropped payload cannot be recovered at all.
"""

import logging
from typing import Callable, Optional

logger = logging.getLogger(__name__)

class DeliveryAckRegistry:
    def __init__(self) -> None:
        self._callbacks: dict[str, Callable[[], None]] = {}

def watch(self, message_id: str, on_delivered: Callable[[], None]) -> None:
        if not message_id:
            return
        self._callbacks[message_id] = on_delivered

def unwatch(self, message_id: str) -> None:
        self._callbacks.pop(message_id, None)

def confirm(self, message_id: Optional[str]) -> None:
        on_delivered = self._callbacks.pop(message_id, None)
        if on_delivered is None:
            return
        try:
            on_delivered()
        except Exception:
            # a producer's bookkeeping must never break a send round, but it
            # failing means the producer will re-send forever: log the
            # traceback, this is the only place that sees it
            logger.exception(
                "Delivery acknowledgement for %s failed", message_id
            )

registry = DeliveryAckRegistry()

Back to Directory