public function get($class, $connexion, array $exchangeOptions, array $queueOptions, $lazy = false, array $qosOptions = [])
    {
        if (!class_exists($class)) {
            throw new \InvalidArgumentException(
                sprintf("Consumer class '%s' doesn't exist", $class)
            );
        }

        if ($lazy) {
            if (!$connexion->isConnected()) {
                $connexion->connect();
            }
        }

        /** @var \AMQPChannel $channel */
        $channel = new $this->channelClass($connexion);

        if (isset($qosOptions['prefetch_size'])) {
            $channel->setPrefetchSize($qosOptions['prefetch_size']);
        }
        if (isset($qosOptions['prefetch_count'])) {
            $channel->setPrefetchCount($qosOptions['prefetch_count']);
        }

        //ensure that exchange exists
        $this->createExchange($this->exchangeClass, $channel, $exchangeOptions);

        /** @var \AMQPQueue $queue */
        $queue = new $this->queueClass($channel);
        if (!empty($queueOptions['name'])) {
            $queue->setName($queueOptions['name']);
        }
        $queue->setArguments($queueOptions['arguments']);
        $queue->setFlags(
            ($queueOptions['passive'] ? AMQP_PASSIVE : AMQP_NOPARAM) |
            ($queueOptions['durable'] ? AMQP_DURABLE : AMQP_NOPARAM) |
            ($queueOptions['exclusive'] ? AMQP_EXCLUSIVE : AMQP_NOPARAM) |
            ($queueOptions['auto_delete'] ? AMQP_AUTODELETE : AMQP_NOPARAM)
        );

        // Declare the queue
        $queue->declareQueue();

        if (count($queueOptions['routing_keys'])) {
            foreach ($queueOptions['routing_keys'] as $routingKey) {
                $queue->bind($exchangeOptions['name'], $routingKey);
            }
        } else {
            $queue->bind($exchangeOptions['name']);
        }

        return new $class($queue, $queueOptions);
    }
public function get($class, $connexion, array $exchangeOptions, array $queueOptions, $lazy = false, array $qosOptions = [])
    {
        if (!class_exists($class)) {
            throw new \InvalidArgumentException(
                sprintf("Consumer class '%s' doesn't exist", $class)
            );
        }

        if ($lazy) {
            if (!$connexion->isConnected()) {
                $connexion->connect();
            }
        }

        /** @var \AMQPChannel $channel */
        $channel = new $this->channelClass($connexion);

        if (isset($qosOptions['prefetch_size'])) {
            $channel->setPrefetchSize($qosOptions['prefetch_size']);
        }
        if (isset($qosOptions['prefetch_count'])) {
            $channel->setPrefetchCount($qosOptions['prefetch_count']);
        }

        //ensure that exchange exists
        $this->createExchange($this->exchangeClass, $channel, $exchangeOptions);

        /** @var \AMQPQueue $queue */
        $queue = new $this->queueClass($channel);
        if (!empty($queueOptions['name'])) {
            $queue->setName($queueOptions['name']);
        }
        $queue->setArguments($queueOptions['arguments']);
        $queue->setFlags(
            ($queueOptions['passive'] ? AMQP_PASSIVE : AMQP_NOPARAM) |
            ($queueOptions['durable'] ? AMQP_DURABLE : AMQP_NOPARAM) |
            ($queueOptions['exclusive'] ? AMQP_EXCLUSIVE : AMQP_NOPARAM) |
            ($queueOptions['auto_delete'] ? AMQP_AUTODELETE : AMQP_NOPARAM)
        );

        // Declare the queue
        $queue->declareQueue();

        if (count($queueOptions['routing_keys'])) {
            foreach ($queueOptions['routing_keys'] as $routingKey) {
                $queue->bind($exchangeOptions['name'], $routingKey);
            }
        } else {
            $queue->bind($exchangeOptions['name']);
        }

        return new $class($queue, $queueOptions);
    }
public function get ( $ class , $ connexion , array $ exchangeOptions , array $ queueOptions , $ lazy = false , array $ qosOptions = [ ] ) { if ( ! class_exists ( $ class ) ) { throw new \ InvalidArgumentException ( sprintf ( "Consumer class '%s' doesn't exist" , $ class ) ) ; } if ( $ lazy ) { if ( ! $ connexion -> isConnected ( ) ) { $ connexion -> connect ( ) ; } } /** @var \AMQPChannel $channel */ $ channel = new $ this -> channelClass ( $ connexion ) ; if ( isset ( $ qosOptions [ 'prefetch_size' ] ) ) { $ channel -> setPrefetchSize ( $ qosOptions [ 'prefetch_size' ] ) ; } if ( isset ( $ qosOptions [ 'prefetch_count' ] ) ) { $ channel -> setPrefetchCount ( $ qosOptions [ 'prefetch_count' ] ) ; } //ensure that exchange exists $ this -> createExchange ( $ this -> exchangeClass , $ channel , $ exchangeOptions ) ; /** @var \AMQPQueue $queue */ $ queue = new $ this -> queueClass ( $ channel ) ; if ( ! empty ( $ queueOptions [ 'name' ] ) ) { $ queue -> setName ( $ queueOptions [ 'name' ] ) ; } $ queue -> setArguments ( $ queueOptions [ 'arguments' ] ) ; $ queue -> setFlags ( ( $ queueOptions [ 'passive' ] ? AMQP_PASSIVE : AMQP_NOPARAM ) | ( $ queueOptions [ 'durable' ] ? AMQP_DURABLE : AMQP_NOPARAM ) | ( $ queueOptions [ 'exclusive' ] ? AMQP_EXCLUSIVE : AMQP_NOPARAM ) | ( $ queueOptions [ 'auto_delete' ] ? AMQP_AUTODELETE : AMQP_NOPARAM ) ) ; // Declare the queue $ queue -> declareQueue ( ) ; if ( count ( $ queueOptions [ 'routing_keys' ] ) ) { foreach ( $ queueOptions [ 'routing_keys' ] as $ routingKey ) { $ queue -> bind ( $ exchangeOptions [ 'name' ] , $ routingKey ) ; } } else { $ queue -> bind ( $ exchangeOptions [ 'name' ] ) ; } return new $ class ( $ queue , $ queueOptions ) ; }
build the consumer class.

@param string          $class           Consumer class name
@param \AMQPConnection $connexion       AMQP connexion
@param array           $exchangeOptions Exchange Options
@param array           $queueOptions    Queue Options
@param bool            $lazy            Specifies if it should connect
@param array           $qosOptions      Qos Options

@return Consumer
build the consumer class .