public function consume($callback, $consumeAmount = null)
    {

        $channel = $this->connection->channel();

        // Exchange setup
        $this->setupExchange($this->exchangeOptions['name'], $this->exchangeOptions['type'], $channel);

        foreach ($this->queueOptions as $queueOption) {
            // Queue setup
            list($channel, ) = $this->setupQueue($queueOption['name'], $channel);

            // Bind the queue to the exchange
            $channel->queue_bind($queueOption['name'], $this->exchangeOptions['name'], $queueOption['bindingKey']);

            if (!$consumeAmount == null) {
                // @todo: Investigate if large unack amounts result in multi consumers being able to process large queues.
                // This currently set the unacked limit to a single message resulting in every consumer getting access to
                // the next message in the queue.
                // $channel->qos(NULL, $consumeAmount, null);
                $channel->basic_qos(null, 1, null);
            }

            // Start the consumer
            $channel->basic_consume(
                $queueOption['name'],
                $this->consumeOptions['consumer_tag'],
                $this->consumeOptions['no_local'],
                $this->consumeOptions['no_ack'],
                $this->consumeOptions['exclusive'],
                $this->consumeOptions['nowait'],
                $callback
            );
        }

        // Wait for messages on the channel
        echo ' [*] Waiting for messages = ' . date('D M j G:i:s T Y') . '. To exit press CTRL+C', PHP_EOL;
        $this->channel = $channel;
        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }

        $this->channel->close();
    }
public function consume($callback, $consumeAmount = null)
    {

        $channel = $this->connection->channel();

        // Exchange setup
        $this->setupExchange($this->exchangeOptions['name'], $this->exchangeOptions['type'], $channel);

        foreach ($this->queueOptions as $queueOption) {
            // Queue setup
            list($channel, ) = $this->setupQueue($queueOption['name'], $channel);

            // Bind the queue to the exchange
            $channel->queue_bind($queueOption['name'], $this->exchangeOptions['name'], $queueOption['bindingKey']);

            if (!$consumeAmount == null) {
                // @todo: Investigate if large unack amounts result in multi consumers being able to process large queues.
                // This currently set the unacked limit to a single message resulting in every consumer getting access to
                // the next message in the queue.
                // $channel->qos(NULL, $consumeAmount, null);
                $channel->basic_qos(null, 1, null);
            }

            // Start the consumer
            $channel->basic_consume(
                $queueOption['name'],
                $this->consumeOptions['consumer_tag'],
                $this->consumeOptions['no_local'],
                $this->consumeOptions['no_ack'],
                $this->consumeOptions['exclusive'],
                $this->consumeOptions['nowait'],
                $callback
            );
        }

        // Wait for messages on the channel
        echo ' [*] Waiting for messages = ' . date('D M j G:i:s T Y') . '. To exit press CTRL+C', PHP_EOL;
        $this->channel = $channel;
        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }

        $this->channel->close();
    }
public function consume ( $ callback , $ consumeAmount = null ) { $ channel = $ this -> connection -> channel ( ) ; // Exchange setup $ this -> setupExchange ( $ this -> exchangeOptions [ 'name' ] , $ this -> exchangeOptions [ 'type' ] , $ channel ) ; foreach ( $ this -> queueOptions as $ queueOption ) { // Queue setup list ( $ channel , ) = $ this -> setupQueue ( $ queueOption [ 'name' ] , $ channel ) ; // Bind the queue to the exchange $ channel -> queue_bind ( $ queueOption [ 'name' ] , $ this -> exchangeOptions [ 'name' ] , $ queueOption [ 'bindingKey' ] ) ; if ( ! $ consumeAmount == null ) { // @todo: Investigate if large unack amounts result in multi consumers being able to process large queues. // This currently set the unacked limit to a single message resulting in every consumer getting access to // the next message in the queue. // $channel->qos(NULL, $consumeAmount, null); $ channel -> basic_qos ( null , 1 , null ) ; } // Start the consumer $ channel -> basic_consume ( $ queueOption [ 'name' ] , $ this -> consumeOptions [ 'consumer_tag' ] , $ this -> consumeOptions [ 'no_local' ] , $ this -> consumeOptions [ 'no_ack' ] , $ this -> consumeOptions [ 'exclusive' ] , $ this -> consumeOptions [ 'nowait' ] , $ callback ) ; } // Wait for messages on the channel echo ' [*] Waiting for messages = ' . date ( 'D M j G:i:s T Y' ) . '. To exit press CTRL+C' , PHP_EOL ; $ this -> channel = $ channel ; while ( count ( $ this -> channel -> callbacks ) ) { $ this -> channel -> wait ( ) ; } $ this -> channel -> close ( ) ; }
Consume messages from the message broker queue.

@param $callback
Callback to handle messages the consumer receives.
@param integer $consumeAmount
The number of message to set as unacked and reserve when consumer is sent messages.
Consume messages from the message broker queue .