Handshake protocols and pooled connections
Logon handshakes and one-call-at-a-time sessions, pooled by the platform.
Some systems do not let you open a connection and start talking. You log on first, the system assigns you a session, and from then on the conversation belongs to that session - one call at a time, in order. Mainframe gateways, trading systems and many older banking protocols all work this way.
A connection like that can never be shared between concurrent calls, so the Connector SDK pools them for you. You write the connector as if there were just one connection - the platform owns the pool, builds connections as they are needed and makes sure each call borrows one for itself.
The connector module
The example wraps a mainframe gateway that expects a logon line first and answers it with a session ID. The connector subclasses PooledConnector instead of Connector - that one change is what puts the platform's pool underneath it.
# -*- coding: utf-8 -*-
# stdlib
import socket
# Zato
from zato.common.sdk import ConnectionLost, Field, PooledConnector
class MainframeConnection:
""" One connection to a mainframe gateway that requires a logon handshake first and then
holds a conversation, one call at a time, so it is never shared between concurrent calls -
the framework pools connections instead.
"""
def __init__(self, host:'str', port:'int', logon_token:'str') -> 'None':
# Connect and keep the socket open for the connection's whole life ..
self.socket = socket.create_connection((host, port))
self.reader = self.socket.makefile('r', encoding='utf8')
# .. log on first - the gateway answers with the session it assigned ..
self._send_line(f'logon {logon_token}')
response = self.reader.readline().strip()
# .. anything other than an ok reply means the logon was rejected.
if not response.startswith('ok '):
raise Exception(f'Logon failed -> {response}')
self.session_id = response[len('ok '):]
def _send_line(self, data:'str') -> 'None':
self.socket.sendall(f'{data}\n'.encode('utf8'))
def send(self, data:'str') -> 'str':
self._send_line(data)
return self.reader.readline().strip()
def close(self) -> 'None':
self.reader.close()
self.socket.close()
class MainframeConnector(PooledConnector):
""" Wraps the mainframe gateway as a connection type that services access through self.out.mainframe.
"""
type = 'mainframe'
# Configuration schema
host = Field.Text()
port = Field.Int(default=9960)
logon_token = Field.Secret()
def create_client(self) -> 'MainframeConnection':
conn = MainframeConnection(self.config.host, self.config.port, self.config.logon_token)
self.logger.info('Mainframe session `%s` logged on for `%s`', conn.session_id, self.name)
return conn
def ping(self, conn:'MainframeConnection') -> 'None':
# A dead socket answers with an empty line - the framework will discard the connection.
response = conn.send('ping')
if not response.endswith('ping'):
raise ConnectionLost(f'The gateway did not answer a ping -> {response!r}')
def on_stop(self, conn:'MainframeConnection') -> 'None':
conn.close()
self.logger.info('Mainframe session `%s` closed for `%s`', conn.session_id, self.name)
def send_command(self, command:'str') -> 'str':
# Each call borrows one connection for its whole duration - the gateway speaks
# one call at a time per session, so the connection is never shared.
with self.get_connection() as conn:
return conn.send(command)
In the code above:
create_clientbuilds one connection, not a shared client - the platform calls it each time the pool needs to grow, up to the pool's sizeget_connectionis how invocation methods talk to the remote end - it borrows a connection from the pool for the duration of thewithblock and no other call can touch it in the meantimeping,on_stopand every other lifecycle method receive one pooled connection, the same waycreate_clientproduced it- The logon handshake lives in the connection's
__init__- by the time a connection enters the pool, it is already logged on
The pool hooks
Two optional hooks run around every borrow. Use them when connections accumulate conversational state that has to be reset between callers:
def on_get_from_pool(self, conn:'MainframeConnection') -> 'None':
# Runs each time a connection is handed out - reset any conversational state here
...
def on_return_to_pool(self, conn:'MainframeConnection') -> 'None':
# Runs each time a with block ends - clean up before others can use the connection
...
Both default to doing nothing, so declare them only if your protocol needs them.
Creating a definition
Definitions are managed with enmasse under a key derived from the connector's type - custom_ plus the type name:
custom_mainframe:
- name: My Mainframe
host: 10.152.81.20
port: 9960
logon_token: my-logon-token
pool_size: 3
pool_size caps how many connections the pool holds. Connections are built lazily - a definition that is never used under load keeps just the one connection its first ping created.
Using it from services
Services do not see the pool at all - they call the connector's methods and each call borrows a connection on its own:
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class SendMainframeCommand(Service):
""" Sends one command to the mainframe gateway over a pooled connection.
"""
name = 'demo.mainframe.send-command'
input = 'command'
def handle(self) -> 'None':
conn = self.out.mainframe['My Mainframe']
response = conn.send_command(self.request.input.command)
self.response.payload = response
When several services - or several invocations of the same service - run concurrently, each of them is served by a distinct connection with its own session, and none of them waits unless the whole pool is busy.