@param      $queueName

@param      $timeout

@param null $refresh

@return RouterRequestConsumer
331945
public function getConsumer($queueName, $timeout = null, $refresh = null)
    {
        $queueName = $this->_prepareQueueName($queueName);
        $thread    = self::THREAD_CONSUME . $queueName.':'.$timeout;
        
        $server = $this->_getServerOrFail($queueName, RouterConfig::SERVER_TAG_DEFAULT);
        
        $config = [
            RouterConfig::REQUEST_CONSUMER_TIME_LIMIT => $timeout,
            RouterConfig::REQUEST_CONSUMER_REFRESH_INTERVAL => $refresh 
        ];
        
        $consumer = $server->getRequestConsumer($thread, $queueName, RouterConfig::CONFIG_PERSISTENT_CONSUMER, $config);
        
        return $consumer;
    }
1332496
public function execute($max = 50, $timeout = self::RECEIVE_TIMEOUT_DEFAULT)
    {
        $this->getManager()->executeQueue($this, $max, $timeout);
    }
959988
protected function wait_frame($timeout = 0)
    {
        if (is_null($this->input))
        {
            $this->setIsConnected(false);
            throw new AMQPConnectionClosedException('Broken pipe or closed connection');
        }

        $currentTimeout = $this->input->getTimeout();
        $this->input->setTimeout($timeout);

        try {
            // frame_type + channel_id + size
            $this->wait_frame_reader->reuse(
                $this->input->read(AMQPReader::OCTET + AMQPReader::SHORT + AMQPReader::LONG)
            );

            $frame_type = $this->wait_frame_reader->read_octet();
            $class = self::$PROTOCOL_CONSTANTS_CLASS;
            if (!array_key_exists($frame_type, $class::$FRAME_TYPES)) {
                throw new AMQPInvalidFrameException('Invalid frame type ' . $frame_type);
            }
            $channel = $this->wait_frame_reader->read_short();
            $size = $this->wait_frame_reader->read_long();

            // payload + ch
            $this->wait_frame_reader->reuse($this->input->read(AMQPReader::OCTET + (int) $size));

            $payload = $this->wait_frame_reader->read($size);
            $ch = $this->wait_frame_reader->read_octet();

        } catch (AMQPTimeoutException $e) {
            $this->input->setTimeout($currentTimeout);
            throw $e;
        } catch (AMQPNoDataException $e) {
            if ($this->input) {
                $this->input->setTimeout($currentTimeout);
            }
            throw $e;
        } catch (AMQPConnectionClosedException $exception) {
            $this->do_close();
            throw $exception;
        }

        $this->input->setTimeout($currentTimeout);

        if ($ch != 0xCE) {
            throw new AMQPInvalidFrameException(sprintf(
                'Framing error, unexpected byte: %x',
                $ch
            ));
        }

        return array($frame_type, $channel, $payload);
    }
238430
public function getQueue($name, $timeout = null)
    {
        if (!is_string($name)) {
            throw new Exception('$name is not a string');
        }

        if ((null !== $timeout) && !is_integer($timeout)) {
            throw new Exception('$timeout must be an integer');
        }

        if(!$this->getQueueId($name)) {
            $this->create($name, $timeout);
        }

        $q = new Queue($name, $timeout);
        $q->setManager($this);
        return $q;
    }
847631
private function resolveEndpoint($queueName)
    {
        if (isset($this->queueMap[$queueName])) {
            return $this->queueMap[$queueName];
        }

        return '/_ah/queue/'.$queueName;
    }
566651
public function getRequestConsumer($thread, $queueName, $queueConfigId, $config = [])
    {
        if (!isset($this->requestConsumers[$thread])) {
            $configuration = $config + [
                RouterConfig::COMMON_QUEUE_NAME             => $queueName,
                RouterConfig::COMMON_QUEUE_CONFIGURATION_ID => $queueConfigId,
            ];
            
            $requestConsumer = new RouterRequestConsumer($this, $thread, $configuration);
            $requestConsumer->setup();
            
            $this->requestConsumers[$thread] = $requestConsumer;
        }
        
        return $this->requestConsumers[$thread];
    }
