amqp.py 文件源码

def process_in_message(self, message):
        if 'fetch_uri' not in message:
            raise InvalidAMQPMessage('Given message has no fetch_uri value.')

        message_copy = message.copy()
        jobid = uuid.uuid1().hex
        _message = dict()
        _message['job_id'] = jobid
        _message['fetch_uri'] = message_copy.pop('fetch_uri')

        # Encode additional fields to send back to client on fetch
        try:
            _message['settings'] = amqp_message_encode(message_copy)
        except:
            # @TODO define a more specific exception handling here
            raise EncodingError("Can't encode message info. %s" % message)

        self.scheduler.schedule(_message)
        self.signal_manager.send_catch_log(signal=signals.request_received, jobid=jobid)