How to use the zmq.NOBLOCK function in zmq

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

def abort_queue(self):
        while True:
            try:
                ident = self.reply_socket.recv(zmq.NOBLOCK)
            except zmq.ZMQError, e:
                if e.errno == zmq.EAGAIN:
                    break
            else:
                assert self.reply_socket.rcvmore(), "Unexpected missing message part."
                msg = self.reply_socket.recv_json()
            print>>sys.__stdout__, "Aborting:"
            print>>sys.__stdout__, Message(msg)
            msg_type = msg['msg_type']
            reply_type = msg_type.split('_')[0] + '_reply'
            reply_msg = self.session.msg(reply_type, {'status' : 'aborted'}, msg)
            print>>sys.__stdout__, Message(reply_msg)
            self.reply_socket.send(ident,zmq.SNDMORE)
            self.reply_socket.send_json(reply_msg)
            # We need to wait a bit for requests to come in. This can probably
            # be set shorter for true asynchronous clients.

def poll():
    setup()
    while True:
        try:
            cmd = socket.recv_json(zmq.NOBLOCK)
            if cmd == 'depth':
                opennpy.sync_update()
                depth,_ = opennpy.sync_get_depth()
                socket.send_json(('depth',depth.tolist()))
            if cmd == 'rgb':
                opennpy.sync_update()
                rgb,_ = opennpy.sync_get_video()
                socket.send_json(('rgb',rgb.tolist()))
        except zmq.ZMQError, e:
            if e.errno == zmq.EAGAIN: break
            else: raise

def _checkMonitor(monitor):
    try:
        event_monitor = monitor.recv(zmq.NOBLOCK)
    except zmq.Again:
        return

    (event_id, event_name, event_value) = _readSocketEvent(event_monitor)
    event_monitor_endpoint = monitor.recv(zmq.NOBLOCK)
    logger.info(
        "Socket monitor: {} {} endpoint {}".format(
            event_name, event_value, event_monitor_endpoint
        )

while self.running:
            # if we have any sockets that needs registering in the poller, register them before we begin the poll
            self._register_pending_sockets()

            # Don't spin out of control if there are no sockets to poll
            if not self._poller.sockets:
                time.sleep(self.poll_timeout/1000.)
                continue

            for polled_socket, polled_type in self._poller.poll():
                # Get hold of a batch of messages. We deal with
                # batches to amortize the overhead of getting them.
                message_batch = list()
                try:
                    for i in range(self.maximum_batch_size):
                        message_batch.append(polled_socket.recv(zmq.NOBLOCK))
                except zmq.ZMQError, e:
                    if e.errno != zmq.EAGAIN:
                        raise
                    # We've emptied the queue. That's fine!

                if polled_socket == self._control_socket_pull:
                    # it was just our control socket. do nothing
                    continue

                # Don't poll more messages from this socket until it
                # is re-registered.  With no messages getting polled
                # out of it, it'll eventually hit the high watermark,
                # making sure that upstream sockets don't deliver us
                # more messages until we're ready to accept them.
                self._poller.unregister(polled_socket)

headers_mod.CONTENT_TYPE: headers_mod.CONTENT_TYPE.JSON,
            headers_mod.DATE: now,
            headers_mod.TIMESTAMP: now
        }
         
        for point, value in results.iteritems():
            if isinstance(value, bool):
                value = int(value)
            self.add('/'+point, value)
            
        try:    
            for point, value in results.iteritems():
                if isinstance(value, bool):
                    value = int(value)
                depth, breadth = self.get_paths_for_point('/'+point)
                self.publish_json(depth, headers, value, self.meta_data[point], flags=zmq.NOBLOCK)
                self.publish_json(breadth, headers, value, self.meta_data[point], flags=zmq.NOBLOCK)
                
            self.publish_json(self.all_path_depth, headers, results, self.meta_data, flags=zmq.NOBLOCK)
            self.publish_json(self.all_path_breadth, headers, results, self.meta_data, flags=zmq.NOBLOCK)
        except zmq.error.Again:
            print ("Warning: platform not running, topics not published. (Data to smap historian is unaffected by this warning)")

def _handle_recv(self):
        """Handle a recv event."""
        if self._flushed:
            return
        try:
            msg = self.socket.recv_multipart(zmq.NOBLOCK, copy=self._recv_copy)
        except zmq.ZMQError as e:
            if e.errno == zmq.EAGAIN:
                # state changed since poll event
                pass
            else:
                gen_log.error("RECV Error: %s"%zmq.strerror(e.errno))
        else:
            if self._recv_callback:
                callback = self._recv_callback
                # self._recv_callback = None
                self._run_callback(callback, msg)

def FlushSocket(socket):
  while True:
    try:
      socket.recv(zmq.NOBLOCK)
    except zmq.ZMQError, e:
      if e.errno == zmq.EAGAIN:
        break
      else:
        raise

if os.WIFEXITED(status):
                        self._workers.discard(pid)
                        self._success = os.WEXITSTATUS(status)
                        if self._success == 0:
                            # A worker has terminated normally. Other workers
                            # are expected to terminate normally too, so
                            # continue.
                            continue
                        else:
                            # A worker has not terminated normally due to an
                            # error or an irrecoverable fault
                            raise PlatoonError("A worker has exited with non-success code: {}".format(self._success))
                    else:  # other status changes are not desirable
                        raise PlatoonError("A worker has changed to a status other than exit.")
                try:
                    query = self.csocket.recv_json(flags=zmq.NOBLOCK)
                except zmq.Again:  # if a query has not happened, try again
                    continue
                except zmq.ZMQError as exc:
                    raise PlatoonError("while receiving using ZMQ socket", exc)

                # try default interface, it may raise PlatoonError
                response = self._handle_base_control(query['req'],
                                                     query['worker_id'],
                                                     query['req_info'])
                if response is None:
                    response = self.handle_control(query['req'],
                                                   query['worker_id'],
                                                   query['req_info'])

                try:
                    self.csocket.send_json(response)

socks = dict(poller.poll(5))

  except KeyboardInterrupt:
    can_socket.close()
    steer_socket.close()
    steer2_socket.close()
    lat_socket.close()
    context.term()
    break      

  for sock in socks:
    
    while 1:
      thisData = None
      try:
        thisData = sock.recv(zmq.NOBLOCK)
      except zmq.error.Again:
        thisData = None
        break
      except KeyboardInterrupt:
        can_socket.close()
        steer_socket.close()
        steer2_socket.close()
        lat_socket.close()
        context.term()
        break      

      receiveTime = int(time.time() * 1000000000)

      doRecordCANData = False
      doRecordCANBytes = False