Execute the actual query and process the response.

@param string $queueName

@return array|null
351560
private function doPopMessage($queueName)
    {
        $query = 'SELECT id, message FROM bernard_messages
                  WHERE queue = :queue AND visible = :visible
                  ORDER BY sentAt LIMIT 1 '.$this->connection->getDatabasePlatform()->getForUpdateSql();

        list($id, $message) = $this->connection->fetchArray($query, [
            'queue' => $queueName,
            'visible' => true,
        ]);

        if ($id) {
            $this->connection->update('bernard_messages', ['visible' => 0], compact('id'));

            return [$message, $id];
        }
    }
1294593
protected function execute_query()
    {
        $query = $this->database->getQueryCollection()->findQuery($this->strQuery);

        if ($query) {
            return $query->result();
        }

        return null;
    }
285620
public BlockingQueue<CollectionDocument> getQueue(String queueName) {
		Preconditions.checkArgument(registry.containsKey(queueName), "Queue %s does not exist", queueName);
		return registry.get(queueName);
	}
740412
public function get($name)
    {
        if (!array_key_exists($name, $this->queues)) {
            throw new \InvalidArgumentException("The queue does not exist. {$name}");
        }

        return $this->queues[$name];
    }
694091
protected function exec($query)
    {
        try {
            Connect::getConn()->exec($query);
        } catch (Exception $e) {
            static::$response["status"] = "error";
            static::$response["response"] = $e->getMessage();
            static::$response["code"] = $e->getCode();

            return static::terminate(static::$response);
        }
        return null;
    }
449549
protected function flushQueueItem(string $connectionName, MessageEnvelopeQueue $queue)
    {
        if ($queue->isEmpty()) {
            return;
        }

        $connection = $this->getConnectionHandler()->getConnection($connectionName);
        /* @var AppleConnection $connection */
        foreach ($queue->getQueue() as $item) {
            /* @var AppleMessageEnvelope $item */
            $connection->send($item->getMessage());
        }

        $response = $this->getResponseData($connection);
        if ($response->isSuccess()) {
            $queue->traverseWith(function (MessageEnvelope $messageEnvelope) {
                $messageEnvelope->setState(MessageEnvelope::SUCCESS);
            });
            $queue->clear();
        } else {
            $this->handleErrorResponse($connectionName, $queue, $response);
        }
    }
49552
public function executeBatch($batch, ZendQueue $queue)
    {
        foreach ($batch as $item) {
            $this->processQueueItem($item);
            $queue->deleteMessage($item);
        }
    }
918982
public boolean isQueue(String queueName) throws Exception {
        Address addr = Address.root().add(SUBSYSTEM, SUBSYSTEM_MESSAGING, HORNETQ_SERVER, "default");
        String haystack = JMS_QUEUE;
        return null != findNodeInList(addr, haystack, queueName);
    }
128763
public function createQueue($queueName)
    {
        if ($this->queueExists($queueName)) {
            return;
        }

        $parameters = [
            'QueueName' => $queueName,
        ];

        if ($this->isFifoQueue($queueName)) {
            $parameters['Attributes'] = [
                'FifoQueue' => 'true',
            ];
        }

        $result = $this->sqs->createQueue($parameters);

        $this->queueUrls[$queueName] = $result['QueueUrl'];
    }
988263
public function getPropertyByQueryName($queryName)
    {
        return isset($this->propertiesByQueryName[$queryName]) ? $this->propertiesByQueryName[$queryName] : null;
    }
515748
public function findServerForQueue ($queueName, $serverTag) 
    {
        if (isset($this->serversByQueues[$queueName])) {
            return $this->serversByQueues[$queueName];
        }
        
        if (isset($this->serversByTags[$serverTag])) {
            return $this->serversByTags[$serverTag];
        }
        
        return null;
    }
