python类PULL的实例源码

def __init__(self, opts=None):
        if opts is None:
            self.opts = self.process_config(CONFIG_LOCATION)
        else:
            self.opts = opts

        self.ctx = zmq.Context()
        self.pub_socket = self.ctx.socket(zmq.PUB)

        self.pub_socket.bind('tcp://127.0.0.1:2000')

        self.loop = zmq.eventloop.IOLoop.instance()
        self.pub_stream = zmq.eventloop.zmqstream.ZMQStream(self.pub_socket, self.loop)

        # Now create PULL socket over IPC to listen to reactor

        self.pull_socket = self.ctx.socket(zmq.PULL)
        self.pull_socket.bind('ipc:///tmp/reactor.ipc')
        self.pull_stream = zmq.eventloop.zmqstream.ZMQStream(self.pull_socket, self.loop)

        self.pull_stream.on_recv(self.republish)

def receive_message(self, event, event_data, listener_data):
        """
        Receives a messages from another processes.

        :param * event: Not used.
        :param * event_data: Not used.
        :param * listener_data: Not used.
        """
        del event, event_data, listener_data

        # Make a poller for all incoming sockets.
        poller = zmq.Poller()
        for socket in self.__end_points.values():
            if socket.type in [zmq.PULL, zmq.REP]:
                poller.register(socket, zmq.POLLIN)

        # Wait for socket is ready for reading.
        socks = dict(poller.poll())

        for name, socket in self.__end_points.items():
            if socket in socks:
                self._receive_message(name, socket)

    # ------------------------------------------------------------------------------------------------------------------

def __register_sockets(self):
        """
        Registers ZMQ sockets for communication with other processes in Enarksh.
        """
        config = Config.get()

        # Register socket for receiving asynchronous incoming messages.
        self.__message_controller.register_end_point('pull', zmq.PULL, config.get_spawner_pull_end_point())

        # Register socket for sending asynchronous messages to the controller.
        self.__message_controller.register_end_point('controller', zmq.PUSH, config.get_controller_pull_end_point())

        # Register socket for sending asynchronous messages to the logger.
        self.__message_controller.register_end_point('logger', zmq.PUSH, config.get_logger_pull_end_point())

    # ------------------------------------------------------------------------------------------------------------------

def __register_sockets(self):
        """
        Registers ZMQ sockets for communication with other processes in Enarksh.
        """
        config = Config.get()

        # Register socket for receiving asynchronous incoming messages.
        self.message_controller.register_end_point('pull', zmq.PULL, config.get_controller_pull_end_point())

        # Create socket for lockstep incoming messages.
        self.message_controller.register_end_point('lockstep', zmq.REP, config.get_controller_lockstep_end_point())

        # Create socket for sending asynchronous messages to the spanner.
        self.message_controller.register_end_point('spawner', zmq.PUSH, config.get_spawner_pull_end_point())

        # Create socket for sending asynchronous messages to the logger.
        self.message_controller.register_end_point('logger', zmq.PUSH, config.get_logger_pull_end_point())

    # ------------------------------------------------------------------------------------------------------------------

def register_end_point(self, name, socket_type, end_point):
        """
        Registers an end point.

        :param str name: The name of the end point.
        :param int socket_type: The socket type, one of
                                - zmq.PULL for asynchronous incoming messages
                                - zmq.REP for lockstep incoming messages
                                - zmq.PUSH for asynchronous outgoing messages
        :param str end_point: The end point.
        """
        socket = self.__zmq_context.socket(socket_type)
        self.__end_points[name] = socket
        if socket_type in [zmq.PULL, zmq.REP]:
            socket.bind(end_point)
        elif socket_type == zmq.PUSH:
            socket.connect(end_point)
        else:
            raise ValueError("Unknown socket type {0}".format(socket_type))

    # ------------------------------------------------------------------------------------------------------------------

