subscriber.py 文件源码

def run(self):
        """ Main loop of the thread. """
        # create event socket
        self.subscriber = EventSubscriber(self.zmq_context,
                                          self.event_port,
                                          self.logger)
        self.configure()
        # create poller and register event subscriber
        poller = zmq.Poller()
        poller.register(self.subscriber.socket, zmq.POLLIN)
        # poll events every seconds
        self.logger.info('entering main loop')
        while not self.stop_event.is_set():
            socks = dict(poller.poll(self._Poll_timeout))
            # check if something happened on the socket
            if self.subscriber.socket in socks and \
                socks[self.subscriber.socket] == zmq.POLLIN:
                self.logger.debug('got message on subscriber')
                try:
                    message = self.subscriber.receive()
                except Exception, e:
                    self.logger.error(
                        'failed to get data from subscriber: {}'.format(
                            e.message))
                else:
                    if message[0] == EventHeaders.SUPVISORS:
                        self.on_supvisors_status(message[1])
                    elif message[0] == EventHeaders.ADDRESS:
                        self.on_address_status(message[1])
                    elif message[0] == EventHeaders.APPLICATION:
                        self.on_application_status(message[1])
                    elif message[0] == EventHeaders.PROCESS_EVENT:
                        self.on_process_event(message[1])
                    elif message[0] == EventHeaders.PROCESS_STATUS:
                        self.on_process_status(message[1])
        self.logger.warn('exiting main loop')
        self.subscriber.close()