

def handle(self, msg):

    def func():
        consumer = self._pool.get()
        amqp_proxy = AMQPProxy(self._channel, msg)

        def put_back(successful_greenlet):
            logger.debug('Successful run, putting consumer back')
            self._pool.put(consumer)

        def recreate(failed_greenlet):
            logger.info('Consume failed, shutting down consumer')
            if (not amqp_proxy.has_responded_to_message):
                amqp_proxy.reject(requeue=True)
            shutdown_greenlet = gevent.Greenlet(consumer.shutdown, failed_greenlet.exception)

            def create_wrapper(*args):
                self._create()
            shutdown_greenlet.link(create_wrapper)
            shutdown_greenlet.start()
        greenlet = self._gm(consumer.consume, amqp_proxy, msg)
        greenlet.link_value(put_back)
        greenlet.link_exception(recreate)
        greenlet.start()
    msg._gm(func).start()

dataset/ETHPy150Open seatgeek/cronq/cronq/runner.py ConsumerPool.handle/VarMisuse@185/264 `self`->`msg`