def test_shadow_pyczmq(self):
        try:
            from pyczmq import zctx, zsocket
        except Exception:
            raise SkipTest("Requires pyczmq")

        ctx = zctx.new()
        ca = zsocket.new(ctx, zmq.PUSH)
        cb = zsocket.new(ctx, zmq.PULL)
        a = zmq.Socket.shadow(ca)
        b = zmq.Socket.shadow(cb)
        a.bind("inproc://a")
        b.connect("inproc://a")
        a.send(b'hi')
        rcvd = self.recv(b)
        self.assertEqual(rcvd, b'hi')

    # Travis can't handle how much memory PyPy uses on this test

def test_cyclic_destroy(self):
        """ctx.destroy should succeed when cyclic ref prevents gc"""
        # test credit @dln (GH #137):
        class CyclicReference(object):
            def __init__(self, parent=None):
                self.parent = parent

            def crash(self, sock):
                self.sock = sock
                self.child = CyclicReference(self)

        def crash_zmq():
            ctx = self.Context()
            sock = ctx.socket(zmq.PULL)
            c = CyclicReference()
            c.crash(sock)
            ctx.destroy()

        crash_zmq()

def test_poll(self):
        @gen.coroutine
        def test():
            a, b = self.create_bound_pair(zmq.PUSH, zmq.PULL)
            f = b.poll(timeout=0)
            self.assertEqual(f.result(), 0)

            f = b.poll(timeout=1)
            assert not f.done()
            evt = yield f
            self.assertEqual(evt, 0)

            f = b.poll(timeout=1000)
            assert not f.done()
            yield a.send_multipart([b'hi', b'there'])
            evt = yield f
            self.assertEqual(evt, zmq.POLLIN)
            recvd = yield b.recv_multipart()
            self.assertEqual(recvd, [b'hi', b'there'])
        self.loop.run_sync(test)

def test_poll(self):
        @asyncio.coroutine
        def test():
            a, b = self.create_bound_pair(zmq.PUSH, zmq.PULL)
            f = b.poll(timeout=0)
            yield from asyncio.sleep(0)
            self.assertEqual(f.result(), 0)

            f = b.poll(timeout=1)
            assert not f.done()
            evt = yield from f

            self.assertEqual(evt, 0)

            f = b.poll(timeout=1000)
            assert not f.done()
            yield from a.send_multipart([b'hi', b'there'])
            evt = yield from f
            self.assertEqual(evt, zmq.POLLIN)
            recvd = yield from b.recv_multipart()
            self.assertEqual(recvd, [b'hi', b'there'])
        self.loop.run_until_complete(test())

def _process_single_event(self, socket):
        """
        Process a socket's event.

        Parameters
        ----------
        socket : zmq.Socket
            Socket that generated the event.
        """
        data = socket.recv()
        address = self.address[socket]
        if address.kind == 'SUB':
            self._process_sub_event(socket, address, data)
        elif address.kind == 'PULL':
            self._process_pull_event(socket, address, data)
        elif address.kind == 'REP':
            self._process_rep_event(socket, address, data)
        else:
            self._process_single_event_complex(address, socket, data)

def _process_pull_event(self, socket, addr, data):
        """
        Process a PULL socket's event.

        Parameters
        ----------
        socket : zmq.Socket
            Socket that generated the event.
        addr : AgentAddress
            AgentAddress associated with the socket that generated the event.
        data : bytes
            Data received on the socket.
        """
        message = deserialize_message(message=data, serializer=addr.serializer)
        handler = self.handler[socket]
        if not isinstance(handler, (list, dict, tuple)):
            handler = [handler]
        for h in handler:
            h(self, message)

