ZMQ Tubes

ZMQ Tubes is a managing system for ZMQ communication. It can manage many ZMQ sockets by one interface. The whole system is hierarchical, based on topics (look at MQTT topics).

Classes

• TubeMessage - This class represents a request/response message. Some types of tubes require a response in this format.
• Tube - This class wraps a ZMQ socket. It represents a connection between client and server.
• TubeMonitor - The class can sniff of the ZMQTube communication. (Supported only with Asyncio version).
• TubeNode - This represents an application interface for communication via tubes.

Asyncio / Threading

The library support bot method. Asyncio from Python 3.9.

from zmq_tubes import TubeNode, Tube            # Asyncio classes
from zmq_tubes.threads import TubeNode, Tube    # Threads classes

Usage:

Node definitions in yml file

We can define all tubes for one TubeNode by yml file. Next examples require install these packages PyYAML, pyzmq and zmq_tubes.

Client service (asyncio example)

# client.yml
tubes:
  - name: Client REQ
    addr:  ipc:///tmp/req.pipe
    tube_type: REQ
    topics:
      - foo/bar

  - name: Client PUB
    addr:  ipc:///tmp/pub.pipe
    tube_type: PUB
    topics:
      - foo/pub/#

# client.py
import asyncio
import yaml
from zmq_tubes import TubeNode, TubeMessage


async def run():
  with open('client.yml', 'r+') as fd:
    schema = yaml.safe_load(fd)
  node = TubeNode(schema=schema)
  async with node:
      print(await node.request('foo/bar', 'message 1'))
      await node.publish('foo/pub/test', 'message 2')

if __name__ == "__main__":
    asyncio.run(run())

> python client.py
topic: foo/bar,  payload: response

Server service (threads example)

# server.yml
tubes:
  - name: server ROUTER
    addr:  ipc:///tmp/req.pipe
    tube_type: ROUTER
    server: True
    topics:
      - foo/bar

  - name: server SUB
    addr:  ipc:///tmp/pub.pipe
    tube_type: SUB
    server: True
    topics:
      - foo/pub/#

# server.py
import yaml
from zmq_tubes.threads import TubeNode, TubeMessage


def handler(request: TubeMessage):
  print(request.payload)
  if request.tube.tube_type_name == 'ROUTER':
    return request.create_response('response')


def run():
  with open('server.yml', 'r+') as fd:
    schema = yaml.safe_load(fd)
  node = TubeNode(schema=schema)
  node.register_handler('foo/#', handler)
  with node:
    node.start().join()

if __name__ == "__main__":
    run()

> python server.py
message 1
message 2

YAML definition

The yaml file starts with a root element tubes, which contains list of all our tube definitions.

