rpc.py 文件源码

def __call__(self):
        """
        Entry point.
        """

        # Establish a connection with Redis server.
        self._redis = redis.StrictRedis(**connection.get_redis_settings())
        self._pubsub = self._redis.pubsub(ignore_subscribe_messages=True)
        self._pubsub.subscribe(connection.QUERYOBSERVER_REDIS_CHANNEL)

        while self._pubsub.subscribed:
            event = self._pubsub.get_message(ignore_subscribe_messages=True, timeout=0.1)
            if not event:
                continue

            # Events are assumed to be pickled data.
            try:
                event = pickle.loads(event['data'])
            except ValueError:
                logger.error("Ignoring received malformed event '{}'.", event['data'][:20])
                continue

            # Handle event.
            try:
                event_name = event.pop('event')
                handler = getattr(self, 'event_%s' % event_name)
            except AttributeError:
                logger.error("Ignoring unimplemented event '{}'.", event_name)
                continue
            except KeyError:
                logger.error("Ignoring received malformed event '{}'.", event)
                continue

            try:
                handler(**event)
            except:
                logger.error("Unhandled exception while executing event '{}'.", event_name)
                logger.error(traceback.format_exc())
            finally:
                db.close_old_connections()

        self._pubsub.close()