854056
public function consume($queueName, array $options = [], callable $callback)
    {
        $consumerTag = array_key_exists('consumerTag', $options) ? $options['consumerTag'] : null;
        $noLocal = array_key_exists('noLocal', $options) ? (bool)$options['noLocal'] : null;
        $noAck = array_key_exists('noAck', $options) ? (bool)$options['noAck'] : null;
        $exclusive = array_key_exists('exclusive', $options) ? (bool)$options['exclusive'] : null;
        $noWait = array_key_exists('noWait', $options) ? (bool)$options['noWait'] : null;
        $arguments = array_key_exists('arguments', $options) ? $options['arguments'] : null;

        $outputFrame = Basic\BasicConsumeFrame::create(
            $queueName,
            $consumerTag,
            $noLocal,
            $noAck,
            $exclusive,
            $noWait,
            $arguments
        );
        $outputFrame->frameChannelId = $this->id;
        $this->connection->command($outputFrame);

        if (is_callable($callback)) {
            $this->on(self::EVENT_ON_CHANNEL_CONSUMEOK_CALLBACK, function (Basic\BasicConsumeOkFrame $incomingFrame) use ($callback) {
                $this->consumers[$incomingFrame->consumerTag] = $callback;
            });
        }
    }
1007665
public function setTimeout($timeout) {
        $this->_timeout = (int) $timeout;
        if ( ! is_null($this->_varnishConn)) {
            stream_set_timeout($this->_varnishConn, $this->_timeout);
        }
        return $this;
    }
768608
@Override
    public java.util.concurrent.Future<GetQueueUrlResult> getQueueUrlAsync(String queueName) {

        return getQueueUrlAsync(new GetQueueUrlRequest().withQueueName(queueName));
    }
69480
protected function _receive($timeout = null)
    {
        if (!is_resource($this->_socket)) {
            /**
             * @see Zend_Mail_Protocol_Exception
             */
            // require_once 'Zend/Mail/Protocol/Exception.php';
            throw new Zend_Mail_Protocol_Exception('No connection has been established to ' . $this->_host);
        }

        // Adapters may wish to supply per-commend timeouts according to appropriate RFC
        if ($timeout !== null) {
            $this->_setStreamTimeout($timeout);
        }

        // Retrieve response
        $reponse = fgets($this->_socket, 1024);

        // Save request to internal log
        $this->_addLog($reponse);

        // Check meta data to ensure connection is still valid
        $info = stream_get_meta_data($this->_socket);

        if (!empty($info['timed_out'])) {
            /**
             * @see Zend_Mail_Protocol_Exception
             */
            // require_once 'Zend/Mail/Protocol/Exception.php';
            throw new Zend_Mail_Protocol_Exception($this->_host . ' has timed out');
        }

        if ($reponse === false) {
            /**
             * @see Zend_Mail_Protocol_Exception
             */
            // require_once 'Zend/Mail/Protocol/Exception.php';
            throw new Zend_Mail_Protocol_Exception('Could not read from ' . $this->_host);
        }

        return $reponse;
    }
292675
public function createQueue($queueName, array $options = [])
    {
        /** @var \Generated\Shared\Transfer\RabbitMqOptionTransfer $rabbitMqOption */
        $rabbitMqOption = $options['rabbitMqConsumerOption'];

        return $this->queueEstablishmentHelper->createQueue($this->channel, $rabbitMqOption);
    }
315092
private static function queue(string $name = null) : Queue
    {
        list($container, $config, $return) = DI::detect(__METHOD__, 'queue', $name);

        if ($return) {
            return $return;
        }

        $db = new Queue($config);

        return DI::set(__METHOD__, $container, $db);
    }
591995
public BatchReceiver getBatchReceiver(String queueName, long timeout){
        try {
            return new BatchReceiver((Queue) jmsServer.lookup(QUEUE_NAMESPACE + queueName), timeout, consumerConnection);
        } catch (Exception e) {
            throw new AsyncException(e);
        }
    }
672795
protected function receiveMessage(array $queues, int $timeout, int $redeliveryDelay): ?RedisMessage
    {
        $startAt = time();
        $thisTimeout = $timeout;

        if (null === $this->queueNames) {
            $this->queueNames = [];
            foreach ($queues as $queue) {
                $this->queueNames[] = $queue->getName();
            }
        }

        while ($thisTimeout > 0) {
            $this->migrateExpiredMessages($this->queueNames);

            if (false == $result = $this->getContext()->getRedis()->brpop($this->queueNames, $thisTimeout)) {
                return null;
            }

            $this->pushQueueNameBack($result->getKey());

            if ($message = $this->processResult($result, $redeliveryDelay)) {
                return $message;
            }

            $thisTimeout -= time() - $startAt;
        }

        return null;
    }
843668
public function timeout($timeout = null)
    {
        $this->timeout = (null === $timeout) ? $this->defaultTimeout : $timeout;
        return $this;
    }
346294
private function createAmqpQueue($queueName)
    {
        /** @var AmqpContext $context */
        $context = $this->context;

        $queue = $context->createQueue($queueName);
        $queue->addFlag(AmqpQueue::FLAG_DURABLE);

        return $queue;
    }
712665
@Override
    public java.util.concurrent.Future<CreateQueueResult> createQueueAsync(String queueName) {

        return createQueueAsync(new CreateQueueRequest().withQueueName(queueName));
    }