def test_agentchannel_async_rep():
    """
    Test basic ASYNC_REP AgentChannel operations: initialization, equivalence
    and basic methods.
    """
    receiver = AgentAddress('ipc', 'addr0', 'PULL', 'server', 'pickle')
    channel = AgentChannel('ASYNC_REP', receiver=receiver, sender=None)
    # Equivalence
    assert channel == AgentChannel('ASYNC_REP', receiver=receiver, sender=None)
    assert not channel == 'foo'
    assert channel != 'foo'
    # Basic methods
    assert channel.twin() == AgentChannel('ASYNC_REQ', sender=receiver.twin(),
                                          receiver=None)
    # Other attributes
    assert hasattr(channel, 'uuid')
    assert channel.transport == 'ipc'
    assert channel.serializer == 'pickle'

def test_agentchannel_sync_pub():
    """
    Test basic SYNC_PUB AgentChannel operations: initialization, equivalence
    and basic methods.
    """
    sender = AgentAddress('ipc', 'addr0', 'PUB', 'server', 'pickle')
    receiver = AgentAddress('ipc', 'addr0', 'PULL', 'server', 'pickle')
    channel = AgentChannel('SYNC_PUB', sender=sender, receiver=receiver)
    # Equivalence
    assert channel == AgentChannel('SYNC_PUB', sender=sender,
                                   receiver=receiver)
    assert not channel == 'foo'
    assert channel != 'foo'
    # Basic methods
    assert channel.twin() == AgentChannel('SYNC_SUB', sender=receiver.twin(),
                                          receiver=sender.twin())
    # Other attributes
    assert hasattr(channel, 'uuid')
    assert channel.transport == 'ipc'
    assert channel.serializer == 'pickle'

def connect(self):

        self.context = zmq.Context()

        if not self.context:
            raise RuntimeError('Failed to create ZMQ context!')

        self.socket = self.context.socket(zmq.PULL)

        if not self.socket:
            raise RuntimeError('Failed to create ZMQ socket!')

        self.socket.bind(self.endpoint)

        self.poller = zmq.Poller()
        self.poller.register(self.socket, zmq.POLLIN)

        self.is_connected = True

def router_main(_, pidx, args):
    log = get_logger('examples.zmqserver.extra', pidx)
    ctx = zmq.Context()
    ctx.linger = 0
    in_sock = ctx.socket(zmq.PULL)
    in_sock.bind('tcp://*:5000')
    out_sock = ctx.socket(zmq.PUSH)
    out_sock.bind('ipc://example-events')
    try:
        log.info('router proxy started')
        zmq.proxy(in_sock, out_sock)
    except KeyboardInterrupt:
        pass
    except:
        log.exception('unexpected error')
    finally:
        log.info('router proxy terminated')
        in_sock.close()
        out_sock.close()
        ctx.term()

def worker_main(loop, pidx, args):
    log = get_logger('examples.zmqserver.worker', pidx)
    router = await aiozmq.create_zmq_stream(
        zmq.PULL,
        connect='ipc://example-events')

    async def process_incoming(router):
        while True:
            try:
                data = await router.read()
            except aiozmq.ZmqStreamClosed:
                break
            log.info(data)

    task = loop.create_task(process_incoming(router))
    log.info('started')

    yield

    router.close()
    await task
    log.info('terminated')

def reset(self):
        self.status = READY

        context = zmq.Context()
        self._socket1 = context.socket(zmq.PUSH)
        self._socket1.bind(self._address1)
        self._socket1.set_hwm(32)
        self._socket2 = context.socket(zmq.PULL)
        self._socket2.set_hwm(32)
        self._socket2.RCVTIMEO = 1
        self._socket2.bind(self._address2)
        self._prev_drained = False
        self._sub_drained = False
        self._conn1_send_count = 0
        self._conn1_recv_count = {}
        self._conn2_send_count = {}
        self._conn2_recv_count = 0
        self._retry_count = 0

