Synchronously returns message if there is any waiting in the queue.

@param string $queue
@param bool $noAck
@return Message|PromiseInterface
777623
public function get($queue = "", $noAck = false)
    {
        if ($this->getDeferred !== null) {
            throw new ChannelException("Another 'basic.get' already in progress. You should use 'basic.consume' instead of multiple 'basic.get'.");
        }

        $response = $this->getImpl($queue, $noAck);

        if ($response instanceof PromiseInterface) {
            $this->getDeferred = new Deferred();

            $response->done(function ($frame) {
                if ($frame instanceof MethodBasicGetEmptyFrame) {
                    // deferred has to be first nullified and then resolved, otherwise results in race condition
                    $deferred = $this->getDeferred;
                    $this->getDeferred = null;
                    $deferred->resolve(null);

                } elseif ($frame instanceof MethodBasicGetOkFrame) {
                    $this->getOkFrame = $frame;
                    $this->state = ChannelStateEnum::AWAITING_HEADER;

                } else {
                    throw new \LogicException("This statement should never be reached.");
                }
            });

            return $this->getDeferred->promise();

        } elseif ($response instanceof MethodBasicGetEmptyFrame) {
            return null;

        } elseif ($response instanceof MethodBasicGetOkFrame) {
            $this->state = ChannelStateEnum::AWAITING_HEADER;

            $headerFrame = $this->getClient()->awaitContentHeader($this->getChannelId());
            $this->headerFrame = $headerFrame;
            $this->bodySizeRemaining = $headerFrame->bodySize;
            $this->state = ChannelStateEnum::AWAITING_BODY;

            while ($this->bodySizeRemaining > 0) {
                $bodyFrame = $this->getClient()->awaitContentBody($this->getChannelId());

                $this->bodyBuffer->append($bodyFrame->payload);
                $this->bodySizeRemaining -= $bodyFrame->payloadSize;

                if ($this->bodySizeRemaining < 0) {
                    $this->state = ChannelStateEnum::ERROR;
                    $this->client->disconnect(Constants::STATUS_SYNTAX_ERROR, $errorMessage = "Body overflow, received " . (-$this->bodySizeRemaining) . " more bytes.");
                    throw new ChannelException($errorMessage);
                }
            }

            $this->state = ChannelStateEnum::READY;

            $message = new Message(
                null,
                $response->deliveryTag,
                $response->redelivered,
                $response->exchange,
                $response->routingKey,
                $this->headerFrame->toArray(),
                $this->bodyBuffer->consume($this->bodyBuffer->getLength())
            );

            $this->headerFrame = null;

            return $message;

        } else {
            throw new \LogicException("This statement should never be reached.");
        }
    }
580083
public function addMessageToQueue($message, $queue)
    {
        $this->queue->sendMessage([
            'QueueUrl' => $this->getQueueURL($queue),
            'MessageBody' => $message,
        ]);
    }
583541
public function get(string $queue = 'default')
    {
        $message = yield $this->kafka->consume();
        if (!is_object($message)) {
            return false;
        }

        if ($message->err === RD_KAFKA_RESP_ERR_NO_ERROR) {//正常
            return $message->payload;
        } elseif ($message->err === RD_KAFKA_RESP_ERR__PARTITION_EOF) {//暂时没有消息
            return null;
        } elseif ($message->err === RD_KAFKA_RESP_ERR__TIMED_OUT) {//超时
            return false;
        } else {
            throw new \RdKafka\Exception('RD_KAFKA_RESP_ERR', $message->err);//其他异常
        }
    }
1166680
public function createMessage($message, $queueId)
    {
        $queueMessage = new QueueMessage();
        $queueMessage->setMessage($message);
        $queueMessage->setQueueId($queueId);
        return $queueMessage;
    }
25737
public function push(QueueMessage $message)
    {
        $queue = $message->getQueueName() ?: $this->queue;

        if (!$message->getId()) {
            $message->set('attempts', 0);
            $message->set('queue', $queue);
            $message->set('id', Resque::generateJobId());
            $message->set('class', static::JOB_CLASS);
            $message->setId($message->getId());
        }

        $delay = $message->getDelay();

        $data = json_decode(json_encode($message), true);

        if ($delay > 0) {
            static::supportDelayed(true);

            \ResqueScheduler::delayedPush(time() + $delay, $data);
        } else {
            Resque::push($queue, $data);
        }

        return $message->getId();
    }
606446
private function hasRunningJobInQueue($queue)
    {
        $running = $this->configurationManager->findByQueueAndState($queue, JobState::STATE_RUNNING);

        return null !== $running;
    }
873573
public function consumeMessages($queue, $callback, $flags = AMQP_NOPARAM)
    {
        if (!isset($this->_queues[$queue])) {
            throw new InvalidKeyException(['consume message from queue failed: `:queue queue is NOT exists`', 'queue' => $queue]);
        }

        try {
            $this->_queues[$queue]->consume(function (\AMQPEnvelope $envelope) use ($callback, $queue) {
                return $callback(new Message($this, $queue, $envelope));
            }, $flags);
        } catch (\Exception $e) {
            throw new AmqpException('consume `:queue` queue message failed: ', $e->getMessage());
        }
    }
