# Multiplexed requests

Many in-flight requests over one persistent socket, matched back by correlation ID.

Some protocols send many concurrent requests over one persistent connection. Each request is tagged with a correlation ID, replies arrive in any order and a reader loop matches every reply back to the call that waits for it. Payment switches using ISO 8583 and trading systems using FIX work this way.

With the [Connector SDK](https://zato.io/docs/sdk/index.html), the whole mechanism is implemented in the client class you write - the connector declares it like any other client and services call plain methods that share the socket underneath.

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

The connector wraps a payment switch. The client holds one socket, a map of in-flight calls and a reader loop that releases each call when its reply arrives:

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

# stdlib
import itertools
import socket
import threading

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

class PaymentsClient:
    """ A client for a payment switch that multiplexes many in-flight requests over one persistent
    socket - each request has a correlation ID, replies can arrive in any order and a reader
    loop matches them back to the calls that wait for them (ISO 8583, FIX style).
    """
    def __init__(self, host:'str', port:'int') -> 'None':

        # One shared socket for all the requests.
        self.socket = socket.create_connection((host, port))
        self.reader_file = self.socket.makefile('r', encoding='utf8')

        # Guards the pending map and writes to the shared socket.
        self.lock = threading.Lock()

        # Calls in flight, keyed by their correlation IDs.
        self.pending = {}

        # Correlation IDs are consecutive integers.
        self.counter = itertools.count(1)

        # Set to False by the reader loop once the socket closes.
        self.is_connected = True

        # The reader loop matches replies back to waiting calls for as long as the connection exists.
        reader_thread = threading.Thread(target=self._read_loop, daemon=True)
        reader_thread.start()

    def _read_loop(self) -> 'None':

        for line in self.reader_file:
            text = line.strip()
            corr_id, _, payload = text.partition(' ')

            # The call this reply belongs to may have timed out and left the map already.
            with self.lock:
                holder = self.pending.pop(corr_id, None)

            if holder:
                holder['response'] = payload
                holder['event'].set()

        # The loop ended because the socket closed - release every call that still waits.
        self.is_connected = False

        with self.lock:
            for holder in self.pending.values():
                holder['event'].set()
            self.pending.clear()

    def request(self, payload:'str') -> 'str':

        # The socket is closed - raising ConnectionLost makes the platform reconnect.
        if not self.is_connected:
            raise ConnectionLost('The payment switch connection is down')

        corr_id = str(next(self.counter))
        holder = {'event': threading.Event(), 'response': None}

        # Register the call and send its request under one lock, so the reply cannot
        # arrive before the call is registered.
        with self.lock:
            self.pending[corr_id] = holder
            self.socket.sendall(f'{corr_id} {payload}\n'.encode('utf8'))

        # Wait for the reader loop to match the reply back to this call.
        _ = holder['event'].wait()

        # An event set without a response means the connection closed while this call was in flight.
        if holder['response'] is None:
            raise ConnectionLost('The payment switch connection went down mid-call')

        return holder['response']

    def close(self) -> 'None':
        self.reader_file.close()
        self.socket.close()

class PaymentsConnector(Connector):
    """ Wraps the payment switch as a connection type that services access through self.out.payments.
    """
    type = 'payments'

    # Configuration schema
    host = Field.Text()
    port = Field.Int(default=9970)

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

    def ping(self, client:'PaymentsClient') -> 'None':
        client.request('ping')

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

    def authorize(self, payload:'str') -> 'str':
        return self.client.request(payload)
```

One client serves all the services on a server. Their concurrent requests are in flight together, over the one socket.

`ConnectionLost` makes the platform evict the client and reconnect. Raise it when the client discovers the socket is closed - the next invocation receives a freshly built client.

When the socket closes, the reader loop releases every waiting call, so each caller receives `ConnectionLost` instead of blocking on a connection that no longer exists.

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

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

```yaml
custom_payments:
  - name: My Payments
    host: 10.152.81.21
    port: 9970
```

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

The service sends one request and reads its reply - the multiplexing stays inside the client:

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

# Zato
from zato.server.service import Service

class AuthorizePayment(Service):
    """ Authorizes one payment over the multiplexed connection to the payment switch.
    """
    name = 'demo.payments.authorize'

    input = 'payload'

    def handle(self) -> 'None':
        conn = self.out.payments['My Payments']
        response = conn.authorize(self.request.input.payload)
        self.response.payload = response
```

Concurrent invocations of this service share the one socket and each still receives its own reply, even when the switch answers out of order.

## See also {#see-also}

- [SDK reference](https://zato.io/docs/sdk/reference.html) - Field types, lifecycle methods and the behavior of every invocation
- [Handshake protocols and pooled connections](https://zato.io/docs/sdk/stateful-sockets.html) - Sessions that serve one call at a time, pooled by the platform
- [Connector SDK tutorial](https://zato.io/docs/sdk/tutorial.html) - Build a complete connector and call it from a service