def reset(self):
        self.status = READY

        context = zmq.Context()
        self._socket = context.socket(zmq.PULL)
        self._socket.RCVTIMEO = 1
        sync_socket = context.socket(zmq.PUSH)
        while self._ports['conn1'] is None or self._ports['sync_conn1'] is None:
            sleep(0.01)

        # Handshake with main process
        self._socket.connect(self._address + ':' + str(self._ports['conn1']))
        sync_socket.connect(self._address + ':' + str(self._ports['sync_conn1']))
        packet = msgpack.dumps(b'SYNC')
        sync_socket.send(packet)
        sync_socket.close()

        self._num_recv = 0
        self._drained = False

def zmq_streamer():
    try:
        context = zmq.Context()
        # Socket facing clients
        frontend = context.socket(zmq.PUSH)
        frontend.bind("tcp://*:%s" % (zmq_queue_port_push))
        # Socket facing services
        backend = context.socket(zmq.PULL)
        backend.bind("tcp://*:%s" % (zmq_queue_port_pull))

        zmq.device(zmq.STREAMER, frontend, backend)
    except Exception as e:
        print(e)
        print("bringing down zmq device")
    finally:
        frontend.close()
        backend.close()
        context.term()

def generator_from_zmq_pull(context, host):
    socket = context.socket(zmq.PULL)
    # TODO: Configure socket with clean properties to avoid message overload.
    if host.endswith('/'):
        host = host[:-1]
    print_item("+", "Binding ZMQ pull socket : " + colorama.Fore.CYAN + "{0}".format(host) + colorama.Style.RESET_ALL)
    socket.bind(host)

    while True:
        try:
            message = socket.recv(flags=zmq.NOBLOCK)
        except zmq.Again as e:
            message = None
        if message is None:
            yield None # NOTE: We have to make the generator non blocking.
        else:
            task = json.loads(message)
            yield task

def __init__(self, push, pull, redis_conf):
        super(MinerClient, self).__init__()

        print("Connecting to Redis cache {} ...".format(redis_conf))
        redis_host, redis_port, redis_db = redis_conf.split(":")
        self.redis = redis.StrictRedis(host=redis_host, port=int(redis_port), db=int(redis_db))
        self.redis.setnx('transaction', 0)
        # NOTE: Expiration times for pending/processed tasks in seconds.
        self.transaction_expiration = 60 * 60
        self.result_expiration = 60 * 10

        context = zmq.Context()

        print("Connecting to push socket '{}' ...".format(push))
        self.push = context.socket(zmq.PUSH)
        self.push.connect(push)

        print("Binding to pull socket '{}' ...".format(pull))
        self.pull = context.socket(zmq.PULL)
        self.pull.bind(pull)

def test_tcp_push_socket(event_loop, socket_factory, connect_or_bind):
    pull_socket = socket_factory.create(zmq.PULL)
    connect_or_bind(pull_socket, 'tcp://127.0.0.1:3333', reverse=True)

    def run():
        assert pull_socket.poll(1000) == zmq.POLLIN
        message = pull_socket.recv_multipart()
        assert message == [b'hello', b'world']

    with run_in_background(run) as event:
        async with azmq.Context(loop=event_loop) as context:
            socket = context.socket(azmq.PUSH)
            connect_or_bind(socket, 'tcp://127.0.0.1:3333')
            await socket.send_multipart([b'hello', b'world'])

            while not event.is_set():
                await asyncio.sleep(0.1)

def test_shadow_pyczmq(self):
        try:
            from pyczmq import zctx, zsocket
        except Exception:
            raise SkipTest("Requires pyczmq")

        ctx = zctx.new()
        ca = zsocket.new(ctx, zmq.PUSH)
        cb = zsocket.new(ctx, zmq.PULL)
        a = zmq.Socket.shadow(ca)
        b = zmq.Socket.shadow(cb)
        a.bind("inproc://a")
        b.connect("inproc://a")
        a.send(b'hi')
        rcvd = self.recv(b)
        self.assertEqual(rcvd, b'hi')