1199148
public function pop($queue = null)
    {
        $queue = $queue ?: $this->queue;

        $now = new \DateTimeImmutable('now');

        $query = $this->db->getQuery(true);

        $query->select('*')
            ->from($query->quoteName($this->table))
            ->where('queue = %q', $queue)
            ->where('visibility <= %q', $now->format('Y-m-d H:i:s'))
            ->orWhere(
                function (Query $query) use ($now) {
                    $query->where('reserved IS NULL')
                        ->where('reserved < %q', $now->modify('-' . $this->timeout . 'seconds')->format('Y-m-d H:i:s'));
                }
            );

        $trans = $this->db->getTransaction()->start();

        try {
            $data = $this->db->setQuery($query . ' FOR UPDATE')->loadOne('assoc');

            if (!$data) {
                return null;
            }

            $data['attempts']++;

            $values = ['reserved' => $now->format('Y-m-d H:i:s'), 'attempts' => $data['attempts']];

            $this->db->getWriter()->updateBatch($this->table, $values, ['id' => $data['id']]);

            $trans->commit();
        } catch (\Throwable $t) {
            $trans->rollback();
        }

        $message = new QueueMessage();

        $message->setId($data['id']);
        $message->setAttempts($data['attempts']);
        $message->setBody(json_decode($data['body'], true));
        $message->setRawBody($data['body']);
        $message->setQueueName($queue);

        return $message;
    }
67990
public function release(QueueMessage $message)
    {
        $this->client->release(
            new Job($message->getId(), ''),
            PheanstalkInterface::DEFAULT_PRIORITY,
            $message->getDelay()
        );

        return $this;
    }
387098
public function waitUntil($script, $seconds = null, $message = null)
    {
        if (! Str::startsWith($script, 'return ')) {
            $script = 'return '.$script;
        }

        if (! Str::endsWith($script, ';')) {
            $script = $script.';';
        }

        return $this->waitUsing($seconds, 100, function () use ($script) {
            return $this->driver->executeScript($script);
        }, $message);
    }
210240
public function awaitContentHeader($channel)
    {
        if ($this instanceof Async\Client) {
            $deferred = new Deferred();
            $this->addAwaitCallback(function ($frame) use ($deferred, $channel) {
                if ($frame instanceof Protocol\ContentHeaderFrame && $frame->channel === $channel) {
                    $deferred->resolve($frame);
                    return true;
                } elseif ($frame instanceof Protocol\MethodChannelCloseFrame && $frame->channel === $channel) {
                    $this->channelCloseOk($channel)->done(function () use ($frame, $deferred) {
                        $deferred->reject(new ClientException($frame->replyText, $frame->replyCode));
                    });
                    return true;
                } elseif ($frame instanceof Protocol\MethodConnectionCloseFrame) {
                    $this->connectionCloseOk()->done(function () use ($frame, $deferred) {
                        $deferred->reject(new ClientException($frame->replyText, $frame->replyCode));
                    });
                    return true;
                }
                return false;
            });
            return $deferred->promise();
        } else {
            for (;;) {
                while (($frame = $this->getReader()->consumeFrame($this->getReadBuffer())) === null) {
                    $this->feedReadBuffer();
                }
                if ($frame instanceof Protocol\ContentHeaderFrame && $frame->channel === $channel) {
                    return $frame;
                } elseif ($frame instanceof Protocol\MethodChannelCloseFrame && $frame->channel === $channel) {
                    $this->channelCloseOk($channel);
                    throw new ClientException($frame->replyText, $frame->replyCode);
                } elseif ($frame instanceof Protocol\MethodConnectionCloseFrame) {
                    $this->connectionCloseOk();
                    throw new ClientException($frame->replyText, $frame->replyCode);
                } else {
                    $this->enqueue($frame);
                }
            }
        }
        throw new \LogicException('This statement should be never reached.');
    }
48706
public function getMessagesTotalCountOnQueue($queue)
    {
        try {
            return $this->queue->statsTube($queue)['total-jobs'];
        } catch (ConnectionException $ex) {
            \PHPUnit\Framework\Assert::fail("queue [$queue] not found");
        }
    }
1033142
public function push(QueueMessage $message)
    {
        $queue = $message->getQueueName() ?: $this->queue;

        $this->queueDeclare($queue);

        $options = [
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
        ];

        if ($message->getDelay() > 0) {
            $options['application_headers'] = new AMQPTable(
                [
                    'x-delay' => $message->getDelay(),
                ]
            );
        }

        $msg = new AMQPMessage(json_encode($message), $options);

        $this->channel->basic_publish($msg, '', $queue);

        return 1;
    }
1206491
private function deferredDelivery(OutboundPackage $package): Promise
    {
        $deferred = new Deferred();

        /**
         * @psalm-suppress MixedTypeCoercion Incorrect amphp types
         * @psalm-suppress InvalidArgument
         */
        Loop::defer(
            function() use ($package, $deferred): \Generator
            {
                try
                {
                    yield call(
                        $this->deliveryRetryHandler,
                        function() use ($package): \Generator
                        {
                            yield $this->transport->send($package);
                        },
                        SendMessageFailed::class
                    );

                    $deferred->resolve();
                }
                catch (\Throwable $throwable)
                {
                    $deferred->fail($throwable);
                }
            }
        );

        return $deferred->promise();
    }
913486
protected function getNextJob( $connection, $queue )
	{
		if ( is_null( $queue ) )
			return $connection->pop();

		foreach ( explode( ',', $queue ) as $queue )
			if ( !is_null( $job = $connection->pop( $queue ) ) )
				return $job;

		return null;
	}
952894
public String waitAck(){
		try {
			this.latch.await(5000, TimeUnit.MILLISECONDS);
			return consumerGroup;
		} catch (Exception e) {} 
		return null;
	}
634667
public function pop($queue)
    {
        $item = $this->getClient()->lpop($this->getKey(self::QUEUE_KEY . $queue));

        if (!$item) {
            return null;
        }

        return json_decode($item, true);
    }