public function consume(AbstractConsumer $consumer, $auto_ack = false)
    {
        if (!$consumer->active()) {
            // prevent dirty consumer been listening on queue
            return;
        }

        $outside_error = null;

        try {
            $consumer->begin($this);

            $this->queue->consume(
                        function (AMQPEnvelope $envelope /*, AMQPQueue $queue*/) use ($consumer) {
                            $delivery = $this->builder->build($envelope);
                            $this->feed($delivery, $consumer);

                            return $consumer->active();
                        }, $auto_ack ? AMQP_AUTOACK : AMQP_NOPARAM
            );

        } catch (Exception $e) {
            $outside_error = $e;
        }

        try {
            $this->queue->cancel();
        } catch (Exception $e) {
        }

        $consumer->end($this, $outside_error);

        if ($outside_error) {
            throw $outside_error;
        }
    }