Same as above, but with a PULL socket.
        (e.g good for pushed 'inproc' messages from another thread).
979970
def add_hook_pull(self, address, callback):
        
        LOGGER.info("Subscriber adding PULL hook %s", str(address))
        socket = get_context().socket(PULL)
        socket.connect(address)
        self._add_hook(socket, callback)
1371218
def dispatch_sockets(self, timeout=None):
        """"""
        for sock in self.select_sockets(timeout=timeout):
            if sock is self.listener:
                listener = sock
                sock, addr = listener.accept()
                self.connected(sock)
            else:
                try:
                    sock.recv(1)
                except socket.error as exc:
                    if exc.errno != ECONNRESET:
                        raise
                self.disconnected(sock)
471222
def zthread_fork(ctx, func, *args, **kwargs):
    
    a = ctx.socket(zmq.PAIR)
    a.setsockopt(zmq.LINGER, 0)
    a.setsockopt(zmq.RCVHWM, 100)
    a.setsockopt(zmq.SNDHWM, 100)
    a.setsockopt(zmq.SNDTIMEO, 5000)
    a.setsockopt(zmq.RCVTIMEO, 5000)
    b = ctx.socket(zmq.PAIR)
    b.setsockopt(zmq.LINGER, 0)
    b.setsockopt(zmq.RCVHWM, 100)
    b.setsockopt(zmq.SNDHWM, 100)
    b.setsockopt(zmq.SNDTIMEO, 5000)
    a.setsockopt(zmq.RCVTIMEO, 5000)
    iface = "inproc://%s" % binascii.hexlify(os.urandom(8))
    a.bind(iface)
    b.connect(iface)

    thread = threading.Thread(target=func, args=((ctx, b) + args), kwargs=kwargs)
    thread.daemon = False
    thread.start()

    return a
206197
def socketpair(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0):
    """"""
    # We create a connected TCP socket. Note the trick with setblocking(0)
    # that prevents us from having to create a thread.
    lsock = socket.socket(family, type, proto)
    lsock.bind(('localhost', 0))
    lsock.listen(1)
    addr, port = lsock.getsockname()
    csock = socket.socket(family, type, proto)
    csock.setblocking(False)
    try:
        csock.connect((addr, port))
    except socket.error as e:
        if e.errno != errno.WSAEWOULDBLOCK:
            lsock.close()
            csock.close()
            raise
    ssock, _ = lsock.accept()
    csock.setblocking(True)
    lsock.close()
    return (ssock, csock)
153866
def send_packed_command(self, command):
        ""
        if not self._sock:
            self.connect()
        try:
            if isinstance(command, str):
                command = [command]
            for item in command:
                self._sock.sendall(item)
        except socket.timeout:
            self.disconnect()
            raise TimeoutError("Timeout writing to socket")
        except socket.error:
            e = sys.exc_info()[1]
            self.disconnect()
            if len(e.args) == 1:
                errno, errmsg = 'UNKNOWN', e.args[0]
            else:
                errno = e.args[0]
                errmsg = e.args[1]
            raise ConnectionError("Error %s while writing to socket. %s." %
                                  (errno, errmsg))
        except:  # noqa: E722
            self.disconnect()
            raise
295824
def recv(self, bufsize):
    """"""
    assert bufsize > 0, 'a positive bufsize is required'

    if len(self._buffer) < bufsize:
      readable, _, _ = safe_select([self._socket], [], [], self._select_timeout)
      if readable:
        recvd = self._socket.recv(max(self._chunk_size, bufsize))
        self._buffer = self._buffer + recvd
    return_buf, self._buffer = self._buffer[:bufsize], self._buffer[bufsize:]
    return return_buf
1256394
def _request(self, msg_type, msg_data=None):
    """"""
    msg = {}
    msg['type'] = msg_type
    if msg_data:
      msg['data'] = msg_data

    done = False
    tries = 0
    while not done and tries < MAX_RETRIES:
      try:
        MessageSocket.send(self, self.sock, msg)
        done = True
      except socket.error as e:
        tries += 1
        if tries >= MAX_RETRIES:
          raise
        print("Socket error: {}".format(e))
        self.sock.close()
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect(self.server_addr)

    logging.debug("sent: {0}".format(msg))
    resp = MessageSocket.receive(self, self.sock)
    logging.debug("received: {0}".format(resp))
    return resp
268632
def _on_fork(cls):
        
        cls._cls_idle_socketpairs = []
        while cls._cls_all_sockets:
            cls._cls_all_sockets.pop().close()
78113
def socket_set_hwm(socket, hwm=-1):
    """"""
    try:
        socket.sndhwm = socket.rcvhwm = hwm
    except AttributeError:
        socket.hwm = hwm