120085
public static function processQueue()
    {
        if (count(self::$request_queue) == 0) {
            return false;
        }

        do {
            $multi_handle = curl_multi_exec(self::$queue_handle, $active);
        } while ($multi_handle == CURLM_CALL_MULTI_PERFORM);

        while ($active && $multi_handle == CURLM_OK) {
            if (curl_multi_select(self::$queue_handle) != -1) {
                do {
                    $multi_handle = curl_multi_exec(self::$queue_handle, $active);
                } while ($multi_handle == CURLM_CALL_MULTI_PERFORM);
            }
        }

        foreach (self::$request_queue as $key => $value) {
            $error = curl_error($value['handle']);
            if (!empty($error)) {
                $info = array(
                    'is_success'  => false,
                    'url'         => $value['url'],
                    'errorno'     => curl_errno($value['handle']),
                    'error'       => $error,
                    'queue'       => 'ON'
                );

                $result = false;
                $header = '';
            } else {
                $info = curl_getinfo($value['handle']);
                $info['is_success'] = true;
                $info['queue'] = 'ON';

                $body = curl_multi_getcontent($value['handle']);
                list($header, $result) = explode("\r\n\r\n", $body, 2);
            }

            self::logProfiling($info);
            self::$request_queue[$key]['response'] = new Response($info, $result, $header);
            $callback = $value['callback'];
            $callback(self::$request_queue[$key]['response']);

            curl_multi_remove_handle(self::$queue_handle, self::$request_queue[$key]['handle']);
        }

        curl_multi_close(self::$queue_handle);
        self::$request_queue = array();
        return true;
    }
1026329
private function resolveUrl($queueName)
    {
        if (isset($this->queueUrls[$queueName])) {
            return $this->queueUrls[$queueName];
        }

        $result = $this->sqs->getQueueUrl(['QueueName' => $queueName]);

        if ($result && $queueUrl = $result->get('QueueUrl')) {
            return $this->queueUrls[$queueName] = $queueUrl;
        }

        throw new \InvalidArgumentException('Queue "'.$queueName.'" cannot be resolved to an url.');
    }
1087455
protected function get($queueName)
    {
        if (isset($this->caches[$queueName])) {
            return $this->caches[$queueName];
        }

        return $this->caches[$queueName] = new \SplQueue();
    }
1047849
public function execute($domain = null)
    {
        $response = $this->client->post(
            $this->url,
            [
                'form_params' => [
                    'login' => $this->login,
                    'passwd' => $this->password,
                    'label' => $domain,
                    'type' => 'standard',
                    'origin_scheme' => 'http',
                    'origin_url' => 'hosted.adback.co',
                    'cname' => $domain,
                ]
            ]
        );

        $data = json_decode($response->getBody()->getContents(), true);

        if ('ok' != $data['status']) {
            throw new QueryErrorException($data['description']);
        }

        return $data['cdnResource']['cdn_url'];
    }
1071665
public function peekMessage(string $queueName, $amount=1){
        if(!$queueName || !$amount){
            return $this->error('参数有误');
        }
        try{
            $responseData = [];
            $queue = $this->client->getQueueRef($queueName);
            if($amount > 1){
                $response = $queue->batchPeekMessage($amount);
                foreach ($response->getMessages() as $messageObject){
                    $responseData[] = $this->parseObjectData(Message::class, $messageObject);
                }
            }else{
                $response = $queue->peekMessage();
                $responseData = $this->parseObjectData(PeekMessageResponse::class, $response);
            }
            return $this->success('查看消息成功', $responseData);
        }catch (MnsException $exception){
            return $this->parseException($exception, '查看消息');
        }
    }
1019318
protected function runQuery($tableName, $query)
    {
        try {
            return $this->connection->query($query);
        } catch (DBALException $e) {
            $this->loggerSystem->critical($e->getMessage(), ['event' => 'exception', 'exception' => $e]);
            $this->loggerFlash->error(
                Trans::__('An error occured while updating `%TABLE%`. The error is %ERROR%',
                ['%TABLE%' => $tableName, '%ERROR%' => $e->getMessage()]
            ));
        }

        return null;
    }