public function getAllMessages($callback)
    {

        $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']);
        }

        $this->channel = $channel;
        $data = [];
        while ($message = $this->channel->basic_get($queueOption['name'])) {
            if ($message->delivery_info) {
                $message->delivery_info['channel'] = $this->channel;
                $data[] = $message;
            }
        }

        call_user_func($callback, $data);
        $this->channel->close();
    }
public function getAllMessages($callback)
    {

        $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']);
        }

        $this->channel = $channel;
        $data = [];
        while ($message = $this->channel->basic_get($queueOption['name'])) {
            if ($message->delivery_info) {
                $message->delivery_info['channel'] = $this->channel;
                $data[] = $message;
            }
        }

        call_user_func($callback, $data);
        $this->channel->close();
    }
public function getAllMessages ( $ callback ) { $ 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' ] ) ; } $ this -> channel = $ channel ; $ data = [ ] ; while ( $ message = $ this -> channel -> basic_get ( $ queueOption [ 'name' ] ) ) { if ( $ message -> delivery_info ) { $ message -> delivery_info [ 'channel' ] = $ this -> channel ; $ data [ ] = $ message ; } } call_user_func ( $ callback , $ data ) ; $ this -> channel -> close ( ) ; }
Consume all messages from the message broker queue.

@param $callback
Callback to handle messages the consumer receives.
Consume all messages from the message broker queue .