458911
public static ZMsg recvMsg(Socket socket, int flag)
    {
        if (socket == null) {
            throw new IllegalArgumentException("socket is null");
        }

        ZMsg msg = new ZMsg();

        while (true) {
            ZFrame f = ZFrame.recvFrame(socket, flag);
            if (f == null) {
                // If receive failed or was interrupted
                msg.destroy();
                msg = null;
                break;
            }
            msg.add(f);
            if (!f.hasMore()) {
                break;
            }
        }
        return msg;
    }
952982
def poll(in_sockets, out_sockets, timeout=-1):
    
    sockets = {}
    # reverse map fd => socket
    fd_sockets = {}
    for s in in_sockets:
        sockets[s.fd] = POLLIN
        fd_sockets[s.fd] = s
    for s in out_sockets:
        modes = sockets.get(s.fd, 0)
        sockets[s.fd] = modes | POLLOUT
        fd_sockets[s.fd] = s

    # convert to milliseconds or -1
    if timeout >= 0:
        timeout_ms = int(timeout*1000)
    else:
        timeout_ms = -1
    res, sockets = wrapper.nn_poll(sockets, timeout_ms)
    _nn_check_positive_rtn(res)
    read_list, write_list = [], []
    for fd, result in sockets.items():
        if (result & POLLIN) != 0:
            read_list.append(fd_sockets[fd])
        if (result & POLLOUT) != 0:
            write_list.append(fd_sockets[fd])

    return read_list, write_list
230864
def _maybe_connect(self, to_pid, callback=None):
    """"""

    callback = stack_context.wrap(callback or (lambda stream: None))

    def streaming_callback(data):
      # we are not guaranteed to get an acknowledgment, but log and discard bytes if we do.
      log.info('Received %d bytes from %s, discarding.' % (len(data), to_pid))
      log.debug('  data: %r' % (data,))

    def on_connect(exit_cb, stream):
      log.info('Connection to %s established' % to_pid)
      with self._connection_callbacks_lock:
        self._connections[to_pid] = stream
      self.__dispatch_on_connect_callbacks(to_pid, stream)
      self.__loop.add_callback(
          stream.read_until_close,
          exit_cb,
          streaming_callback=streaming_callback)

    create = False
    with self._connection_callbacks_lock:
      stream = self._connections.get(to_pid)
      callbacks = self._connection_callbacks.get(to_pid)

      if not stream:
        self._connection_callbacks[to_pid].append(callback)

        if not callbacks:
          create = True

    if stream:
      self.__loop.add_callback(callback, stream)
      return

    if not create:
      return

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    if not sock:
      raise self.SocketError('Failed opening socket')

    stream = IOStream(sock, io_loop=self.__loop)
    stream.set_nodelay(True)
    stream.set_close_callback(partial(self.__on_exit, to_pid, b'reached end of stream'))

    connect_callback = partial(on_connect, partial(self.__on_exit, to_pid), stream)

    log.info('Establishing connection to %s' % to_pid)

    stream.connect((to_pid.ip, to_pid.port), callback=connect_callback)

    if stream.closed():
      raise self.SocketError('Failed to initiate stream connection')

    log.info('Maybe connected to %s' % to_pid)
563411
def _emit_message(cls, message):
        """"""
        sys.stdout.write(message)
        sys.stdout.flush()
294111
def _reader_thread_proc(self, timeout):
    """"""
    for data in self.stream.read_until_close(timeout_ms=timeout):
      if self.stdout is not None:
        self.stdout.write(data)
1054191
def initialize_socket(self):
        """"""
        try:
            _LOGGER.debug("Trying to open socket.")
            self._socket = socket.socket(
                socket.AF_INET,     # IPv4
                socket.SOCK_DGRAM   # UDP
            )
            self._socket.bind(('', self._udp_port))
        except socket.error as err:
            raise err
        else:
            _LOGGER.debug("Socket open.")
            socket_thread = threading.Thread(
                name="SocketThread", target=socket_worker,
                args=(self._socket, self.messages,))
            socket_thread.setDaemon(True)
            socket_thread.start()
519599
def send(self, timeout=None):
        """"""
        if self.sigint_event and is_main_thread():
            with ReplacedSigIntHandler(self.sigint_handler):
                return self._send(timeout)
        else:
            return self._send(timeout)
820091
public static Socket fork(ZContext ctx, IAttachedRunnable runnable, Object... args)
    {
        Socket pipe = ctx.createSocket(SocketType.PAIR);

        if (pipe != null) {
            pipe.bind(String.format("inproc://zctx-pipe-%d", pipe.hashCode()));
        }
        else {
            return null;
        }

        //  Connect child pipe to our pipe
        ZContext ccontext = ZContext.shadow(ctx);
        Socket cpipe = ccontext.createSocket(SocketType.PAIR);
        if (cpipe == null) {
            return null;
        }
        cpipe.connect(String.format("inproc://zctx-pipe-%d", pipe.hashCode()));

        //  Prepare child thread
        Thread shim = new ShimThread(ccontext, runnable, args, cpipe);
        shim.start();

        return pipe;
    }