• name - string - name of the tube.
• addr - string - connection or bind address in format transport://address (see more http://api.zeromq.org/2-1:zmq-connect)
• server - bool - is this tube server side (bind to addr) or client side (connect to addr)
• tube_type - string - type of this tube (see more https://zguide.zeromq.org/docs/chapter2/#Messaging-Patterns)
• identity - string - (optional) we can setup custom tube identity
• utf8_decoding - bool - (default = True), if this is True, the payload is automatically UTF8 decode.
• sockopts - dict - (optional) we can setup sockopts for this tube (see more http://api.zeromq.org/4-2:zmq-setsockopt)
• monitor - string - (optional) bind address of tube monitor (see more Debugging / Monitoring)

Request / Response

Deferred responses with threads (ROUTER)

Since version 1.17.0, a threaded ROUTER handler may return a concurrent.futures.Future instead of an immediate TubeMessage. This releases the transport worker while application code waits for an external event. The future does not start a thread and no asyncio event loop is required.

from concurrent.futures import Future


def handler(request):
    response = Future()
    # Register with an application event source. The callback may run in
    # another thread, or immediately if the event has already happened.
    def finished(value):
        if response.set_running_or_notify_cancel():
            response.set_result(request.create_response(value))

    register_completion_callback(finished)
    return response

register_completion_callback represents the application's event subscription; it is not a library API. It must deliver completion at most once. Do not wait on the future inside the request handler or submit a blocking wait back to the transport's worker pool.

The result must be a TubeMessage created from the original request. Completion only queues that response; socket writes still run in the node's main thread. Already-completed futures and out-of-order responses are supported. Cancelled futures do not send a response. Exceptions or invalid results are logged and do not send a response; the client's existing request timeout still applies.

Stopping or closing the node cancels pending futures where possible and ignores late completions. Applications should remove their event subscriptions when a future is cancelled. Deferred responses are not supported by threaded REP handlers or the asyncio API. Existing synchronous handlers retain their API.

This is a simple scenario, the server processes the requests serially.

Server:

from zmq_tubes import Tube, TubeNode, TubeMessage


async def handler(request: TubeMessage):
  print(request.payload)
  return 'answer'
  # or return request.create_response('response')


tube = Tube(
  name='Server',
  addr='ipc:///tmp/req_resp.pipe',
  server=True,
  tube_type='REP'
)

node = TubeNode()
node.register_tube(tube, 'test/#')
node.register_handler('test/#', handler)
await node.start()

# output: 'question'

Client:

from zmq_tubes import Tube, TubeNode

tube = Tube(
  name='Client',
  addr='ipc:///tmp/req_resp.pipe',
  tube_type='REQ'
)

node = TubeNode()
node.register_tube(tube, 'test/#')
response = await node.request('test/xxx', 'question')
print(response.payload)
# output: 'answer'

The method request accepts the optional parameter utf8_decoding. When we set this parameter to False in previous example, the returned payload is not automatically decoded, we get bytes.

Subscribe / Publisher

Server:

from zmq_tubes import Tube, TubeNode, TubeMessage


async def handler(request: TubeMessage):
  print(request.payload)


tube = Tube(
  name='Server',
  addr='ipc:///tmp/sub_pub.pipe',
  server=True,
  tube_type='SUB'
)

node = TubeNode()
node.register_tube(tube, 'test/#')
node.register_handler('test/#', handler)
await node.start()
# output: 'message'

Client:

from zmq_tubes import Tube, TubeNode

tube = Tube(
  name='Client',
  addr='ipc:///tmp/sub_pub.pipe',
  tube_type='PUB'
)
# In the case of publishing, the first message is very often
# lost. The workaround is to connect the tube manually as soon as possible.
tube.connect()

node = TubeNode()
node.register_tube(tube, 'test/#')
node.publish('test/xxx', 'message')

Request / Router

The server is asynchronous. It means it is able to process more requests at the same time.

Server:

import asyncio
from zmq_tubes import Tube, TubeNode, TubeMessage


async def handler(request: TubeMessage):
  print(request.payload)
  if request.payload == 'wait':
    await asyncio.sleep(10)
  return request.create_response(request.payload)


tube = Tube(
  name='Server',
  addr='ipc:///tmp/req_router.pipe',
  server=True,
  tube_type='ROUTER'
)

node = TubeNode()
node.register_tube(tube, 'test/#')
node.register_handler('test/#', handler)
await node.start()
# output: 'wait'
# output: 'message'

Client:

import asyncio
from zmq_tubes import Tube, TubeNode

tube = Tube(
  name='Client',
  addr='ipc:///tmp/req_router.pipe',
  tube_type='REQ'
)


async def task(node, text):
  print(await node.request('test/xxx', text))


node = TubeNode()
node.register_tube(tube, 'test/#')
asyncio.create_task(task(node, 'wait'))
asyncio.create_task(task(node, 'message'))
# output: 'message'
# output: 'wait'

Dealer / Response

The client is asynchronous. It means it is able to send more requests at the same time.

Server:

from zmq_tubes import Tube, TubeNode, TubeMessage


async def handler(request: TubeMessage):
  print(request.payload)
  return 'response'
  # or return requset.create_response('response')


tube = Tube(
  name='Server',
  addr='ipc:///tmp/dealer_resp.pipe',
  server=True,
  tube_type='REP'
)

node = TubeNode()
node.register_tube(tube, 'test/#')
node.register_handler('test/#', handler)
await node.start()
# output: 'message'

Client:

from zmq_tubes import Tube, TubeNode, TubeMessage

tube = Tube(
  name='Client',
  addr='ipc:///tmp/dealer_resp.pipe',
  tube_type='DEALER'
)


async def handler(response: TubeMessage):
  print(response.payload)


node = TubeNode()
node.register_tube(tube, 'test/#')
node.register_handler('test/#', handler)

await node.send('test/xxx', 'message')

# output: 'response'

Dealer / Router

The client and server are asynchronous. It means it is able to send and process more requests/responses at the same time.

Server:

import asyncio
from zmq_tubes import Tube, TubeNode, TubeMessage


async def handler(request: TubeMessage):
  print(request.payload)
  if request.payload == 'wait':
    await asyncio.sleep(10)
  return request.create_response(request.payload)


tube = Tube(
  name='Server',
  addr='ipc:///tmp/dealer_router.pipe',
  server=True,
  tube_type='ROUTER'
)

node = TubeNode()
node.register_tube(tube, 'test/#')
node.register_handler('test/#', handler)
await node.start()
# output: 'wait'
# output: 'message'

Client:

from zmq_tubes import Tube, TubeNode, TubeMessage

tube = Tube(
  name='Client',
  addr='ipc:///tmp/dealer_router.pipe',
  tube_type='DEALER'
)


async def handler(response: TubeMessage):
  print(response.payload)


node = TubeNode()
node.register_tube(tube, 'test/#')
node.register_handler('test/#', handler)

await node.send('test/xxx', 'wait')
await node.send('test/xxx', 'message')

# output: 'message'
# output: 'wait'

Dealer / Dealer

The client and server are asynchronous. It means it is able to send and process more requests/responses at the same time.

Server:

from zmq_tubes import Tube, TubeNode, TubeMessage

tube = Tube(
  name='Server',
  addr='ipc:///tmp/dealer_dealer.pipe',
  server=True,
  tube_type='DEALER'
)


async def handler(response: TubeMessage):
  print(response.payload)


node = TubeNode()
node.register_tube(tube, 'test/#')
node.register_handler('test/#', handler)

await node.send('test/xxx', 'message from server')
# output: 'message from client'

Client:

from zmq_tubes import Tube, TubeNode, TubeMessage

tube = Tube(
  name='Client',
  addr='ipc:///tmp/dealer_dealer.pipe',
  tube_type='DEALER'
)


async def handler(response: TubeMessage):
  print(response.payload)


node = TubeNode()
node.register_tube(tube, 'test/#')
node.register_handler('test/#', handler)

await node.send('test/xxx', 'message from client')
# output: 'message from server'

Debugging / Monitoring

We can assign a monitor socket to our zmq tubes. By this monitor socket, we can sniff zmq communication or get a zmq tube configuration.

tubes:
  - name: ServerRouter
    addr:  ipc:///tmp/router.pipe
    monitor: ipc:///tmp/test.monitor
    tube_type: ROUTER
    server: yes
    topics:
      - foo/#

This is example of a yaml definition. We can use the same monitor socket for more tubes in the same tubeNode. When we add the monitor attribute to our tube definition, the application automatically create a new socket monitor: /tmp/test.monitor. Your application works as a server side. The logs are sent to the socket only for the time, when the monitoring tool is running.

Monitoring tool

After enabling of the monitoring in the application, we can use the monitoring tool for sniff.

# get the server tube configuration
> zmqtube-monitor get_schema ipc:///tmp/display.monitor
    tubes:
      - addr: ipc:///tmp/router.pipe
        monitor: ipc:///tmp/test.monitor
        name: ServerRouter
        server: 'yes'
        tube_type: ROUTER

# the log tube communication. Logs will be saved to dump.rec as well.
> zmqtube-monitor logs -d ./dump.rec ipc:///tmp/display.monitor
 0.28026580810546875 ServerRouter < foo/test Request
 0.0901789665222168 ServerRouter > foo/test Response

# The format of output
# <relative time> <tube name> <direction> <topic> <message>`

Simulation of the client side

When we have a dump file (e.g. dump.rec), we can simulate the communication with our app. The first step is prepare the mock client schema file. For this, We can get the tube node configuration from our application and after that edit it.

> zmqtube-monitor get_schema ipc:///tmp/display.monitor > mock_schema.yaml
> vim mock_schema.yaml
...
# Now, we have to update the file mock_schema.yaml.
# We change configuration to the mock client configuration.
# The names of the tubes must be the same as are in your app.
# We can remove monitoring attribute and change server and
# tube_type attributes. In this mock file, the topics are not
# required, they are ignored.

> cat mock_schema.yaml
tubes:
- addr: ipc:///tmp/router.pipe
  name: ServerRouter
  tube_type: REQ

Now, we can start the simulation of the client communication.

> zmqtube-monitor simulate mock_schema.yaml dump.rec

If the response of our app is not the same as tool expects (the response saved in dump file), then the monitoring tool warns us. We can modify speed of the simulation by the parameter --speed.

In the default configuration, is the simulation run the same speed as original communication (parameter --speed=1).

Example of programming declaration of the monitoring.

import zmq
from zmq_tubes.threads import Tube, TubeNode, TubeMessage, TubeMonitor


def handler(request: TubeMessage):
  print(request.payload)
  return request.create_response('response')

resp_tube = Tube(
  name='REP',
  addr='ipc:///tmp/rep.pipe',
  server='yes',
  tube_type=zmq.REP
)

req_tube = Tube(
  name='REQ',
  addr='ipc:///tmp/rep.pipe',
  tube_type=zmq.REQ
)

node = TubeNode()
node.register_tube(resp_tube, f"foo/#")
node.register_tube(req_tube, f"foo/#")
node.register_handler(f"foo/#", handler)

node.register_monitor(resp_tube, TubeMonitor(addr='ipc:///tmp/test.monitor'))

with node:
    print(node.request('foo/xxx', 'message 2'))