How to use zmq - 10 common examples

else:
            tasks, count = tasks_to_process[name]
            tasks.append(task)
            tasks_to_process[name] = (tasks, count + point_count)

    processed_points = 0
    points_in_progress = 0
    previous_percent = 0
    points_in_pnts = 0

    max_splitting_jobs_count = max(1, jobs // 2)

    # zmq setup
    context = zmq.Context()

    zmq_skt = context.socket(zmq.ROUTER)
    zmq_skt.bind('ipc:///tmp/py3dtiles1')

    zmq_idle_clients = []

    state = State(infos['portions'])

    zmq_processes_killed = -1

    zmq_processes = [multiprocessing.Process(
        target=zmq_process,
        args=(
            graph, projection, node_store, octree_metadata, outfolder, rgb, verbose)) for i in range(jobs)]

    for p in zmq_processes:
        p.start()
    activities = [p.pid for p in zmq_processes]

def test_pubsub(self):
        pub, sub = self.create_bound_pair(zmq.PUB, zmq.SUB)
        sub.setsockopt(zmq.SUBSCRIBE,b'')
        wait()

        loop = pyuv.Loop.default_loop()
        s = ZMQ(loop, sub)
        s1 = ZMQ(loop, pub)

        r = []
        def cb(stream, msg, err):
            r.append(msg[0])
            s.stop()
            s1.stop()

        s.start_read(cb)
        s1.write(b"message")

def stdinHandler(publisher):
    line = sys.stdin.readline()
    if not line:
        # EOF... end program
        ioloop.IOLoop.instance().stop()
        return
    line = line[:-1]
    topic, body = line.split(':', 1)
    logging.debug('publishing: %s:%s', topic, body)
    publisher.sendRaw(topic, body)

import simplejson as json
import zmq

ctx = zmq.Context()
subscriber = ctx.socket(zmq.SUB)
subscriber.linger = 0
subscriber.setsockopt(zmq.SUBSCRIBE, b'')
subscriber.connect("tcp://localhost:5557")

while True:
    sequence = int(subscriber.recv())
    entry = json.loads(subscriber.recv_string())
    print('{}: {}'.format(sequence, entry))

def run(self):
        context = zmq.Context()
        push_socket = context.socket(zmq.PUSH)
        pull_socket = context.socket(zmq.PULL)
        identity = u'worker-%d' % self.id
        push_socket.identity = identity.encode('ascii')
        push_socket.connect('tcp://localhost:5555')
        pull_socket.identity = identity.encode('ascii')
        pull_socket.connect('tcp://localhost:5556')
        print('Client %s started' % (identity))
        poll = zmq.Poller()
        poll.register(pull_socket, zmq.POLLIN)
        reqs = 0
        while True:
            reqs = reqs + 1
            if reqs % 100 == 0:
                print('Req #%d sent..' % (reqs))
            _, frame = self.stream.read()
            md = dict(
                dtype = str(frame.dtype),
                shape = frame.shape,
            )
            push_socket.send_json(md, zmq.SNDMORE)
            push_socket.send(memoryview(frame.data), 0, copy=True, track=False)

            sockets = dict(poll.poll(10))
            if pull_socket in sockets:

# Set up a debug socket, if address is given.

            debug_pub = ctx.socket(zmq.PUB)
            debug_pub.bind("tcp://127.0.0.1:%s" % debug_port)

            zmq.proxy(xpub, xsub, debug_pub)

        def send_message(pub, msg):
            while True:
                pub.publish(msg)
                time.sleep(.1)

        t = Thread(target=run_proxy, args=(ctx,), daemon=True)
        t.start()
        time.sleep(1)
        test_sub = ctx.socket(zmq.SUB)
        test_sub.connect("tcp://127.0.0.1:%s" % sub_port)
        test_sub.setsockopt(zmq.SUBSCRIBE, b"")

        publisher = Publisher("tcp://127.0.0.1:%s" % port, 'TestPub', ctx=ctx)
        publisher.start()
        name, topic, data = 'TestNode', 'testing', ['Raw', 'this', 'is', 'data']
        msg = Envelope(topic, name, data)
        sender_t = Thread(target=send_message, args=(publisher, msg), daemon=True)
        sender_t.start()

        i = 0
        while i < 10:
            try:
                frames = test_sub.recv_multipart(zmq.NOBLOCK)
            except zmq.error.Again:
                frames = []

import zmq
from random import randrange
import time
import numpy as np

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:5556")

context2 = zmq.Context()
socket_sessions = context2.socket(zmq.PUB)
socket_sessions.bind("tcp://*:5557")


def send_array(socket, A, session="buq123", flags=0, copy=False, track=False):
    """send a numpy array with metadata"""
    md = dict(
        dtype = str(A.dtype),
        shape = A.shape,
    )
    socket.send_string(f"{session} Plot{np.random.randint(3)} 111", flags|zmq.SNDMORE) # Session name, plot name, subplot number
    socket.send_json(md, flags|zmq.SNDMORE) # Array metadata
    return socket.send(A, flags, copy=copy, track=track) # Array data

def setUp(self):

        # create the io_loop
        self._io_loop = IOLoop.instance()

        # and the context
        self._ctx = zmq.Context(1)

        # setup the mgmt sockets
        self._setup_mgmt_sockets()

        # setup the data sockets
        self._setup_data_sockets()

        # setup the management interface
        self._mgmt = ZmqMgmt( self._mgmt_sockets['worker_sub'],
            self._mgmt_sockets['worker_pub'], io_loop=self._io_loop)
        self._mgmt.start()
        self._mgmt.add_callback(ZMQ_SPYDER_MGMT_WORKER, self.on_mgmt_end)

def broker_should_be_quiet(self):
        sleep(time_to_wait / 1000.0)
        with self.assertRaises(zmq.ZMQError):
            self.api.recv_json(zmq.NOBLOCK)

def setUp(self):

        # create the io_loop
        self._io_loop = IOLoop.instance()

        # and the context
        self._ctx = zmq.Context(1)

        # setup the mgmt sockets
        self._setup_mgmt_sockets()

        # setup the data sockets
        self._setup_data_sockets()

        # setup the management interface
        self._mgmt = ZmqMgmt( self._mgmt_sockets['worker_sub'],
            self._mgmt_sockets['worker_pub'], io_loop=self._io_loop)
        self._mgmt.start()
        self._mgmt.add_callback(ZMQ_SPYDER_MGMT_WORKER, self.on_mgmt_end)