public function consume($queueAlias, $timeout = 0)
    {
        $queues = is_array($queueAlias)
            ? $this
                ->queueAliasResolver
                ->getQueues($queueAlias)
            : $this
                ->queueAliasResolver
                ->getQueue($queueAlias);

        $payloadArray = $this
            ->redis
            ->blPop(
                $queues,
                $timeout
            );

        if (empty($payloadArray)) {
            return [];
        }

        list($givenQueue, $payloadSerialized) = $payloadArray;
        $payload = $this->serializer->revert($payloadSerialized);
        $givenQueueAlias = $this
            ->queueAliasResolver
            ->getQueueAliasByQueueName($givenQueue);

        /*
         * Dispatching consumer event...
         */
        $consumerEvent = new RSQueueConsumerEvent(
            $payload,
            $payloadSerialized,
            $givenQueue,
            $this->redis
        );

        $this
            ->eventDispatcher
            ->dispatch(RSQueueEvents::RSQUEUE_CONSUMER, $consumerEvent);

        return [$givenQueueAlias, $payload];
    }
public function consume($queueAlias, $timeout = 0)
    {
        $queues = is_array($queueAlias)
            ? $this
                ->queueAliasResolver
                ->getQueues($queueAlias)
            : $this
                ->queueAliasResolver
                ->getQueue($queueAlias);

        $payloadArray = $this
            ->redis
            ->blPop(
                $queues,
                $timeout
            );

        if (empty($payloadArray)) {
            return [];
        }

        list($givenQueue, $payloadSerialized) = $payloadArray;
        $payload = $this->serializer->revert($payloadSerialized);
        $givenQueueAlias = $this
            ->queueAliasResolver
            ->getQueueAliasByQueueName($givenQueue);

        /*
         * Dispatching consumer event...
         */
        $consumerEvent = new RSQueueConsumerEvent(
            $payload,
            $payloadSerialized,
            $givenQueue,
            $this->redis
        );

        $this
            ->eventDispatcher
            ->dispatch(RSQueueEvents::RSQUEUE_CONSUMER, $consumerEvent);

        return [$givenQueueAlias, $payload];
    }
public function consume ( $ queueAlias , $ timeout = 0 ) { $ queues = is_array ( $ queueAlias ) ? $ this -> queueAliasResolver -> getQueues ( $ queueAlias ) : $ this -> queueAliasResolver -> getQueue ( $ queueAlias ) ; $ payloadArray = $ this -> redis -> blPop ( $ queues , $ timeout ) ; if ( empty ( $ payloadArray ) ) { return [ ] ; } list ( $ givenQueue , $ payloadSerialized ) = $ payloadArray ; $ payload = $ this -> serializer -> revert ( $ payloadSerialized ) ; $ givenQueueAlias = $ this -> queueAliasResolver -> getQueueAliasByQueueName ( $ givenQueue ) ; /*
         * Dispatching consumer event...
         */ $ consumerEvent = new RSQueueConsumerEvent ( $ payload , $ payloadSerialized , $ givenQueue , $ this -> redis ) ; $ this -> eventDispatcher -> dispatch ( RSQueueEvents :: RSQUEUE_CONSUMER , $ consumerEvent ) ; return [ $ givenQueueAlias , $ payload ] ; }
Retrieve queue value, with a defined timeout.

This method accepts a single queue alias or an array with alias
Every new element will be popped from one of defined queue

Also, new Consumer event is triggered everytime a new element is popped

@param mixed $queueAlias Alias of queue to consume from ( Can be an array of alias )
@param int   $timeout    Timeout. By default, 0

@return array
Retrieve queue value with a defined timeout .