diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 8dadc45d9..9f2e1b2c8 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -30,7 +30,7 @@ jobs: strategy: fail-fast: false matrix: - php: [ '8.1', '8.2' ] + php: [ '8.2', '8.3' ] steps: - name: Checkout. uses: actions/checkout@v2 diff --git a/.github/workflows/static.yml b/.github/workflows/static.yml index be42e8f9a..5126dc889 100644 --- a/.github/workflows/static.yml +++ b/.github/workflows/static.yml @@ -32,7 +32,7 @@ jobs: strategy: fail-fast: false matrix: - php: [ '8.1', '8.2' ] + php: [ '8.2', '8.3', '8.4' ] steps: - name: Checkout. uses: actions/checkout@v2 diff --git a/CHANGELOG.md b/CHANGELOG.md index a4c06012b..4bdf0fc75 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,10 +1,16 @@ Yii2 Queue Extension Change Log =============================== -2.3.7 under development +2.3.8 under development ----------------------- +- Enh #516: Ensure Redis driver messages are consumed at least once (soul11201) +- Bug #522: Fix SQS driver type error with custom value passed to `queue/listen` (flaviovs) -- no changes in this release. + +2.3.7 April 29, 2024 +-------------------- + +- Enh #509: Add StatisticsProviderInterface to get statistics from queue (kalmer) 2.3.6 October 03, 2023 diff --git a/Makefile b/Makefile index 7f260cc75..c01e40b11 100644 --- a/Makefile +++ b/Makefile @@ -1,35 +1,35 @@ help: ## Display help information @fgrep -h "##" $(MAKEFILE_LIST) | fgrep -v fgrep | sed -e 's/\\$$//' | sed -e 's/##//' -build: ## Build an image from a docker-compose file. Params: {{ v=8.1 }}. Default latest PHP 8.1 +build: ## Build an image from a docker-compose file. Params: {{ v=8.2 }}. Default latest PHP 8.2 @cp -n .env.example .env - PHP_VERSION=$(filter-out $@,$(v)) docker-compose up -d --build + PHP_VERSION=$(filter-out $@,$(v)) docker compose up -d --build make create-sqs-queue make create-sqs-fifo-queue -test: ## Run tests. Params: {{ v=8.1 }}. Default latest PHP 8.1 +test: ## Run tests. Params: {{ v=8.2 }}. Default latest PHP 8.2 make build - PHP_VERSION=$(filter-out $@,$(v)) docker-compose run yii2-queue-php vendor/bin/phpunit --coverage-clover coverage.xml + PHP_VERSION=$(filter-out $@,$(v)) docker compose run yii2-queue-php vendor/bin/phpunit --coverage-clover coverage.xml make down down: ## Stop and remove containers, networks - docker-compose down + docker compose down -benchmark: ## Run benchmark. Params: {{ v=8.1 }}. Default latest PHP 8.1 - PHP_VERSION=$(filter-out $@,$(v)) docker-compose build --pull yii2-queue-php - PHP_VERSION=$(filter-out $@,$(v)) docker-compose run yii2-queue-php tests/yii benchmark/waiting +benchmark: ## Run benchmark. Params: {{ v=8.2 }}. Default latest PHP 8.2 + PHP_VERSION=$(filter-out $@,$(v)) docker compose build --pull yii2-queue-php + PHP_VERSION=$(filter-out $@,$(v)) docker compose run yii2-queue-php tests/yii benchmark/waiting make down sh: ## Enter the container with the application docker exec -it yii2-queue-php sh -static-analyze: ## Run code static analyze. Params: {{ v=8.1 }}. Default latest PHP 8.1 - PHP_VERSION=$(filter-out $@,$(v)) docker-compose build --pull yii2-queue-php - PHP_VERSION=$(filter-out $@,$(v)) docker-compose run yii2-queue-php vendor/bin/psalm --config=psalm.xml --shepherd --stats --php-version=$(v) +static-analyze: ## Run code static analyze. Params: {{ v=8.2 }}. Default latest PHP 8.2 + PHP_VERSION=$(filter-out $@,$(v)) docker compose build --pull yii2-queue-php + PHP_VERSION=$(filter-out $@,$(v)) docker compose run yii2-queue-php vendor/bin/psalm --config=psalm.xml --shepherd --stats --php-version=$(v) make down clean: - docker-compose down + docker compose down rm -rf tests/runtime/* rm -f .php_cs.cache rm -rf composer.lock diff --git a/README.md b/README.md index a28bd8f40..e336b02c2 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ Documentation is at [docs/guide/README.md](docs/guide/README.md). ## Requirements -- PHP 8.1 or higher. +- PHP 8.2 or higher. Installation ------------ diff --git a/composer.json b/composer.json index 0da6de72b..be729e785 100644 --- a/composer.json +++ b/composer.json @@ -1,6 +1,6 @@ { "name": "yiisoft/yii2-queue", - "description": "Yii2 Queue Extension which supported DB, Redis, RabbitMQ, Beanstalk, SQS and Gearman", + "description": "Yii2 Queue Extension which supports queues based on DB, Redis, RabbitMQ, Beanstalk, SQS, and Gearman", "type": "yii2-extension", "keywords": ["yii", "queue", "async", "gii", "db", "redis", "rabbitmq", "beanstalk", "gearman", "sqs"], "license": "BSD-3-Clause", @@ -16,9 +16,9 @@ "docs": "https://github.com/yiisoft/yii2-queue/blob/master/docs/guide" }, "require": { - "php": ">=8.1", - "yiisoft/yii2": "~2.0.14", - "symfony/process": "^6.3", + "php": ">=8.2", + "yiisoft/yii2": "~2.0.50", + "symfony/process": "^7.0", "laravel/serializable-closure": "^v1.3.0" }, "require-dev": { @@ -33,7 +33,7 @@ "enqueue/stomp": "^0.10.0", "pda/pheanstalk": "^5.0.0", "aws/aws-sdk-php": "3.285.0", - "vimeo/psalm": "^5.10.0" + "vimeo/psalm": "^6.0.0" }, "suggest": { "ext-pcntl": "Need for process signals.", diff --git a/docs/guide-zh-CN/retryable.md b/docs/guide-zh-CN/retryable.md index 738cb1916..34f139e98 100644 --- a/docs/guide-zh-CN/retryable.md +++ b/docs/guide-zh-CN/retryable.md @@ -1,10 +1,10 @@ 错误与重复执行 ========================= -在作业处理期间可以抛出异常。 当请求的服务和外部资源不可用时，由于代码编写的比较糟糕而导致的内部错误。 -在第二种情况下，可以在一段时间后重新尝试这个作业。 +作业的执行可能会失败。这可能是由于内部错误造成的，这些错误是由编写不当的代码导致的，应该首先修复。但是，它们也可能由于外部问题（例如服务或资源不可用）而失败。这可能会导致异常或超时。 +在后一种情况下，最好能够在一段时间后重试作业。有几种方法可以做到这一点。 -有几种方法可以做到这一点。 +>注意：下面描述的 ttr 功能需要安装 [PHP Process Control （pcntl）](https://www.php.net/manual/en/book.pcntl.php) 扩展，并且 worker 命令必须使用 --isolate 选项（默认启用）。 重试选项 ------------- @@ -21,7 +21,12 @@ ], ``` - `ttr` 选项设置了在队列中保留工作的时间。如果一份作业在这段时间没有执行，它将返回队列进行重试。 + `ttr` 选项设置了作业必须成功完成的秒数。因此，可能会发生两件事来使工作失败: + 1. 作业在`ttr`结束之前发生异常 + 2. 完成作业所需的时间将比 ttr 长（超时），工作线程停止作业执行 + +在这两种情况下，作业都将发送回队列进行重试。但请注意，在第一种情况下，即使作业在运行后立即停止，也不会立即重试，要等到`ttr`"用完"。也就是说，在将作业发送回队列之前，必须经过剩余的 ttr 秒数。 + `attempts` 选项设置了最大的尝试次数。如果尝试已经结束，作业作还没有完成，它将从队列中移除。 这种将全局设置队列中的所有作业，如果您需要为多个作业进行不同的设置可以使用， @@ -30,7 +35,7 @@ 重试作业接口 ---------------------- -Separate control of retry is implemented by `RetryableJobInterface` 接口。 示例: +为了更好地控制重试逻辑，作业可以实现 `RetryableJobInterface` 接口。 示例: ```php class SomeJob extends BaseObject implements RetryableJobInterface @@ -85,9 +90,9 @@ Yii::$app->queue->on(Queue::EVENT_AFTER_ERROR, function (ExecEvent $event) { 限制 ------------ -完全支持 [Beanstalk], [DB], [File] 和 [Redis] 驱动程序的重试工具。 -[Sync] 驱动不会重试失败的工作 [Gearman] 不支持重试。 -[RabbitMQ] 基本版支持，但重试编号无法得到。 +[Beanstalk], [DB], [File] 和 [Redis] 驱动程序实现了对可重试作业的完全支持。 +[Sync] 驱动不会重试失败的作业 [Gearman] 不支持重试作业。 +[RabbitMQ] 只有其基本的可重试支持，其中无法设置尝试次数。 [Beanstalk]: driver-beanstalk.md [DB]: driver-db.md @@ -95,4 +100,4 @@ Yii::$app->queue->on(Queue::EVENT_AFTER_ERROR, function (ExecEvent $event) { [Redis]: driver-redis.md [Sync]: driver-sync.md [Gearman]: driver-gearman.md -[RabbitMQ]: driver-amqp.md \ No newline at end of file +[RabbitMQ]: driver-amqp.md diff --git a/psalm.xml b/psalm.xml index d92109732..ec0f9676e 100644 --- a/psalm.xml +++ b/psalm.xml @@ -27,5 +27,19 @@ + + + + + + + + + + + + + + diff --git a/src/InvalidJobException.php b/src/InvalidJobException.php index ae345f1d2..eb7988fb3 100644 --- a/src/InvalidJobException.php +++ b/src/InvalidJobException.php @@ -33,7 +33,7 @@ public function __construct( private readonly string $serialized, string $message = '', int $code = 0, - Throwable $previous = null + ?Throwable $previous = null ) { parent::__construct($message, $code, $previous); diff --git a/src/JobEvent.php b/src/JobEvent.php index fab622150..31968b09c 100644 --- a/src/JobEvent.php +++ b/src/JobEvent.php @@ -27,7 +27,7 @@ abstract class JobEvent extends Event /** * @var Queue * @inheritdoc - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public $sender; /** diff --git a/src/cli/InfoAction.php b/src/cli/InfoAction.php new file mode 100644 index 000000000..7f3ed669c --- /dev/null +++ b/src/cli/InfoAction.php @@ -0,0 +1,68 @@ + + * + * @property Controller $controller + */ +class InfoAction extends Action +{ + /** + * @var Queue + */ + public Queue $queue; + + /** + * Info about queue status. + */ + public function run(): void + { + if (!($this->queue instanceof StatisticsProviderInterface)) { + throw new NotSupportedException('Queue does not support ' . StatisticsProviderInterface::class); + } + + $this->controller->stdout('Jobs' . PHP_EOL, Console::FG_GREEN); + $statisticsProvider = $this->queue->getStatisticsProvider(); + + if ($statisticsProvider instanceof WaitingCountInterface) { + $this->controller->stdout('- waiting: ', Console::FG_YELLOW); + $this->controller->stdout($statisticsProvider->getWaitingCount() . PHP_EOL); + } + + if ($statisticsProvider instanceof DelayedCountInterface) { + $this->controller->stdout('- delayed: ', Console::FG_YELLOW); + $this->controller->stdout($statisticsProvider->getDelayedCount() . PHP_EOL); + } + + if ($statisticsProvider instanceof ReservedCountInterface) { + $this->controller->stdout('- reserved: ', Console::FG_YELLOW); + $this->controller->stdout($statisticsProvider->getReservedCount() . PHP_EOL); + } + + if ($statisticsProvider instanceof DoneCountInterface) { + $this->controller->stdout('- done: ', Console::FG_YELLOW); + $this->controller->stdout($statisticsProvider->getDoneCount() . PHP_EOL); + } + } +} diff --git a/src/cli/VerboseBehavior.php b/src/cli/VerboseBehavior.php index e4dd27613..f9ea59a55 100644 --- a/src/cli/VerboseBehavior.php +++ b/src/cli/VerboseBehavior.php @@ -31,7 +31,6 @@ class VerboseBehavior extends Behavior public $owner; /** * @var Controller - * @psalm-suppress PropertyNotSetInConstructor */ public Controller $command; diff --git a/src/cli/WorkerEvent.php b/src/cli/WorkerEvent.php index 6dc4bf3c6..530c8ebd5 100644 --- a/src/cli/WorkerEvent.php +++ b/src/cli/WorkerEvent.php @@ -27,12 +27,11 @@ class WorkerEvent extends Event /** * @var Queue * @inheritdoc - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public $sender; /** * @var LoopInterface - * @psalm-suppress PropertyNotSetInConstructor */ public LoopInterface $loop; /** diff --git a/src/closure/Behavior.php b/src/closure/Behavior.php index 75b051181..f6a2b1dd4 100644 --- a/src/closure/Behavior.php +++ b/src/closure/Behavior.php @@ -33,7 +33,7 @@ class Behavior extends \yii\base\Behavior { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public $owner; diff --git a/src/debug/Panel.php b/src/debug/Panel.php index 2dcabeb8d..0d63cfddd 100644 --- a/src/debug/Panel.php +++ b/src/debug/Panel.php @@ -24,7 +24,6 @@ * Debug Panel. * * @author Roman Zhuravlev - * @psalm-suppress PropertyNotSetInConstructor */ class Panel extends BasePanel implements ViewContextInterface { diff --git a/src/drivers/amqp_interop/Command.php b/src/drivers/amqp_interop/Command.php index 1dcd2b5cd..b24f1cc25 100644 --- a/src/drivers/amqp_interop/Command.php +++ b/src/drivers/amqp_interop/Command.php @@ -23,7 +23,7 @@ class Command extends CliCommand { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public CliQueue $queue; diff --git a/src/drivers/beanstalk/Command.php b/src/drivers/beanstalk/Command.php index ebb54b4bf..2df4e720a 100644 --- a/src/drivers/beanstalk/Command.php +++ b/src/drivers/beanstalk/Command.php @@ -23,7 +23,7 @@ class Command extends CliCommand { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantPropertyType, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantPropertyType, NonInvariantDocblockPropertyType */ public CliQueue $queue; /** diff --git a/src/drivers/db/Command.php b/src/drivers/db/Command.php index 41ca3f434..4978bdd7f 100644 --- a/src/drivers/db/Command.php +++ b/src/drivers/db/Command.php @@ -13,6 +13,7 @@ use yii\console\Exception; use yii\queue\cli\Command as CliCommand; use yii\queue\cli\Queue as CliQueue; +use yii\queue\cli\InfoAction; /** * Manages application db-queue. @@ -23,7 +24,7 @@ class Command extends CliCommand { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public CliQueue $queue; /** diff --git a/src/drivers/db/InfoAction.php b/src/drivers/db/InfoAction.php index dbc5b4f7a..1280fa9a0 100644 --- a/src/drivers/db/InfoAction.php +++ b/src/drivers/db/InfoAction.php @@ -20,13 +20,15 @@ /** * Info about queue status. * + * @deprecated Will be removed in 3.0. Use yii\queue\cli\InfoAction instead. + * * @author Roman Zhuravlev */ class InfoAction extends Action { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public CliQueue $queue; diff --git a/src/drivers/db/Queue.php b/src/drivers/db/Queue.php index a98f88a4c..c26ad79b1 100644 --- a/src/drivers/db/Queue.php +++ b/src/drivers/db/Queue.php @@ -17,13 +17,17 @@ use yii\di\Instance; use yii\mutex\Mutex; use yii\queue\cli\Queue as CliQueue; +use yii\queue\interfaces\StatisticsInterface; +use yii\queue\interfaces\StatisticsProviderInterface; /** * Db Queue. * + * @property-read StatisticsProvider $statisticsProvider + * * @author Roman Zhuravlev */ -class Queue extends CliQueue +class Queue extends CliQueue implements StatisticsProviderInterface { /** * @var Connection|array|string @@ -86,8 +90,8 @@ public function run(bool $repeat, int $timeout = 0) if ($this->handleMessage( $payload['id'], $payload['job'], - (int)$payload['ttr'], - (int)$payload['attempt'] + (int) $payload['ttr'], + (int) $payload['attempt'] )) { $this->release($payload); } @@ -281,4 +285,17 @@ private function getMutex(): Mutex } return $mutex; } + + private StatisticsInterface $_statisticsProvider; + + /** + * @return StatisticsInterface + */ + public function getStatisticsProvider(): StatisticsInterface + { + if (!isset($this->_statisticsProvider)) { + $this->_statisticsProvider = new StatisticsProvider($this); + } + return $this->_statisticsProvider; + } } diff --git a/src/drivers/db/StatisticsProvider.php b/src/drivers/db/StatisticsProvider.php new file mode 100644 index 000000000..c5d0f3a9b --- /dev/null +++ b/src/drivers/db/StatisticsProvider.php @@ -0,0 +1,98 @@ + + */ +class StatisticsProvider extends BaseObject implements + DoneCountInterface, + WaitingCountInterface, + DelayedCountInterface, + ReservedCountInterface, + StatisticsInterface +{ + /** + * @var Queue + */ + protected Queue $queue; + + public function __construct(Queue $queue, array $config = []) + { + $this->queue = $queue; + parent::__construct($config); + } + + /** + * @inheritdoc + */ + public function getWaitingCount(): int + { + /** @psalm-var \yii\db\Connection $this->queue->db */ + return (int) (new Query()) + ->from($this->queue->tableName) + ->andWhere(['channel' => $this->queue->channel]) + ->andWhere(['reserved_at' => null]) + ->andWhere(['delay' => 0]) + ->count('*', $this->queue->db); + } + + /** + * @inheritdoc + */ + public function getDelayedCount(): int + { + /** @psalm-var \yii\db\Connection $this->queue->db */ + return (int) (new Query()) + ->from($this->queue->tableName) + ->andWhere(['channel' => $this->queue->channel]) + ->andWhere(['reserved_at' => null]) + ->andWhere(['>', 'delay', 0]) + ->count('*', $this->queue->db); + } + + /** + * @inheritdoc + */ + public function getReservedCount(): int + { + /** @psalm-var \yii\db\Connection $this->queue->db */ + return (int) (new Query()) + ->from($this->queue->tableName) + ->andWhere(['channel' => $this->queue->channel]) + ->andWhere('[[reserved_at]] is not null') + ->andWhere(['done_at' => null]) + ->count('*', $this->queue->db); + } + + /** + * @inheritdoc + */ + public function getDoneCount(): int + { + /** @psalm-var \yii\db\Connection $this->queue->db */ + return (int) (new Query()) + ->from($this->queue->tableName) + ->andWhere(['channel' => $this->queue->channel]) + ->andWhere('[[done_at]] is not null') + ->count('*', $this->queue->db); + } +} diff --git a/src/drivers/file/Command.php b/src/drivers/file/Command.php index 354c373c9..abe94f4d3 100644 --- a/src/drivers/file/Command.php +++ b/src/drivers/file/Command.php @@ -13,6 +13,7 @@ use yii\console\Exception; use yii\queue\cli\Command as CliCommand; use yii\queue\cli\Queue as CliQueue; +use yii\queue\cli\InfoAction; /** * Manages application file-queue. @@ -23,7 +24,7 @@ class Command extends CliCommand { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public CliQueue $queue; /** diff --git a/src/drivers/file/InfoAction.php b/src/drivers/file/InfoAction.php index c91d895fd..a2bc34d28 100644 --- a/src/drivers/file/InfoAction.php +++ b/src/drivers/file/InfoAction.php @@ -18,13 +18,15 @@ /** * Info about queue status. * + * @deprecated Will be removed in 3.0. Use yii\queue\cli\InfoAction instead. + * * @author Roman Zhuravlev */ class InfoAction extends Action { /** * @var Queue - * @psalm-suppress NonInvariantDocblockPropertyType, PropertyNotSetInConstructor + * @psalm-suppress NonInvariantDocblockPropertyType */ public CliQueue $queue; diff --git a/src/drivers/file/Queue.php b/src/drivers/file/Queue.php index 5a7ee35b4..2bfcade70 100644 --- a/src/drivers/file/Queue.php +++ b/src/drivers/file/Queue.php @@ -16,13 +16,17 @@ use yii\base\NotSupportedException; use yii\helpers\FileHelper; use yii\queue\cli\Queue as CliQueue; +use yii\queue\interfaces\StatisticsInterface; +use yii\queue\interfaces\StatisticsProviderInterface; /** * File Queue. * + * @property-read StatisticsProvider $statisticsProvider + * * @author Roman Zhuravlev */ -class Queue extends CliQueue +class Queue extends CliQueue implements StatisticsProviderInterface { /** * @var string @@ -257,6 +261,7 @@ protected function pushMessage(string $payload, int $ttr, int $delay, mixed $pri if ($priority !== null) { throw new NotSupportedException('Job priority is not supported in the driver.'); } + $id = 0; $this->touchIndex(function (array &$data) use ($payload, $ttr, $delay, &$id) { /** @var array{lastId: int, waiting: array, delayed: array} $data */ @@ -335,4 +340,17 @@ private function touchIndex(callable $callback): void fclose($file); } } + + private StatisticsInterface $_statisticsProvider; + + /** + * @return StatisticsInterface + */ + public function getStatisticsProvider(): StatisticsInterface + { + if (!isset($this->_statisticsProvider)) { + $this->_statisticsProvider = new StatisticsProvider($this); + } + return $this->_statisticsProvider; + } } diff --git a/src/drivers/file/StatisticsProvider.php b/src/drivers/file/StatisticsProvider.php new file mode 100644 index 000000000..27cc1a75f --- /dev/null +++ b/src/drivers/file/StatisticsProvider.php @@ -0,0 +1,96 @@ + + */ +class StatisticsProvider extends BaseObject implements + DoneCountInterface, + WaitingCountInterface, + DelayedCountInterface, + ReservedCountInterface, + StatisticsInterface +{ + /** + * @var Queue + */ + protected Queue $queue; + + public function __construct(Queue $queue, array $config = []) + { + $this->queue = $queue; + parent::__construct($config); + } + + /** + * @inheritdoc + */ + public function getWaitingCount(): int + { + /** @var array{waiting:array} $data */ + $data = $this->getIndexData(); + return !empty($data['waiting']) ? count($data['waiting']) : 0; + } + + /** + * @inheritdoc + */ + public function getDelayedCount(): int + { + /** @var array{delayed:array} $data */ + $data = $this->getIndexData(); + return !empty($data['delayed']) ? count($data['delayed']) : 0; + } + + /** + * @inheritdoc + */ + public function getReservedCount(): int + { + /** @var array{reserved:array} $data */ + $data = $this->getIndexData(); + return !empty($data['reserved']) ? count($data['reserved']) : 0; + } + + /** + * @inheritdoc + */ + public function getDoneCount(): int + { + /** @var array{lastId:int} $data */ + $data = $this->getIndexData(); + $total = isset($data['lastId']) ? $data['lastId'] : 0; + return $total - $this->getDelayedCount() - $this->getWaitingCount(); + } + + /** + * @psalm-suppress MissingReturnType + */ + protected function getIndexData() + { + $fileName = $this->queue->path . '/index.data'; + if (file_exists($fileName)) { + return call_user_func($this->queue->indexDeserializer, file_get_contents($fileName)); + } + + return []; + } +} diff --git a/src/drivers/gearman/Command.php b/src/drivers/gearman/Command.php index 3bf256cac..d21c191e4 100644 --- a/src/drivers/gearman/Command.php +++ b/src/drivers/gearman/Command.php @@ -22,7 +22,7 @@ class Command extends CliCommand { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public CliQueue $queue; diff --git a/src/drivers/redis/Command.php b/src/drivers/redis/Command.php index 0872e319c..50da9ac51 100644 --- a/src/drivers/redis/Command.php +++ b/src/drivers/redis/Command.php @@ -13,6 +13,7 @@ use yii\console\Exception; use yii\queue\cli\Command as CliCommand; use yii\queue\cli\Queue as CliQueue; +use yii\queue\cli\InfoAction; /** * Manages application redis-queue. @@ -23,7 +24,7 @@ class Command extends CliCommand { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public CliQueue $queue; /** diff --git a/src/drivers/redis/InfoAction.php b/src/drivers/redis/InfoAction.php index e052feb4a..97fe1fa1e 100644 --- a/src/drivers/redis/InfoAction.php +++ b/src/drivers/redis/InfoAction.php @@ -17,13 +17,15 @@ /** * Info about queue status. * + * @deprecated Will be removed in 3.0. Use yii\queue\cli\InfoAction instead. + * * @author Roman Zhuravlev */ class InfoAction extends Action { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public CliQueue $queue; diff --git a/src/drivers/redis/Queue.php b/src/drivers/redis/Queue.php index f9004c941..48232e3c0 100644 --- a/src/drivers/redis/Queue.php +++ b/src/drivers/redis/Queue.php @@ -14,14 +14,18 @@ use yii\base\NotSupportedException; use yii\di\Instance; use yii\queue\cli\Queue as CliQueue; +use yii\queue\interfaces\StatisticsInterface; +use yii\queue\interfaces\StatisticsProviderInterface; use yii\redis\Connection; /** * Redis Queue. * + * @property-read StatisticsProvider $statisticsProvider + * * @author Roman Zhuravlev */ -class Queue extends CliQueue +class Queue extends CliQueue implements StatisticsProviderInterface { /** * @var Connection|array|string @@ -182,10 +186,10 @@ protected function moveExpired(string $from): void { $now = time(); if ($expired = $this->redis->zrevrangebyscore($from, $now, '-inf')) { - $this->redis->zremrangebyscore($from, '-inf', $now); foreach ($expired as $id) { $this->redis->rpush("$this->channel.waiting", $id); } + $this->redis->zremrangebyscore($from, '-inf', $now); } } @@ -221,4 +225,17 @@ protected function pushMessage(string $payload, int $ttr, int $delay, mixed $pri return $id; } + + private StatisticsInterface $_statisticsProvider; + + /** + * @return StatisticsInterface + */ + public function getStatisticsProvider(): StatisticsInterface + { + if (!isset($this->_statisticsProvider)) { + $this->_statisticsProvider = new StatisticsProvider($this); + } + return $this->_statisticsProvider; + } } diff --git a/src/drivers/redis/StatisticsProvider.php b/src/drivers/redis/StatisticsProvider.php new file mode 100644 index 000000000..5df423f33 --- /dev/null +++ b/src/drivers/redis/StatisticsProvider.php @@ -0,0 +1,83 @@ + + */ +class StatisticsProvider extends BaseObject implements + DoneCountInterface, + WaitingCountInterface, + DelayedCountInterface, + ReservedCountInterface, + StatisticsInterface +{ + /** + * @var Queue + */ + protected Queue $queue; + + + public function __construct(Queue $queue, array $config = []) + { + $this->queue = $queue; + parent::__construct($config); + } + + /** + * @inheritdoc + */ + public function getWaitingCount(): int + { + $prefix = $this->queue->channel; + return (int) $this->queue->redis->llen("$prefix.waiting"); + } + + /** + * @inheritdoc + */ + public function getDelayedCount(): int + { + $prefix = $this->queue->channel; + return (int) $this->queue->redis->zcount("$prefix.delayed", '-inf', '+inf'); + } + + /** + * @inheritdoc + */ + public function getReservedCount(): int + { + $prefix = $this->queue->channel; + return (int) $this->queue->redis->zcount("$prefix.reserved", '-inf', '+inf'); + } + + /** + * @inheritdoc + */ + public function getDoneCount(): int + { + $prefix = $this->queue->channel; + $waiting = $this->getWaitingCount(); + $delayed = $this->getDelayedCount(); + $reserved = $this->getReservedCount(); + $total = (int) $this->queue->redis->get("$prefix.message_id"); + return $total - $waiting - $delayed - $reserved; + } +} diff --git a/src/drivers/sqs/Command.php b/src/drivers/sqs/Command.php index e5f90c490..9cb5850ad 100644 --- a/src/drivers/sqs/Command.php +++ b/src/drivers/sqs/Command.php @@ -24,7 +24,7 @@ class Command extends CliCommand { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public CliQueue $queue; diff --git a/src/drivers/stomp/Command.php b/src/drivers/stomp/Command.php index 71bbfd9d1..974a0d1c3 100644 --- a/src/drivers/stomp/Command.php +++ b/src/drivers/stomp/Command.php @@ -24,7 +24,7 @@ class Command extends CliCommand { /** * @var Queue - * @psalm-suppress PropertyNotSetInConstructor, NonInvariantDocblockPropertyType + * @psalm-suppress NonInvariantDocblockPropertyType */ public CliQueue $queue; diff --git a/src/interfaces/DelayedCountInterface.php b/src/interfaces/DelayedCountInterface.php new file mode 100644 index 000000000..fcf19fd22 --- /dev/null +++ b/src/interfaces/DelayedCountInterface.php @@ -0,0 +1,24 @@ + + */ +interface DelayedCountInterface +{ + /** + * @return int + */ + public function getDelayedCount(): int; +} diff --git a/src/interfaces/DoneCountInterface.php b/src/interfaces/DoneCountInterface.php new file mode 100644 index 000000000..14c0930b6 --- /dev/null +++ b/src/interfaces/DoneCountInterface.php @@ -0,0 +1,24 @@ + + */ +interface DoneCountInterface +{ + /** + * @return int + */ + public function getDoneCount(): int; +} diff --git a/src/interfaces/ReservedCountInterface.php b/src/interfaces/ReservedCountInterface.php new file mode 100644 index 000000000..984a442c8 --- /dev/null +++ b/src/interfaces/ReservedCountInterface.php @@ -0,0 +1,24 @@ + + */ +interface ReservedCountInterface +{ + /** + * @return int + */ + public function getReservedCount(): int; +} diff --git a/src/interfaces/StatisticsInterface.php b/src/interfaces/StatisticsInterface.php new file mode 100644 index 000000000..34cbfde77 --- /dev/null +++ b/src/interfaces/StatisticsInterface.php @@ -0,0 +1,18 @@ + + */ +interface StatisticsProviderInterface +{ + /** + * @return StatisticsInterface + */ + public function getStatisticsProvider(): StatisticsInterface; +} diff --git a/src/interfaces/WaitingCountInterface.php b/src/interfaces/WaitingCountInterface.php new file mode 100644 index 000000000..ec20be06c --- /dev/null +++ b/src/interfaces/WaitingCountInterface.php @@ -0,0 +1,24 @@ + + */ +interface WaitingCountInterface +{ + /** + * @return int + */ + public function getWaitingCount(): int; +} diff --git a/src/serializers/JsonSerializer.php b/src/serializers/JsonSerializer.php index 0483dcbfb..30010494d 100644 --- a/src/serializers/JsonSerializer.php +++ b/src/serializers/JsonSerializer.php @@ -29,7 +29,7 @@ class JsonSerializer extends BaseObject implements SerializerInterface /** * @var int */ - public $options = 0; + public int $options = 0; /** * @inheritdoc diff --git a/tests/app/config/main.php b/tests/app/config/main.php index cf47d8faf..66f104ed0 100644 --- a/tests/app/config/main.php +++ b/tests/app/config/main.php @@ -62,6 +62,7 @@ 'class' => MysqlMutex::class, 'db' => 'mysql', ], + 'deleteReleased' => false, ], 'sqlite' => [ 'class' => Connection::class, @@ -71,6 +72,7 @@ 'class' => DbQueue::class, 'db' => 'sqlite', 'mutex' => FileMutex::class, + 'deleteReleased' => false, ], 'pgsql' => [ 'class' => Connection::class, @@ -91,6 +93,7 @@ 'db' => 'pgsql', ], 'mutexTimeout' => 0, + 'deleteReleased' => false, ], 'redis' => [ 'class' => RedisConnection::class, diff --git a/tests/cli/InfoActionTest.php b/tests/cli/InfoActionTest.php new file mode 100644 index 000000000..de8a3b876 --- /dev/null +++ b/tests/cli/InfoActionTest.php @@ -0,0 +1,186 @@ + + */ +class InfoActionTest extends TestCase +{ + public function testWaitingCount(): void + { + $controller = $this->getMockBuilder(Controller::class) + ->setConstructorArgs(['testController', new Module('testModule')]) + ->getMock(); + + $controller->expects(self::exactly(3)) + ->method('stdout') + ->willReturnOnConsecutiveCalls( + [ + 'Jobs' . PHP_EOL, + Console::FG_GREEN, + ], + [ + '- waiting: ', + Console::FG_YELLOW, + ], + [ + 10 . PHP_EOL + ] + ); + + $queue = $this->getMockBuilder(Queue::class)->getMock(); + + $provider = $this->getMockBuilder(WaitingCountProvider::class) + ->setConstructorArgs([$queue]) + ->getMock(); + $provider->expects(self::once()) + ->method('getWaitingCount') + ->willReturn(10); + + $queue->method('getStatisticsProvider')->willReturn($provider); + + $action = (new InfoAction('infoAction', $controller, [ + 'queue' => $queue, + ])); + $action->run(); + } + + public function testDelayedCount(): void + { + $controller = $this->getMockBuilder(Controller::class) + ->setConstructorArgs(['testController', new Module('testModule')]) + ->getMock(); + + $controller->expects(self::exactly(3)) + ->method('stdout') + ->willReturnOnConsecutiveCalls( + [ + 'Jobs' . PHP_EOL, + Console::FG_GREEN, + ], + [ + '- delayed: ', + Console::FG_YELLOW, + ], + [ + 10 . PHP_EOL + ] + ); + + $queue = $this->getMockBuilder(Queue::class)->getMock(); + + $provider = $this->getMockBuilder(DelayedCountProvider::class) + ->setConstructorArgs([$queue]) + ->getMock(); + $provider->expects(self::once()) + ->method('getDelayedCount') + ->willReturn(10); + + $queue->method('getStatisticsProvider')->willReturn($provider); + + $action = (new InfoAction('infoAction', $controller, [ + 'queue' => $queue, + ])); + $action->run(); + } + + public function testReservedCount(): void + { + $controller = $this->getMockBuilder(Controller::class) + ->setConstructorArgs(['testController', new Module('testModule')]) + ->getMock(); + + $controller->expects(self::exactly(3)) + ->method('stdout') + ->willReturnOnConsecutiveCalls( + [ + 'Jobs' . PHP_EOL, + Console::FG_GREEN, + ], + [ + '- reserved: ', + Console::FG_YELLOW, + ], + [ + 10 . PHP_EOL + ] + ); + + $queue = $this->getMockBuilder(Queue::class)->getMock(); + + $provider = $this->getMockBuilder(ReservedCountProvider::class) + ->setConstructorArgs([$queue]) + ->getMock() + ; + $provider->expects(self::once()) + ->method('getReservedCount') + ->willReturn(10); + + $queue->method('getStatisticsProvider')->willReturn($provider); + + $action = (new InfoAction('infoAction', $controller, [ + 'queue' => $queue, + ])); + $action->run(); + } + + public function testDoneCount(): void + { + $controller = $this->getMockBuilder(Controller::class) + ->setConstructorArgs(['testController', new Module('testModule')]) + ->getMock(); + + $controller->expects(self::exactly(3)) + ->method('stdout') + ->willReturnOnConsecutiveCalls( + [ + 'Jobs' . PHP_EOL, + Console::FG_GREEN, + ], + [ + '- done: ', + Console::FG_YELLOW, + ], + [ + 10 . PHP_EOL + ] + ); + + $queue = $this->getMockBuilder(Queue::class)->getMock(); + + $provider = $this->getMockBuilder(DoneCountProvider::class) + ->setConstructorArgs([$queue]) + ->getMock(); + $provider->expects(self::once()) + ->method('getDoneCount') + ->willReturn(10); + + $queue->method('getStatisticsProvider')->willReturn($provider); + + $action = (new InfoAction('infoAction', $controller, [ + 'queue' => $queue, + ])); + $action->run(); + } +} diff --git a/tests/cli/Queue.php b/tests/cli/Queue.php new file mode 100644 index 000000000..e7b5b981c --- /dev/null +++ b/tests/cli/Queue.php @@ -0,0 +1,53 @@ + + */ +class Queue extends CliQueue implements StatisticsProviderInterface +{ + /** + * @inheritdoc + */ + public function status($id): int + { + throw new NotSupportedException('"status" method is not supported.'); + } + /** + * @inheritdoc + */ + protected function pushMessage(string $payload, $ttr, $delay, $priority): int|string|null + { + throw new NotSupportedException('"pushMessage" method is not supported.'); + } + + private StatisticsInterface $_statisticsProvider; + + /** + * @return StatisticsInterface + */ + public function getStatisticsProvider(): StatisticsInterface + { + if (!isset($this->_statisticsProvider)) { + $this->_statisticsProvider = new BaseStatisticsProvider($this); + } + return $this->_statisticsProvider; + } +} diff --git a/tests/cli/providers/BaseStatisticsProvider.php b/tests/cli/providers/BaseStatisticsProvider.php new file mode 100644 index 000000000..3e7acce1a --- /dev/null +++ b/tests/cli/providers/BaseStatisticsProvider.php @@ -0,0 +1,34 @@ + + */ +class BaseStatisticsProvider extends BaseObject implements StatisticsInterface +{ + /** + * @var Queue + */ + protected Queue $queue; + + public function __construct(Queue $queue, array $config = []) + { + $this->queue = $queue; + parent::__construct($config); + } +} diff --git a/tests/cli/providers/DelayedCountProvider.php b/tests/cli/providers/DelayedCountProvider.php new file mode 100644 index 000000000..17e7d6dcd --- /dev/null +++ b/tests/cli/providers/DelayedCountProvider.php @@ -0,0 +1,29 @@ + + */ +class DelayedCountProvider extends BaseStatisticsProvider implements DelayedCountInterface +{ + /** + * @inheritdoc + */ + public function getDelayedCount(): int + { + return 10; + } +} diff --git a/tests/cli/providers/DoneCountProvider.php b/tests/cli/providers/DoneCountProvider.php new file mode 100644 index 000000000..1fa1b651d --- /dev/null +++ b/tests/cli/providers/DoneCountProvider.php @@ -0,0 +1,29 @@ + + */ +class DoneCountProvider extends BaseStatisticsProvider implements DoneCountInterface +{ + /** + * @inheritdoc + */ + public function getDoneCount(): int + { + return 10; + } +} diff --git a/tests/cli/providers/ReservedCountProvider.php b/tests/cli/providers/ReservedCountProvider.php new file mode 100644 index 000000000..2b1cd8879 --- /dev/null +++ b/tests/cli/providers/ReservedCountProvider.php @@ -0,0 +1,29 @@ + + */ +class ReservedCountProvider extends BaseStatisticsProvider implements ReservedCountInterface +{ + /** + * @inheritdoc + */ + public function getReservedCount(): int + { + return 10; + } +} diff --git a/tests/cli/providers/WaitingCountProvider.php b/tests/cli/providers/WaitingCountProvider.php new file mode 100644 index 000000000..2ae306797 --- /dev/null +++ b/tests/cli/providers/WaitingCountProvider.php @@ -0,0 +1,29 @@ + + */ +class WaitingCountProvider extends BaseStatisticsProvider implements WaitingCountInterface +{ + /** + * @inheritdoc + */ + public function getWaitingCount(): int + { + return 10; + } +} diff --git a/tests/docker-compose.yml b/tests/docker-compose.yml index 3c641de34..0efa42219 100644 --- a/tests/docker-compose.yml +++ b/tests/docker-compose.yml @@ -1,6 +1,5 @@ -version: "3.5" +--- services: - # https://hub.docker.com/_/php/ yii2-queue-php: container_name: yii2-queue-php @@ -8,7 +7,7 @@ services: context: .. dockerfile: tests/docker/php/Dockerfile args: - PHP_VERSION: ${PHP_VERSION:-8.1} + PHP_VERSION: ${PHP_VERSION:-8.2} volumes: - ./runtime/.composer:/root/.composer - ..:/code @@ -72,6 +71,10 @@ services: MYSQL_DATABASE: yii2_queue_test networks: net: {} + ulimits: + nofile: + soft: 262144 + hard: 262144 # https://hub.docker.com/_/postgres/ postgres: diff --git a/tests/docker/php/Dockerfile b/tests/docker/php/Dockerfile index da2ee33bb..c76d6726d 100644 --- a/tests/docker/php/Dockerfile +++ b/tests/docker/php/Dockerfile @@ -1,7 +1,7 @@ # Important! Do not use this image in production! ARG PHP_VERSION -FROM --platform=linux/amd64 php:${PHP_VERSION}-cli-alpine +FROM php:${PHP_VERSION}-cli-alpine RUN echo https://dl-cdn.alpinelinux.org/alpine/edge/main >> /etc/apk/repositories \ && echo https://dl-cdn.alpinelinux.org/alpine/edge/community >> /etc/apk/repositories \ diff --git a/tests/drivers/db/TestCase.php b/tests/drivers/db/TestCase.php index 61b36544a..57648e9a7 100644 --- a/tests/drivers/db/TestCase.php +++ b/tests/drivers/db/TestCase.php @@ -13,6 +13,7 @@ use tests\app\PriorityJob; use tests\app\RetryJob; use tests\drivers\CliTestCase; +use Yii; use yii\db\Query; /** @@ -108,8 +109,48 @@ public function testRemove(): void $this->assertEquals(0, $actual); } + public function testWaitingCount(): void + { + $this->getQueue()->push($this->createSimpleJob()); + + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getWaitingCount()); + } + + public function testDelayedCount(): void + { + $this->getQueue()->delay(5)->push($this->createSimpleJob()); + + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getDelayedCount()); + } + + public function testReservedCount(): void + { + $this->getQueue()->messageHandler = function () { + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getReservedCount()); + return true; + }; + + $job = $this->createSimpleJob(); + $this->getQueue()->push($job); + $this->getQueue()->run(false); + } + + public function testDoneCount(): void + { + $this->getQueue()->messageHandler = static function () { + return true; + }; + + $job = $this->createSimpleJob(); + $this->getQueue()->push($job); + $this->getQueue()->run(false); + + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getDoneCount()); + } + protected function tearDown(): void { + $this->getQueue()->messageHandler = null; $this->getQueue()->db->createCommand() ->delete($this->getQueue()->tableName) ->execute(); diff --git a/tests/drivers/file/QueueTest.php b/tests/drivers/file/QueueTest.php index 5fda9c247..cab469415 100644 --- a/tests/drivers/file/QueueTest.php +++ b/tests/drivers/file/QueueTest.php @@ -90,6 +90,42 @@ public function testRemove(): void $this->assertFileDoesNotExist($this->getQueue()->path . "/job$id.data"); } + public function testWaitingCount(): void + { + $this->getQueue()->push($this->createSimpleJob()); + + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getWaitingCount()); + } + + public function testDelayedCount(): void + { + $this->getQueue()->delay(5)->push($this->createSimpleJob()); + + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getDelayedCount()); + } + + public function testReservedCount(): void + { + $this->getQueue()->messageHandler = function () { + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getReservedCount()); + return true; + }; + + $this->getQueue()->push($this->createSimpleJob()); + $this->getQueue()->run(false); + } + + public function testDoneCount(): void + { + $this->startProcess(['php', 'yii', 'queue/listen', '1']); + $job = $this->createSimpleJob(); + $this->getQueue()->push($job); + + $this->assertSimpleJobDone($job); + + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getDoneCount()); + } + /** * @return Queue */ @@ -100,6 +136,7 @@ protected function getQueue(): Queue protected function tearDown(): void { + $this->getQueue()->messageHandler = null; foreach (glob(Yii::getAlias("@runtime/queue/*")) as $fileName) { unlink($fileName); } diff --git a/tests/drivers/redis/QueueTest.php b/tests/drivers/redis/QueueTest.php index 48ac8ff47..6777c0a45 100644 --- a/tests/drivers/redis/QueueTest.php +++ b/tests/drivers/redis/QueueTest.php @@ -13,7 +13,9 @@ use tests\app\RetryJob; use tests\drivers\CliTestCase; use Yii; +use yii\di\Instance; use yii\queue\redis\Queue; +use yii\redis\Connection; /** * Redis Queue Test. @@ -90,6 +92,42 @@ public function testRemove(): void $this->assertFalse((bool) $this->getQueue()->redis->hexists($this->getQueue()->channel . '.messages', $id)); } + public function testWaitingCount(): void + { + $this->getQueue()->push($this->createSimpleJob()); + + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getWaitingCount()); + } + + public function testDelayedCount(): void + { + $this->getQueue()->delay(5)->push($this->createSimpleJob()); + + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getDelayedCount()); + } + + public function testReservedCount(): void + { + $this->getQueue()->messageHandler = function () { + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getReservedCount()); + return true; + }; + + $this->getQueue()->push($this->createSimpleJob()); + $this->getQueue()->run(false); + } + + public function testDoneCount(): void + { + $this->startProcess(['php', 'yii', 'queue/listen', '1']); + $job = $this->createSimpleJob(); + $this->getQueue()->push($job); + + $this->assertSimpleJobDone($job); + + $this->assertEquals(1, $this->getQueue()->getStatisticsProvider()->getDoneCount()); + } + /** * @return Queue */ @@ -100,7 +138,58 @@ protected function getQueue(): Queue protected function tearDown(): void { + $this->getQueue()->messageHandler = null; $this->getQueue()->redis->flushdb(); parent::tearDown(); } + + /** + * Verify that Redis data persists when process crashes during moveExpired. + * + * Steps: + * 1. Push a delayed job into queue + * 2. Wait for the job to expire + * 3. Mock Redis to simulate crash during moveExpired + * 4. Successfully process job after recovery + */ + public function testConsumeDelayedMessageAtLeastOnce(): void + { + $job = $this->createSimpleJob(); + $this->getQueue()->delay(1)->push($job); + // Expect a single message to be received. + $messageCount = 0; + $this->getQueue()->messageHandler = static function () use(&$messageCount) { + $messageCount++; + return true; + }; + + // Ensure the delayed message can be consumed when more time passed than the delay is. + sleep(2); + + // Based on the implemention, emulate a crash when redis "rpush" + // command should be executed. + $mockRedis = Instance::ensure([ + 'class' => RedisCrashMock::class, + 'hostname' => getenv('REDIS_HOST') ?: 'localhost', + 'database' => getenv('REDIS_DB') ?: 1, + 'crashOnCommand' => 'rpush' // Crash when trying to move job to waiting queue. + ], Connection::class); + + $queue = $this->getQueue(); + $old = $queue->redis; + $queue->redis = $mockRedis; + + try { + $queue->run(false); + } catch (\Exception $e) { + // Ignore exceptions. + } finally { + $queue->redis = $old; + } + + // Ensure the red lock is invalid. The red lock is valid for 1s. + sleep(2); + $this->getQueue()->run(false); + $this->assertEquals(1, $messageCount); + } } diff --git a/tests/drivers/redis/RedisCrashMock.php b/tests/drivers/redis/RedisCrashMock.php new file mode 100644 index 000000000..b332d6abe --- /dev/null +++ b/tests/drivers/redis/RedisCrashMock.php @@ -0,0 +1,20 @@ +crashOnCommand) { + throw new \RuntimeException('Simulated Redis crash'); + } + return parent::executeCommand($name, $params); + } +}