src/AMQPy/AbstractListener.php
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;
        }
    }
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;
        }
    }
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 ; } }
Attach consumer to process payload from queue

@param AbstractConsumer $consumer Consumer to process payload and handle possible errors
@param bool             $auto_ack Should message been acknowledged upon receive

@return mixed

@throws SerializerException
@throws Exception           Any exception from pre/post-consume handlers and from exception handler
Attach consumer to process payload from queue