pyzmqbench.py

import time

import zmq
from zmq.eventloop.future import Context, Poller
from zmq.eventloop.ioloop import IOLoop

from tornado import gen

url = 'tcp://127.0.0.1:5555'

ctx = Context()


@gen.coroutine
def ping():
    """print dots to indicate idleness"""
    while True:
        yield gen.sleep(0.50)
        print('.')

async def receiver():
    """receive messages with poll and timeout"""
    print("connecting")
    sock = ctx.socket(zmq.ROUTER)
    sock.connect(url)

    n = 0
    while True:
        frames = await sock.recv_multipart(copy=False)
        assert len(frames) == 3  # envelope + two "real" frames
        n += 1
        if n % 1000 == 0:
            print(n)

async def sender():
    """send a message every second"""
    sock = ctx.socket(zmq.DEALER)
    sock.bind(url)

    frames = [b"foo", b"bar"]

    while True:
        N = 10
        for _ in range(N):
            sock.send_multipart(frames, copy=False)

        await gen.moment  # let other coroutines run


loop = IOLoop.instance()

loop.add_callback(ping)
loop.add_callback(sender)
loop.add_callback(receiver)
loop.start()