

def run(self):
    '\n        Tries to continually dispatch messages to consumers.\n        '
    if self.signal_handlers:
        self.install_signal_handler()
    channels = self.channel_layer.router.channels
    while (not self.termed):
        self.in_job = False
        (channel, content) = self.channel_layer.receive_many(channels, block=True)
        self.in_job = True
        if (channel is None):
            time.sleep(0.01)
            continue
        logger.debug('Worker got message on %s: repl %s', channel, content.get('reply_channel', 'none'))
        message = Message(content=content, channel_name=channel, channel_layer=self.channel_layer)
        if (content.get('__retries__', 0) == self.message_retries):
            message.__doomed__ = True
        match = self.channel_layer.router.match(message)
        if (match is None):
            logger.exception('Could not find match for message on %s! Check your routing.', channel)
            continue
        else:
            (consumer, kwargs) = match
        if self.callback:
            self.callback(channel, message)
        try:
            consumer(message, **kwargs)
        except ConsumeLater:
            content['__retries__'] = (content.get('__retries__', 0) + 1)
            if (content['__retries__'] > self.message_retries):
                logger.warning('Exceeded number of retries for message on channel %s: %s', channel, repr(content)[:100])
                continue
            self.channel_layer.send(channel, content)
        except:
            logger.exception('Error processing message with consumer %s:', name_that_thing(consumer))

dataset/ETHPy150Open andrewgodwin/channels/channels/worker.py Worker.run/original