src/Listener.php
public function listen($queue_name, array $options = null, Closure $closure)
    {
        $this->queue_name = $queue_name;

        if ($options)
            $this->setOptions($options);        

        $GLOBALS['messages_proccesed'] = 0;
        $GLOBALS['start_time'] = time();

        $connection = new Connection($this->buildConnectionOptions());
        $connection->open();

        $listenerObject = $this;

        $connection->channel->basic_consume($this->queue_name, $connection->consumer_tag, false, false, false, false, function ($msg) use ($closure, $listenerObject) {

            try
            {
                $closure($msg->body);
            }            
            catch (Exception $e)
            {
                throw $e;
            }

            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);

            //Update counters
            $GLOBALS['messages_proccesed']++;

            //Check if necesary to close consumer      
            if ($listenerObject->message_limit && $GLOBALS['messages_proccesed'] >= $listenerObject->message_limit)
                $msg->delivery_info['channel']->basic_cancel($msg->delivery_info['consumer_tag']);
            
            if ($listenerObject->time && (time()-$GLOBALS['start_time']>= $listenerObject->time))
                $msg->delivery_info['channel']->basic_cancel($msg->delivery_info['consumer_tag']);
        });

        register_shutdown_function(function ($connection) {
            $connection->close();
        }, $connection);

        // Loop as long as the channel has callbacks registered
        try
        {
            while (count($connection->channel->callbacks)) {
                $connection->channel->wait(null, false, $this->empty_queue_timeout);
            }
        }
        catch (AMQPTimeoutException $e)
        {
            return false;
        }
        catch (Exception $e)
        {
            throw $e;
        }        
    }
public function listen($queue_name, array $options = null, Closure $closure)
    {
        $this->queue_name = $queue_name;

        if ($options)
            $this->setOptions($options);        

        $GLOBALS['messages_proccesed'] = 0;
        $GLOBALS['start_time'] = time();

        $connection = new Connection($this->buildConnectionOptions());
        $connection->open();

        $listenerObject = $this;

        $connection->channel->basic_consume($this->queue_name, $connection->consumer_tag, false, false, false, false, function ($msg) use ($closure, $listenerObject) {

            try
            {
                $closure($msg->body);
            }            
            catch (Exception $e)
            {
                throw $e;
            }

            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);

            //Update counters
            $GLOBALS['messages_proccesed']++;

            //Check if necesary to close consumer      
            if ($listenerObject->message_limit && $GLOBALS['messages_proccesed'] >= $listenerObject->message_limit)
                $msg->delivery_info['channel']->basic_cancel($msg->delivery_info['consumer_tag']);
            
            if ($listenerObject->time && (time()-$GLOBALS['start_time']>= $listenerObject->time))
                $msg->delivery_info['channel']->basic_cancel($msg->delivery_info['consumer_tag']);
        });

        register_shutdown_function(function ($connection) {
            $connection->close();
        }, $connection);

        // Loop as long as the channel has callbacks registered
        try
        {
            while (count($connection->channel->callbacks)) {
                $connection->channel->wait(null, false, $this->empty_queue_timeout);
            }
        }
        catch (AMQPTimeoutException $e)
        {
            return false;
        }
        catch (Exception $e)
        {
            throw $e;
        }        
    }
Listen queue server for given queue name

@param string $queue_name  Queue name to listen
@param array $options  Options to listen
@param Closure $closure Function to run for every message

@return void