# Fire-and-forget senders

Send without waiting for replies, with buffered events delivered when the connection stops.

Wrap a sender that expects no reply - an audit trail, a metrics collector, a notification gateway - so that services append events to a client-side buffer and continue without waiting. A background loop delivers the buffer in batches and, when the connection stops because its definition is deleted or edited, the remaining events are flushed before the client closes.

The buffering is ordinary client code. The [Connector SDK](https://zato.io/docs/sdk/index.html) supplies the lifecycle - `on_stop` runs when the connection stops, so the flush-on-stop is one line.

## The connector module {#the-connector-module}

The connector wraps a client for an audit collector. Events accumulate in the client's buffer, a background loop flushes them in batches and `close` - called from `on_stop` - sends the remainder:

```python
# -*- coding: utf-8 -*-

# stdlib
import socket
import threading
import time

# Zato
from zato.common.sdk import Connector, Field

class AuditClient:
    """ A fire-and-forget sender for an audit collector - events are buffered client-side
    and flushed in batches, each batch over its own short-lived connection, and a failed batch
    returns to the buffer for the next flush.
    """
    def __init__(self, host:'str', port:'int', flush_interval:'int') -> 'None':
        self.host = host
        self.port = port
        self.flush_interval = flush_interval

        # Guards the buffer.
        self.lock = threading.Lock()

        # Events waiting for the next flush.
        self.buffer = []

        # Set by close, which stops the flusher loop.
        self.is_stopped = False

        # Flush periodically in the background.
        flusher_thread = threading.Thread(target=self._flush_loop, daemon=True)
        flusher_thread.start()

    def _flush_loop(self) -> 'None':

        while not self.is_stopped:

            # Waiting in small slices makes close take effect quickly even with long intervals.
            waited = 0.0
            while waited < self.flush_interval and not self.is_stopped:
                time.sleep(0.1)
                waited += 0.1

            if self.is_stopped:
                return

            try:
                self.flush()
            except OSError:
                # The collector is down - the events stay in the buffer for the next flush.
                pass

    def add(self, event:'str') -> 'None':
        with self.lock:
            self.buffer.append(event)

    def flush(self) -> 'None':

        # Take the whole buffer under the lock ..
        with self.lock:
            batch = self.buffer[:]
            self.buffer.clear()

        # .. an empty batch means there is nothing to send.
        if not batch:
            return

        # .. and send it over one short-lived connection.
        payload = ''.join(f'{event}\n' for event in batch)

        try:
            with socket.create_connection((self.host, self.port)) as conn:
                conn.sendall(payload.encode('utf8'))
        except OSError:
            # The collector is down - put the batch back for the next flush.
            with self.lock:
                self.buffer[0:0] = batch
            raise

    def close(self) -> 'None':
        """ Stops the flusher and sends the events that remain in the buffer - the flush-on-stop.
        """
        self.is_stopped = True
        self.flush()

class AuditConnector(Connector):
    """ Wraps the audit collector as a connection type that services access through self.out.audit.
    """
    type = 'audit'

    # Configuration schema
    host = Field.Text()
    port = Field.Int(default=9990)
    flush_interval = Field.Int(default=2)

    def create_client(self) -> 'AuditClient':
        return AuditClient(self.config.host, self.config.port, self.config.flush_interval)

    def ping(self, client:'AuditClient') -> 'None':

        # The collector sends no replies, so the check is that the socket opens.
        conn = socket.create_connection((client.host, client.port))
        conn.close()

    def on_stop(self, client:'AuditClient') -> 'None':
        client.close()

    def send_event(self, event:'str') -> 'None':
        self.client.add(event)
```

`send_event` only appends to the buffer, so services return immediately, without waiting for the collector.

Each batch is sent over its own short-lived connection and a failed batch returns to the buffer, so a collector that is down for a while receives the events on a later flush.

`on_stop` runs when the definition is deleted or edited and when the server shuts down. Calling `close` there delivers the buffered remainder.

## Create a definition {#create-a-definition}

Create the definition in an [enmasse](https://zato.io/docs/admin/enmasse.html) file, under the `custom_audit` key - `custom_` followed by the connector's type:

```yaml
custom_audit:
  - name: My Audit
    host: 10.152.81.23
    port: 9990
    flush_interval: 2
```

## Call the connection from a service {#call-the-connection-from-a-service}

The service appends one event to the buffer and returns - the background loop delivers it later:

```python
# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class SendAuditEvent(Service):
    """ Sends one fire-and-forget event to the audit collector.
    """
    name = 'demo.audit.send-event'

    input = 'event'

    def handle(self) -> 'None':
        conn = self.out.audit['My Audit']
        conn.send_event(self.request.input.event)
```

Services append events and continue - the client delivers them in batches, and the flush in `on_stop` delivers the remainder when the connection stops.

## See also {#see-also}

- [SDK reference](https://zato.io/docs/sdk/reference.html) - Field types, lifecycle methods and the behavior of every invocation
- [Server push and subscriptions](https://zato.io/docs/sdk/subscriptions.html) - Traffic in the other direction - messages the remote side pushes
- [Connector SDK tutorial](https://zato.io/docs/sdk/tutorial.html) - Build a complete connector and call it from a service