def test_cyclic_destroy(self):
        """ctx.destroy should succeed when cyclic ref prevents gc"""
        # test credit @dln (GH #137):
        class CyclicReference(object):
            def __init__(self, parent=None):
                self.parent = parent

            def crash(self, sock):
                self.sock = sock
                self.child = CyclicReference(self)

        def crash_zmq():
            ctx = self.Context()
            sock = ctx.socket(zmq.PULL)
            c = CyclicReference()
            c.crash(sock)
            ctx.destroy()

        crash_zmq()

def test_cyclic_destroy(self):
        """ctx.destroy should succeed when cyclic ref prevents gc"""
        # test credit @dln (GH #137):
        class CyclicReference(object):
            def __init__(self, parent=None):
                self.parent = parent

            def crash(self, sock):
                self.sock = sock
                self.child = CyclicReference(self)

        def crash_zmq():
            ctx = self.Context()
            sock = ctx.socket(zmq.PULL)
            c = CyclicReference()
            c.crash(sock)
            ctx.destroy()

        crash_zmq()

def test_shadow_pyczmq(self):
        try:
            from pyczmq import zctx, zsocket, zstr
        except Exception:
            raise SkipTest("Requires pyczmq")

        ctx = zctx.new()
        a = zsocket.new(ctx, zmq.PUSH)
        zsocket.bind(a, "inproc://a")
        ctx2 = self.Context.shadow_pyczmq(ctx)
        b = ctx2.socket(zmq.PULL)
        b.connect("inproc://a")
        zstr.send(a, b'hi')
        rcvd = self.recv(b)
        self.assertEqual(rcvd, b'hi')
        b.close()

def test_shadow_pyczmq(self):
        try:
            from pyczmq import zctx, zsocket
        except Exception:
            raise SkipTest("Requires pyczmq")

        ctx = zctx.new()
        ca = zsocket.new(ctx, zmq.PUSH)
        cb = zsocket.new(ctx, zmq.PULL)
        a = zmq.Socket.shadow(ca)
        b = zmq.Socket.shadow(cb)
        a.bind("inproc://a")
        b.connect("inproc://a")
        a.send(b'hi')
        rcvd = self.recv(b)
        self.assertEqual(rcvd, b'hi')

def test_cyclic_destroy(self):
        """ctx.destroy should succeed when cyclic ref prevents gc"""
        # test credit @dln (GH #137):
        class CyclicReference(object):
            def __init__(self, parent=None):
                self.parent = parent

            def crash(self, sock):
                self.sock = sock
                self.child = CyclicReference(self)

        def crash_zmq():
            ctx = self.Context()
            sock = ctx.socket(zmq.PULL)
            c = CyclicReference()
            c.crash(sock)
            ctx.destroy()

        crash_zmq()

def test_shadow_pyczmq(self):
        try:
            from pyczmq import zctx, zsocket, zstr
        except Exception:
            raise SkipTest("Requires pyczmq")

        ctx = zctx.new()
        a = zsocket.new(ctx, zmq.PUSH)
        zsocket.bind(a, "inproc://a")
        ctx2 = self.Context.shadow_pyczmq(ctx)
        b = ctx2.socket(zmq.PULL)
        b.connect("inproc://a")
        zstr.send(a, b'hi')
        rcvd = self.recv(b)
        self.assertEqual(rcvd, b'hi')
        b.close()

def test_shadow_pyczmq(self):
        try:
            from pyczmq import zctx, zsocket
        except Exception:
            raise SkipTest("Requires pyczmq")

        ctx = zctx.new()
        ca = zsocket.new(ctx, zmq.PUSH)
        cb = zsocket.new(ctx, zmq.PULL)
        a = zmq.Socket.shadow(ca)
        b = zmq.Socket.shadow(cb)
        a.bind("inproc://a")
        b.connect("inproc://a")
        a.send(b'hi')
        rcvd = self.recv(b)
        self.assertEqual(rcvd, b'hi')