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;
        }        
    }
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
Listen queue server for given queue name