diff --git a/.github/workflows/docs.yml b/.github/workflows/docs.yml
index 0937a92..4cb0907 100644
--- a/.github/workflows/docs.yml
+++ b/.github/workflows/docs.yml
@@ -4,21 +4,99 @@ on:
push:
branches:
- develop
+
+ tags:
+ - 'v*'
+
paths:
- - 'docs/**/*'
+ - 'docs/**'
- 'mkdocs.yml'
- '.github/workflows/docs.yml'
permissions:
contents: write
+concurrency:
+ group: docs-deploy
+ cancel-in-progress: false
+
jobs:
deploy:
runs-on: ubuntu-latest
+
steps:
- - uses: actions/checkout@v4
- - uses: actions/setup-python@v5
+ - name: Checkout repository
+ uses: actions/checkout@v7
+
+ - name: Setup Python
+ uses: actions/setup-python@v7
with:
- python-version: 3.x
- - run: pip install mkdocs-material
- - run: mkdocs gh-deploy --force
+ python-version: '3.x'
+
+ - name: Install dependencies
+ run: pip install mkdocs-material mike
+
+ - name: Fetch gh-pages
+ run: git fetch origin gh-pages --depth=1 || true
+
+ - name: Configure git
+ run: |
+ git config user.name github-actions[bot]
+ git config user.email 41898282+github-actions[bot]@users.noreply.github.com
+
+ - name: Deploy docs with mike
+ shell: bash
+ run: |
+ set -e
+
+ deploy_version () {
+ local VERSION="$1"
+ local UPDATE_LATEST="${2:-false}"
+
+ VERSION="${VERSION#v}"
+
+ MINOR=$(echo "$VERSION" | sed -E 's/^([0-9]+\.[0-9]+).*/\1/')
+
+ echo "Deploying docs version: $MINOR"
+
+ if [ "$UPDATE_LATEST" = "true" ]; then
+ mike deploy --push --update-aliases "$MINOR" latest
+ return
+ fi
+
+ mike deploy --push "$MINOR"
+ }
+
+ #
+ # Develop branch
+ #
+ if [ "${{ github.ref_type }}" = "branch" ] && [ "${{ github.ref_name }}" = "develop" ]; then
+ echo "Deploying dev docs"
+
+ mike deploy --push dev
+
+ exit 0
+ fi
+
+ #
+ # Release tags
+ #
+ if [ "${{ github.ref_type }}" = "tag" ]; then
+ VERSION="${{ github.ref_name }}"
+
+ if [[ ! "$VERSION" =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
+ echo "Skipping non-stable release tag: $VERSION"
+ exit 0
+ fi
+
+ deploy_version "$VERSION" true
+
+ echo "Setting default docs version: latest"
+
+ mike set-default --push latest
+
+ exit 0
+ fi
+
+ echo "Unsupported ref"
+ exit 1
diff --git a/.github/workflows/phpcpd.yml b/.github/workflows/phpcpd.yml
index c755c44..abc6ca1 100644
--- a/.github/workflows/phpcpd.yml
+++ b/.github/workflows/phpcpd.yml
@@ -22,7 +22,7 @@ jobs:
steps:
- name: Checkout
- uses: actions/checkout@v4
+ uses: actions/checkout@v7
- name: Setup PHP
uses: shivammathur/setup-php@v2
diff --git a/.github/workflows/phpcsfixer.yml b/.github/workflows/phpcsfixer.yml
index b1169be..f141827 100644
--- a/.github/workflows/phpcsfixer.yml
+++ b/.github/workflows/phpcsfixer.yml
@@ -22,7 +22,7 @@ jobs:
steps:
- name: Checkout
- uses: actions/checkout@v4
+ uses: actions/checkout@v7
- name: Set up PHP
uses: shivammathur/setup-php@v2
@@ -37,7 +37,7 @@ jobs:
run: echo "COMPOSER_CACHE_FILES_DIR=$(composer config cache-files-dir)" >> $GITHUB_ENV
- name: Cache composer dependencies
- uses: actions/cache@v4
+ uses: actions/cache@v6
with:
path: ${{ env.COMPOSER_CACHE_FILES_DIR }}
key: ${{ runner.os }}-composer-${{ hashFiles('**/composer.json') }}-${{ hashFiles('**/composer.lock') }}
diff --git a/.github/workflows/phpstan.yml b/.github/workflows/phpstan.yml
index 8e6ec0f..7d6d225 100644
--- a/.github/workflows/phpstan.yml
+++ b/.github/workflows/phpstan.yml
@@ -30,7 +30,7 @@ jobs:
steps:
- name: Checkout
- uses: actions/checkout@v4
+ uses: actions/checkout@v7
- name: Setup PHP
uses: shivammathur/setup-php@v2
@@ -46,7 +46,7 @@ jobs:
run: echo "COMPOSER_CACHE_FILES_DIR=$(composer config cache-files-dir)" >> $GITHUB_ENV
- name: Cache composer dependencies
- uses: actions/cache@v4
+ uses: actions/cache@v6
with:
path: ${{ env.COMPOSER_CACHE_FILES_DIR }}
key: ${{ runner.os }}-composer-${{ hashFiles('**/composer.json') }}-${{ hashFiles('**/composer.lock') }}
@@ -56,7 +56,7 @@ jobs:
run: mkdir -p build/phpstan
- name: Cache PHPStan results
- uses: actions/cache@v4
+ uses: actions/cache@v6
with:
path: build/phpstan
key: ${{ runner.os }}-phpstan-${{ github.sha }}
diff --git a/.github/workflows/phpunit.yml b/.github/workflows/phpunit.yml
index bb545b6..5763623 100644
--- a/.github/workflows/phpunit.yml
+++ b/.github/workflows/phpunit.yml
@@ -152,7 +152,7 @@ jobs:
run: sqlcmd -S 127.0.0.1 -U sa -P 1Secure*Password1 -Q "CREATE DATABASE test"
- name: Checkout
- uses: actions/checkout@v4
+ uses: actions/checkout@v7
- name: Set up PHP
uses: shivammathur/setup-php@v2
@@ -168,7 +168,7 @@ jobs:
run: echo "COMPOSER_CACHE_FILES_DIR=$(composer config cache-files-dir)" >> $GITHUB_ENV
- name: Cache composer dependencies
- uses: actions/cache@v4
+ uses: actions/cache@v6
with:
path: ${{ env.COMPOSER_CACHE_FILES_DIR }}
key: ${{ runner.os }}-composer-${{ hashFiles('**/composer.json') }}-${{ hashFiles('**/composer.lock') }}
diff --git a/.github/workflows/psalm.yml b/.github/workflows/psalm.yml
index a5cb6ad..62519bc 100644
--- a/.github/workflows/psalm.yml
+++ b/.github/workflows/psalm.yml
@@ -26,7 +26,7 @@ jobs:
steps:
- name: Checkout
- uses: actions/checkout@v4
+ uses: actions/checkout@v7
- name: Setup PHP
uses: shivammathur/setup-php@v2
@@ -42,7 +42,7 @@ jobs:
run: echo "COMPOSER_CACHE_FILES_DIR=$(composer config cache-files-dir)" >> $GITHUB_ENV
- name: Cache composer dependencies
- uses: actions/cache@v4
+ uses: actions/cache@v6
with:
path: ${{ env.COMPOSER_CACHE_FILES_DIR }}
key: ${{ runner.os }}-composer-${{ hashFiles('**/composer.json') }}-${{ hashFiles('**/composer.lock') }}
@@ -52,7 +52,7 @@ jobs:
run: mkdir -p build/psalm
- name: Cache Psalm results
- uses: actions/cache@v4
+ uses: actions/cache@v6
with:
path: build/psalm
key: ${{ runner.os }}-psalm-${{ github.sha }}
diff --git a/.github/workflows/rector.yml b/.github/workflows/rector.yml
index 194bf9e..610b551 100644
--- a/.github/workflows/rector.yml
+++ b/.github/workflows/rector.yml
@@ -37,7 +37,7 @@ jobs:
steps:
- name: Checkout
- uses: actions/checkout@v4
+ uses: actions/checkout@v7
- name: Set up PHP
uses: shivammathur/setup-php@v2
@@ -53,7 +53,7 @@ jobs:
run: echo "COMPOSER_CACHE_FILES_DIR=$(composer config cache-files-dir)" >> $GITHUB_ENV
- name: Cache composer dependencies
- uses: actions/cache@v4
+ uses: actions/cache@v6
with:
path: ${{ env.COMPOSER_CACHE_FILES_DIR }}
key: ${{ runner.os }}-composer-${{ hashFiles('**/composer.json') }}-${{ hashFiles('**/composer.lock') }}
diff --git a/composer.json b/composer.json
index 0320b25..d314be5 100644
--- a/composer.json
+++ b/composer.json
@@ -18,7 +18,7 @@
"require-dev": {
"codeigniter4/devkit": "^1.3",
"codeigniter4/framework": "^4.3",
- "predis/predis": "^2.0",
+ "predis/predis": "^2.0 || ^3.0",
"phpstan/phpstan-strict-rules": "^2.0",
"php-amqplib/php-amqplib": "^3.7"
},
diff --git a/docs/CNAME b/docs/CNAME
deleted file mode 100644
index ec5fa62..0000000
--- a/docs/CNAME
+++ /dev/null
@@ -1 +0,0 @@
-queue.codeigniter.com
diff --git a/docs/configuration.md b/docs/configuration.md
index 03d0a3d..3c3ede6 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -43,6 +43,8 @@ The configuration settings for `database` handler.
The [Strict Mode](https://codeigniter.com/user_guide/database/transactions.html#strict-mode) for the given `dbGroup` is automatically disabled - due to the nature of the queue worker.
+ When using **SQLite3** with multiple worker processes, configure `busyTimeout` in the selected connection group. This lets a worker wait briefly when another process holds SQLite's write lock instead of failing immediately with `SQLITE_BUSY`.
+
### $redis
The configuration settings for `redis` handler. You need to have a [ext-redis](https://github.com/phpredis/phpredis) installed to use it.
diff --git a/docs/overrides/main.html b/docs/overrides/main.html
new file mode 100644
index 0000000..bd5293a
--- /dev/null
+++ b/docs/overrides/main.html
@@ -0,0 +1,16 @@
+{% extends "base.html" %}
+
+{% block outdated %}
+ {% if config.site_url.rstrip("/").endswith("/dev") %}
+ You're viewing the development version.
+ Some features may not be available in a stable release yet.
+
+ Go to the latest stable version.
+
+ {% else %}
+ You're not viewing the latest version.
+
+ Click here to go to latest.
+
+ {% endif %}
+{% endblock %}
diff --git a/docs/running-queues.md b/docs/running-queues.md
index 5bae1c0..3ab20bd 100644
--- a/docs/running-queues.md
+++ b/docs/running-queues.md
@@ -117,6 +117,8 @@ When using the `chain()` method, there are a few important differences compared
As mentioned above, sometimes we may want to have multiple instances of the same command running at the same time. The queue is safe to use in that scenario with all databases as long as you keep the `skipLocked` to `true` in the config file. Only for SQLite3 driver, this setting is not relevant as it provides atomicity without the need for explicit concurrency control.
+When using **SQLite3** with multiple worker processes, configure `busyTimeout` in the selected connection group. This lets a worker wait briefly when another process holds SQLite's write lock instead of failing immediately with `SQLITE_BUSY`.
+
The PHPRedis and Predis drivers are also safe to use with multiple instances of the same command.
### Handling long-running process
diff --git a/mkdocs.yml b/mkdocs.yml
index 620eda4..b220898 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -1,8 +1,12 @@
site_name: CodeIgniter Queue
site_description: Queues documentation for CodeIgniter 4 framework
+exclude_docs: |
+ overrides/
+
theme:
name: material
+ custom_dir: docs/overrides
logo: assets/flame.svg
favicon: assets/favicon.ico
icon:
@@ -35,6 +39,8 @@ theme:
extra:
homepage: https://codeigniter.com
generator: false
+ version:
+ provider: mike
social:
- icon: material/github
diff --git a/rector.php b/rector.php
index 43fdac7..8d58a6f 100644
--- a/rector.php
+++ b/rector.php
@@ -29,7 +29,6 @@
use Rector\EarlyReturn\Rector\Return_\PreparedValueToEarlyReturnRector;
use Rector\Php55\Rector\String_\StringClassNameToClassConstantRector;
use Rector\Php73\Rector\FuncCall\StringifyStrNeedlesRector;
-use Rector\Php81\Rector\ClassMethod\NewInInitializerRector;
use Rector\PHPUnit\AnnotationsToAttributes\Rector\Class_\AnnotationWithValueToAttributeRector;
use Rector\PHPUnit\AnnotationsToAttributes\Rector\ClassMethod\DataProviderAnnotationToAttributeRector;
use Rector\PHPUnit\CodeQuality\Rector\Class_\YieldDataProviderRector;
@@ -103,11 +102,6 @@
RemoveUnusedPrivateMethodRector::class => [
__DIR__ . '/src/Commands/QueueWork.php',
],
-
- // Skip for PayloadMetadata in the constructor
- NewInInitializerRector::class => [
- __DIR__ . '/src/Payloads/Payload.php',
- ],
]);
// auto import fully qualified class names
diff --git a/src/Commands/QueueStop.php b/src/Commands/QueueStop.php
index 219db56..97cc315 100644
--- a/src/Commands/QueueStop.php
+++ b/src/Commands/QueueStop.php
@@ -60,8 +60,7 @@ class QueueStop extends BaseCommand
*
* @var array
*/
- protected $options = [
- ];
+ protected $options = [];
/**
* Actually execute a command.
diff --git a/src/Handlers/RabbitMQHandler.php b/src/Handlers/RabbitMQHandler.php
index d2f7692..0afe85c 100644
--- a/src/Handlers/RabbitMQHandler.php
+++ b/src/Handlers/RabbitMQHandler.php
@@ -27,6 +27,7 @@
use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Connection\AMQPConnectionConfig;
use PhpAmqpLib\Connection\AMQPConnectionFactory;
+use PhpAmqpLib\Exception\AMQPProtocolChannelException;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use Throwable;
@@ -35,8 +36,9 @@ class RabbitMQHandler extends BaseHandler
{
private readonly AbstractConnection $connection;
private readonly AMQPChannel $channel;
- private array $declaredQueues = [];
- private array $declaredExchanges = [];
+ private array $declaredLogicalQueues = [];
+ private array $declaredQueues = [];
+ private array $declaredExchanges = [];
public function __construct(protected QueueConfig $config)
{
@@ -273,8 +275,12 @@ public function clear(?string $queue = null): bool
{
try {
if ($queue === null) {
- // Clear all configured queues
- foreach (array_keys($this->config->queuePriorities) as $queueName) {
+ $queueNames = array_unique([
+ ...array_keys($this->config->queuePriorities),
+ ...array_keys($this->declaredLogicalQueues),
+ ]);
+
+ foreach ($queueNames as $queueName) {
$this->clearQueue($queueName);
}
} else {
@@ -300,7 +306,8 @@ public function clear(?string $queue = null): bool
*/
private function declareQueue(string $queue): void
{
- $priorities = $this->config->queuePriorities[$queue] ?? ['default'];
+ $this->declaredLogicalQueues[$queue] = true;
+ $priorities = $this->config->queuePriorities[$queue] ?? ['default'];
foreach ($priorities as $priority) {
$queueName = $this->getQueueName($queue, $priority);
@@ -459,15 +466,38 @@ private function publishWithOptionalConfirm(AMQPMessage $message, string $exchan
*/
private function clearQueue(string $queue): void
{
+ // Purge delayed jobs first so they cannot expire into a queue
+ // that has already been purged.
+ $this->purgeQueueIfExists($this->getDelayQueueName($queue));
+
$priorities = $this->config->queuePriorities[$queue] ?? ['default'];
foreach ($priorities as $priority) {
- $queueName = $this->getQueueName($queue, $priority);
+ $this->purgeQueueIfExists($this->getQueueName($queue, $priority));
+ }
+ }
+
+ /**
+ * Purge a queue without risking the handler's main channel.
+ */
+ private function purgeQueueIfExists(string $queue): void
+ {
+ $channel = $this->connection->channel();
+ try {
+ $channel->queue_purge($queue);
+ } catch (Throwable $e) {
+ // RabbitMQ closes the channel with 404 when the queue is missing.
+ if (! $e instanceof AMQPProtocolChannelException || $e->getCode() !== 404) {
+ throw $e;
+ }
+ } finally {
try {
- $this->channel->queue_purge($queueName);
+ if ($channel->is_open()) {
+ $channel->close();
+ }
} catch (Throwable) {
- // Queue might not exist, ignore
+ // The broker may already have closed this disposable channel.
}
}
}
diff --git a/src/Models/QueueJobFailedModel.php b/src/Models/QueueJobFailedModel.php
index e98b5f3..9946bdd 100644
--- a/src/Models/QueueJobFailedModel.php
+++ b/src/Models/QueueJobFailedModel.php
@@ -46,10 +46,8 @@ public function __construct(?ConnectionInterface $db = null, ?ValidationInterfac
{
$this->DBGroup = config('Queue')->database['dbGroup'];
- /**
- * @var BaseConnection|null $db
- */
$db ??= Database::connect($this->DBGroup);
+ assert($db instanceof BaseConnection);
// Turn off the Strict Mode
$db->transStrict(false);
diff --git a/src/Models/QueueJobModel.php b/src/Models/QueueJobModel.php
index 4df5bd1..2e13c8b 100644
--- a/src/Models/QueueJobModel.php
+++ b/src/Models/QueueJobModel.php
@@ -23,6 +23,7 @@
use CodeIgniter\Validation\ValidationInterface;
use Config\Database;
use ReflectionException;
+use Throwable;
class QueueJobModel extends Model
{
@@ -50,10 +51,8 @@ public function __construct(?ConnectionInterface $db = null, ?ValidationInterfac
{
$this->DBGroup = config('Queue')->database['dbGroup'];
- /**
- * @var BaseConnection|null $db
- */
$db ??= Database::connect($this->DBGroup);
+ assert($db instanceof BaseConnection);
// Turn off the Strict Mode
$db->transStrict(false);
@@ -74,6 +73,80 @@ public function getFromQueue(string $name, array $priority): ?QueueJob
// Make sure we still have the connection
$this->db->reconnect();
}
+
+ return match ($this->db->DBDriver) {
+ 'SQLite3' => $this->popAtomic($name, $priority),
+ default => $this->popWithLock($name, $priority),
+ };
+ }
+
+ /**
+ * Claim the oldest pending job while holding SQLite's write lock.
+ *
+ * SQLite has no FOR UPDATE SKIP LOCKED. BEGIN IMMEDIATE takes the write
+ * lock before the SELECT, so the candidate cannot be claimed by another
+ * worker between selecting it and marking it RESERVED.
+ *
+ * @throws ReflectionException
+ */
+ private function popAtomic(string $name, array $priority): ?QueueJob
+ {
+ if ($this->db->simpleQuery('BEGIN IMMEDIATE') === false) {
+ return null;
+ }
+
+ try {
+ $builder = $this->builder()
+ ->where('queue', $name)
+ ->where('status', Status::PENDING->value)
+ ->where('available_at <=', Time::now()->timestamp)
+ ->limit(1);
+
+ $builder = $this->setPriority($builder, $priority);
+ $sql = $builder->getCompiledSelect();
+
+ $query = $this->db->query($sql);
+
+ if ($query === false) {
+ $this->db->simpleQuery('ROLLBACK');
+
+ return null;
+ }
+
+ /** @var QueueJob|null $row */
+ $row = $query->getCustomRowObject(0, QueueJob::class);
+
+ if ($row === null) {
+ $this->db->simpleQuery('COMMIT');
+
+ return null;
+ }
+
+ $this->builder()
+ ->where('id', $row->id)
+ ->where('status', Status::PENDING->value)
+ ->update(['status' => Status::RESERVED->value]);
+
+ $claimed = $this->db->affectedRows() === 1;
+
+ $this->db->simpleQuery('COMMIT');
+
+ return $claimed ? $row : null;
+ } catch (Throwable $e) {
+ $this->db->simpleQuery('ROLLBACK');
+
+ throw $e;
+ }
+ }
+
+ /**
+ * Claim the oldest pending job inside a transaction, using
+ * FOR UPDATE SKIP LOCKED where supported.
+ *
+ * @throws ReflectionException
+ */
+ private function popWithLock(string $name, array $priority): ?QueueJob
+ {
// Start transaction
$this->db->transStart();
@@ -89,6 +162,8 @@ public function getFromQueue(string $name, array $priority): ?QueueJob
$query = $this->db->query($this->skipLocked($sql));
if ($query === false) {
+ $this->db->transComplete();
+
return null;
}
/** @var QueueJob|null $row */
@@ -133,15 +208,19 @@ private function skipLocked(string $sql): string
*/
private function setPriority(BaseBuilder $builder, array $priority): BaseBuilder
{
+ $priority = array_values($priority);
+
$builder->whereIn('priority', $priority);
if ($priority !== ['default']) {
+ $escapedPriority = array_map($this->db->escape(...), $priority);
+
if ($this->db->DBDriver !== 'MySQLi') {
$builder->orderBy(
sprintf('CASE %s ', $this->db->protectIdentifiers('priority'))
. implode(
' ',
- array_map(static fn ($value, $key) => "WHEN '{$value}' THEN {$key}", $priority, array_keys($priority)),
+ array_map(static fn ($value, $key) => "WHEN {$value} THEN {$key}", $escapedPriority, array_keys($escapedPriority)),
)
. ' END',
'',
@@ -150,10 +229,7 @@ private function setPriority(BaseBuilder $builder, array $priority): BaseBuilder
} else {
$builder->orderBy(
'FIELD(priority, '
- . implode(
- ',',
- array_map(static fn ($value) => "'{$value}'", $priority),
- )
+ . implode(',', $escapedPriority)
. ')',
'',
false,
diff --git a/tests/Models/QueueJobModelTest.php b/tests/Models/QueueJobModelTest.php
index 4d61f7e..00ef79f 100644
--- a/tests/Models/QueueJobModelTest.php
+++ b/tests/Models/QueueJobModelTest.php
@@ -13,8 +13,11 @@
namespace Tests\Models;
+use CodeIgniter\Queue\Entities\QueueJob;
+use CodeIgniter\Queue\Enums\Status;
use CodeIgniter\Queue\Models\QueueJobModel;
use CodeIgniter\Test\ReflectionHelper;
+use Tests\Support\Database\Seeds\TestDatabaseQueueSeeder;
use Tests\Support\TestCase;
/**
@@ -24,6 +27,8 @@ final class QueueJobModelTest extends TestCase
{
use ReflectionHelper;
+ protected $seed = TestDatabaseQueueSeeder::class;
+
public function testQueueJobModel(): void
{
$model = model(QueueJobModel::class);
@@ -59,4 +64,72 @@ public function testSkipLockedFalse(): void
$this->assertSame($sql, $result);
}
+
+ public function testGetFromQueueReservesPendingJob(): void
+ {
+ $model = model(QueueJobModel::class);
+
+ $job = $model->getFromQueue('queue1', ['default']);
+
+ $this->assertInstanceOf(QueueJob::class, $job);
+ $this->assertSame('queue1', $job->queue);
+ $this->seeInDatabase('queue_jobs', [
+ 'id' => $job->id,
+ 'status' => Status::RESERVED->value,
+ ]);
+ }
+
+ public function testGetFromQueueReturnsNullWhenNothingPending(): void
+ {
+ $model = model(QueueJobModel::class);
+
+ $this->assertNull($model->getFromQueue('queue123', ['default']));
+ }
+
+ public function testGetFromQueueOnSQLite3ClaimsEachPendingJobOnce(): void
+ {
+ $model = model(QueueJobModel::class);
+
+ if ($model->db->DBDriver !== 'SQLite3') {
+ $this->markTestSkipped('BEGIN IMMEDIATE claim is SQLite3-only.');
+ }
+
+ $model->insert(new QueueJob([
+ 'queue' => 'queue1',
+ 'payload' => ['job' => 'extra', 'data' => []],
+ 'priority' => 'default',
+ 'status' => Status::PENDING->value,
+ 'attempts' => 0,
+ 'available_at' => 1_697_269_865,
+ ]));
+
+ $job1 = $model->getFromQueue('queue1', ['default']);
+ $job2 = $model->getFromQueue('queue1', ['default']);
+
+ $this->assertInstanceOf(QueueJob::class, $job1);
+ $this->assertInstanceOf(QueueJob::class, $job2);
+ $this->assertSame(2, $job1->id);
+ $this->assertSame(3, $job2->id);
+ $this->assertNull($model->getFromQueue('queue1', ['default']));
+ $this->seeInDatabase('queue_jobs', ['id' => 2, 'status' => Status::RESERVED->value]);
+ $this->seeInDatabase('queue_jobs', ['id' => 3, 'status' => Status::RESERVED->value]);
+ }
+
+ public function testSetPriorityEscapesPriorityValues(): void
+ {
+ $model = model(QueueJobModel::class);
+ $method = $this->getPrivateMethodInvoker($model, 'setPriority');
+ $builder = $model->builder();
+
+ $priority = [
+ 'priority_key' => "default' THEN 0 ELSE 1 END --",
+ 'default',
+ ];
+
+ $result = $method($builder, $priority);
+ $sql = (string) $result->getCompiledSelect();
+
+ $this->assertStringContainsString($model->db->escape($priority['priority_key']), $sql);
+ $this->assertStringNotContainsString('priority_key', $sql);
+ }
}
diff --git a/tests/PushAndPopWithDelayTest.php b/tests/PushAndPopWithDelayTest.php
index f06b0f9..7ac34ae 100644
--- a/tests/PushAndPopWithDelayTest.php
+++ b/tests/PushAndPopWithDelayTest.php
@@ -44,43 +44,47 @@ public function testPushAndPopWithDelay(string $name, string $class): void
Time::setTestNow('2023-12-29 14:15:16');
$handler = new $class($this->config);
- $result = $handler->setDelay(MINUTE)->push('queue-delay', 'success', ['key1' => 'value1']);
-
- $this->assertNotNull($result);
-
- $result = $handler->push('queue-delay', 'success', ['key2' => 'value2']);
-
- $this->assertNotNull($result);
-
- if ($name === 'database') {
- $this->seeInDatabase('queue_jobs', [
- 'queue' => 'queue-delay',
- 'payload' => json_encode(['job' => 'success', 'data' => ['key1' => 'value1'], 'metadata' => []]),
- 'available_at' => 1703859376,
- ]);
-
- $this->seeInDatabase('queue_jobs', [
- 'queue' => 'queue-delay',
- 'payload' => json_encode(['job' => 'success', 'data' => ['key2' => 'value2'], 'metadata' => []]),
- 'available_at' => 1703859316,
- ]);
+ $handler->clear('queue-delay');
+
+ try {
+ $result = $handler->setDelay(MINUTE)->push('queue-delay', 'success', ['key1' => 'value1']);
+ $this->assertTrue($result->getStatus());
+
+ $result = $handler->push('queue-delay', 'success', ['key2' => 'value2']);
+ $this->assertTrue($result->getStatus());
+
+ if ($name === 'database') {
+ $this->seeInDatabase('queue_jobs', [
+ 'queue' => 'queue-delay',
+ 'payload' => json_encode(['job' => 'success', 'data' => ['key1' => 'value1'], 'metadata' => []]),
+ 'available_at' => 1703859376,
+ ]);
+
+ $this->seeInDatabase('queue_jobs', [
+ 'queue' => 'queue-delay',
+ 'payload' => json_encode(['job' => 'success', 'data' => ['key2' => 'value2'], 'metadata' => []]),
+ 'available_at' => 1703859316,
+ ]);
+ }
+
+ $result = $handler->pop('queue-delay', ['default']);
+ $this->assertInstanceOf(QueueJob::class, $result);
+ $payload = ['job' => 'success', 'data' => ['key2' => 'value2'], 'metadata' => []];
+ $this->assertSame($payload, $result->payload);
+
+ $result = $handler->pop('queue-delay', ['default']);
+ $this->assertNull($result);
+
+ // add 1 minute
+ Time::setTestNow('2023-12-29 14:16:16');
+
+ $result = $handler->pop('queue-delay', ['default']);
+ $this->assertInstanceOf(QueueJob::class, $result);
+ $payload = ['job' => 'success', 'data' => ['key1' => 'value1'], 'metadata' => []];
+ $this->assertSame($payload, $result->payload);
+ } finally {
+ $handler->clear('queue-delay');
}
-
- $result = $handler->pop('queue-delay', ['default']);
- $this->assertInstanceOf(QueueJob::class, $result);
- $payload = ['job' => 'success', 'data' => ['key2' => 'value2'], 'metadata' => []];
- $this->assertSame($payload, $result->payload);
-
- $result = $handler->pop('queue-delay', ['default']);
- $this->assertNull($result);
-
- // add 1 minute
- Time::setTestNow('2023-12-29 14:16:16');
-
- $result = $handler->pop('queue-delay', ['default']);
- $this->assertInstanceOf(QueueJob::class, $result);
- $payload = ['job' => 'success', 'data' => ['key1' => 'value1'], 'metadata' => []];
- $this->assertSame($payload, $result->payload);
}
public static function providePushAndPopWithDelay(): iterable
diff --git a/tests/RabbitMQDelayTest.php b/tests/RabbitMQDelayTest.php
index b4787f2..233db96 100644
--- a/tests/RabbitMQDelayTest.php
+++ b/tests/RabbitMQDelayTest.php
@@ -17,6 +17,7 @@
use CodeIgniter\Queue\Entities\QueueJob;
use CodeIgniter\Queue\Handlers\RabbitMQHandler;
use CodeIgniter\Queue\QueuePushResult;
+use CodeIgniter\Test\ReflectionHelper;
use PhpAmqpLib\Connection\AMQPConnectionFactory;
use Tests\Support\Config\Queue as QueueConfig;
use Tests\Support\TestCase;
@@ -29,13 +30,17 @@
*/
final class RabbitMQDelayTest extends TestCase
{
+ use ReflectionHelper;
+
private ?RabbitMQHandler $handler = null;
+ private string $queue;
protected function setUp(): void
{
parent::setUp();
- $config = config(QueueConfig::class);
+ $config = config(QueueConfig::class);
+ $this->queue = 'delay-test-queue-' . bin2hex(random_bytes(6));
// Skip tests if RabbitMQ is not available
if (! $this->isRabbitMQAvailable()) {
@@ -52,9 +57,8 @@ protected function setUp(): void
protected function tearDown(): void
{
if ($this->handler !== null) {
- // Clear test queues
try {
- $this->handler->clear('delay-test-queue');
+ $this->deleteDeclaredResources();
} catch (Throwable) {
// Ignore cleanup errors
}
@@ -70,23 +74,23 @@ public function testDelayedMessageWithRealTiming(): void
$startTime = time();
// Push a delayed job
- $result = $this->handler->setDelay($delaySeconds)->push('delay-test-queue', 'success', ['type' => 'delayed']);
+ $result = $this->handler->setDelay($delaySeconds)->push($this->queue, 'success', ['type' => 'delayed']);
$this->assertInstanceOf(QueuePushResult::class, $result);
$this->assertTrue($result->getStatus());
// Push an immediate job
- $result = $this->handler->push('delay-test-queue', 'success', ['type' => 'immediate']);
+ $result = $this->handler->push($this->queue, 'success', ['type' => 'immediate']);
$this->assertInstanceOf(QueuePushResult::class, $result);
$this->assertTrue($result->getStatus());
// Should get immediate job first
- $job = $this->handler->pop('delay-test-queue', ['default']);
+ $job = $this->handler->pop($this->queue, ['default']);
$this->assertInstanceOf(QueueJob::class, $job);
$this->assertSame('immediate', $job->payload['data']['type']);
$this->handler->done($job);
// Should not get delayed job yet (within first second)
- $job = $this->handler->pop('delay-test-queue', ['default']);
+ $job = $this->handler->pop($this->queue, ['default']);
$this->assertNull($job);
// Wait for delay to expire (with a small buffer)
@@ -94,7 +98,7 @@ public function testDelayedMessageWithRealTiming(): void
sleep($waitTime);
// Should now get the delayed job
- $job = $this->handler->pop('delay-test-queue', ['default']);
+ $job = $this->handler->pop($this->queue, ['default']);
$this->assertInstanceOf(QueueJob::class, $job);
$this->assertSame('delayed', $job->payload['data']['type']);
@@ -108,35 +112,32 @@ public function testDelayedMessageWithRealTiming(): void
public function testMultipleDelayedJobsWithDifferentDelays(): void
{
- // Push jobs with different delays
- $result1 = $this->handler->setDelay(1)->push('delay-test-queue', 'success', ['order' => 'first', 'delay' => 1]);
- $result2 = $this->handler->setDelay(3)->push('delay-test-queue', 'success', ['order' => 'second', 'delay' => 3]);
- $result3 = $this->handler->push('delay-test-queue', 'success', ['order' => 'immediate', 'delay' => 0]);
+ // Push the immediate job first so slow test setup cannot allow a
+ // delayed job to reach the main queue ahead of it.
+ $result1 = $this->handler->push($this->queue, 'success', ['order' => 'immediate', 'delay' => 0]);
+ $result2 = $this->handler->setDelay(1)->push($this->queue, 'success', ['order' => 'first', 'delay' => 1]);
+ $result3 = $this->handler->setDelay(3)->push($this->queue, 'success', ['order' => 'second', 'delay' => 3]);
$this->assertTrue($result1->getStatus());
$this->assertTrue($result2->getStatus());
$this->assertTrue($result3->getStatus());
// Should get immediate job first
- $job = $this->handler->pop('delay-test-queue', ['default']);
+ $job = $this->handler->pop($this->queue, ['default']);
$this->assertInstanceOf(QueueJob::class, $job);
$this->assertSame('immediate', $job->payload['data']['order']);
$this->handler->done($job);
// Wait 2 seconds - should get first delayed job
sleep(2);
- $job = $this->handler->pop('delay-test-queue', ['default']);
+ $job = $this->handler->pop($this->queue, ['default']);
$this->assertInstanceOf(QueueJob::class, $job);
$this->assertSame('first', $job->payload['data']['order']);
$this->handler->done($job);
- // Should not get second job yet
- $job = $this->handler->pop('delay-test-queue', ['default']);
- $this->assertNull($job);
-
// Wait another 2 seconds - should get second delayed job
sleep(2);
- $job = $this->handler->pop('delay-test-queue', ['default']);
+ $job = $this->handler->pop($this->queue, ['default']);
$this->assertInstanceOf(QueueJob::class, $job);
$this->assertSame('second', $job->payload['data']['order']);
$this->handler->done($job);
@@ -145,17 +146,33 @@ public function testMultipleDelayedJobsWithDifferentDelays(): void
public function testZeroDelayWorksImmediately(): void
{
// Jobs with 0 delay should work immediately
- $result = $this->handler->setDelay(0)->push('delay-test-queue', 'success', ['type' => 'zero-delay']);
+ $result = $this->handler->setDelay(0)->push($this->queue, 'success', ['type' => 'zero-delay']);
$this->assertTrue($result->getStatus());
// Should be able to pop immediately
- $job = $this->handler->pop('delay-test-queue', ['default']);
+ $job = $this->handler->pop($this->queue, ['default']);
$this->assertInstanceOf(QueueJob::class, $job);
$this->assertSame('zero-delay', $job->payload['data']['type']);
$this->handler->done($job);
}
+ public function testFreshHandlerClearsImmediateAndDelayedJobs(): void
+ {
+ $delayedResult = $this->handler->setDelay(1)->push($this->queue, 'success', ['type' => 'delayed']);
+ $this->assertTrue($delayedResult->getStatus());
+
+ $immediateResult = $this->handler->push($this->queue, 'success', ['type' => 'immediate']);
+ $this->assertTrue($immediateResult->getStatus());
+
+ $clearer = new RabbitMQHandler(config(QueueConfig::class));
+ $this->assertTrue($clearer->clear($this->queue));
+
+ sleep(2);
+
+ $this->assertNull($clearer->pop($this->queue, ['default']));
+ }
+
/**
* Check if RabbitMQ is available for testing.
*/
@@ -163,4 +180,17 @@ private function isRabbitMQAvailable(): bool
{
return class_exists(AMQPConnectionFactory::class);
}
+
+ private function deleteDeclaredResources(): void
+ {
+ $channel = self::getPrivateProperty($this->handler, 'channel');
+
+ foreach (array_keys(self::getPrivateProperty($this->handler, 'declaredQueues')) as $queue) {
+ $channel->queue_delete($queue);
+ }
+
+ foreach (array_keys(self::getPrivateProperty($this->handler, 'declaredExchanges')) as $exchange) {
+ $channel->exchange_delete($exchange);
+ }
+ }
}
diff --git a/tests/RabbitMQHandlerTest.php b/tests/RabbitMQHandlerTest.php
index d9b26c2..c2335a6 100644
--- a/tests/RabbitMQHandlerTest.php
+++ b/tests/RabbitMQHandlerTest.php
@@ -32,14 +32,27 @@ final class RabbitMQHandlerTest extends TestCase
{
use ReflectionHelper;
+ private string $customPriorityQueue;
+ private string $emptyQueue;
private QueueConfig $config;
private ?RabbitMQHandler $handler = null;
+ private string $priorityQueue;
+ private string $testQueue;
+ private string $testQueue1;
+ private string $testQueue2;
protected function setUp(): void
{
parent::setUp();
- $this->config = config(QueueConfig::class);
+ $suffix = bin2hex(random_bytes(6));
+ $this->testQueue = 'test-queue-' . $suffix;
+ $this->testQueue1 = 'test-queue-1-' . $suffix;
+ $this->testQueue2 = 'test-queue-2-' . $suffix;
+ $this->priorityQueue = 'priority-test-' . $suffix;
+ $this->customPriorityQueue = 'custom-priority-queue-' . $suffix;
+ $this->emptyQueue = 'empty-queue-' . $suffix;
+ $this->config = config(QueueConfig::class);
// Skip tests if RabbitMQ is not available
if (! $this->isRabbitMQAvailable()) {
@@ -56,13 +69,8 @@ protected function setUp(): void
protected function tearDown(): void
{
if ($this->handler !== null) {
- // Clear test queues
try {
- $this->handler->clear('test-queue');
- $this->handler->clear('test-queue-1');
- $this->handler->clear('test-queue-2');
- $this->handler->clear('priority-test');
- $this->handler->clear('custom-priority-queue');
+ $this->deleteDeclaredResources();
} catch (Throwable) {
// Ignore cleanup errors
}
@@ -91,7 +99,7 @@ public function testRabbitMQConnectionFailure(): void
public function testPushJob(): void
{
- $result = $this->handler->push('test-queue', 'success', ['message' => 'Hello World']);
+ $result = $this->handler->push($this->testQueue, 'success', ['message' => 'Hello World']);
$this->assertInstanceOf(QueuePushResult::class, $result);
$this->assertTrue($result->getStatus());
@@ -101,7 +109,7 @@ public function testPushJob(): void
public function testPushJobWithDelay(): void
{
- $result = $this->handler->setDelay(30)->push('test-queue', 'success', ['message' => 'Delayed']);
+ $result = $this->handler->setDelay(30)->push($this->testQueue, 'success', ['message' => 'Delayed']);
$this->assertInstanceOf(QueuePushResult::class, $result);
$this->assertTrue($result->getStatus());
@@ -109,103 +117,100 @@ public function testPushJobWithDelay(): void
public function testPushJobWithPriority(): void
{
- $this->config->queuePriorities['priority-test'] = ['high', 'default', 'low'];
+ $this->config->queuePriorities[$this->priorityQueue] = ['high', 'default', 'low'];
- $result = $this->handler->setPriority('high')->push('priority-test', 'success', ['priority' => 'high']);
+ $result = $this->handler->setPriority('high')->push($this->priorityQueue, 'success', ['priority' => 'high']);
$this->assertTrue($result->getStatus());
}
public function testPopJob(): void
{
- $this->handler->push('test-queue', 'success', ['message' => 'Test Pop']);
+ $this->handler->push($this->testQueue, 'success', ['message' => 'Test Pop']);
- // Give RabbitMQ a moment to process
- usleep(100_000);
+ $job = $this->popEventually($this->testQueue, ['default']);
- $job = $this->handler->pop('test-queue', ['default']);
+ $this->assertInstanceOf(QueueJob::class, $job);
+ $this->assertSame($this->testQueue, $job->queue);
+ $this->assertSame('success', $job->payload['job']);
+ $this->assertSame(['message' => 'Test Pop'], $job->payload['data']);
- if ($job !== null) {
- $this->assertInstanceOf(QueueJob::class, $job);
- $this->assertSame('test-queue', $job->queue);
- $this->assertSame('success', $job->payload['job']);
- $this->assertSame(['message' => 'Test Pop'], $job->payload['data']);
-
- // Clean up - mark as done
- $this->handler->done($job);
- }
+ // Clean up - mark as done
+ $this->handler->done($job);
}
public function testPopJobWithPriorities(): void
{
- $this->config->queuePriorities['priority-test'] = ['high', 'default', 'low'];
+ $this->config->queuePriorities[$this->priorityQueue] = ['high', 'default', 'low'];
// Push jobs with different priorities
- $this->handler->setPriority('low')->push('priority-test', 'success', ['priority' => 'low']);
- $this->handler->setPriority('high')->push('priority-test', 'success', ['priority' => 'high']);
- $this->handler->setPriority('default')->push('priority-test', 'success', ['priority' => 'default']);
-
- usleep(100_000);
+ $this->handler->setPriority('low')->push($this->priorityQueue, 'success', ['priority' => 'low']);
+ $this->handler->setPriority('high')->push($this->priorityQueue, 'success', ['priority' => 'high']);
+ $this->handler->setPriority('default')->push($this->priorityQueue, 'success', ['priority' => 'default']);
// Should get high priority job first
- $job = $this->handler->pop('priority-test', ['high', 'default', 'low']);
+ $job = $this->popEventually($this->priorityQueue, ['high', 'default', 'low']);
- if ($job !== null) {
- $this->assertSame('high', $job->priority);
- $this->handler->done($job);
- }
+ $this->assertInstanceOf(QueueJob::class, $job);
+ $this->assertSame('high', $job->priority);
+ $this->handler->done($job);
}
public function testJobFailure(): void
{
- $this->handler->push('test-queue', 'failure', ['message' => 'Will Fail']);
-
- usleep(100_000);
+ $this->handler->push($this->testQueue, 'failure', ['message' => 'Will Fail']);
- $job = $this->handler->pop('test-queue', ['default']);
+ $job = $this->popEventually($this->testQueue, ['default']);
- if ($job !== null) {
- $exception = new Exception('Test failure');
- $result = $this->handler->failed($job, $exception, false);
+ $this->assertInstanceOf(QueueJob::class, $job);
+ $exception = new Exception('Test failure');
+ $result = $this->handler->failed($job, $exception, false);
- $this->assertTrue($result);
- }
+ $this->assertTrue($result);
}
public function testJobLater(): void
{
- $this->handler->push('test-queue', 'success', ['message' => 'Reschedule']);
-
- usleep(100_000);
+ $this->handler->push($this->testQueue, 'success', ['message' => 'Reschedule']);
- $job = $this->handler->pop('test-queue', ['default']);
+ $job = $this->popEventually($this->testQueue, ['default']);
- if ($job !== null) {
- $result = $this->handler->later($job, 60);
- $this->assertTrue($result);
- }
+ $this->assertInstanceOf(QueueJob::class, $job);
+ $result = $this->handler->later($job, 60);
+ $this->assertTrue($result);
}
public function testClearQueue(): void
{
- $this->handler->push('test-queue', 'success', ['message' => 'Clear Test 1']);
- $this->handler->push('test-queue', 'success', ['message' => 'Clear Test 2']);
+ $this->handler->push($this->testQueue, 'success', ['message' => 'Clear Test 1']);
+ $this->handler->push($this->testQueue, 'success', ['message' => 'Clear Test 2']);
- usleep(100_000);
-
- $result = $this->handler->clear('test-queue');
+ $result = $this->handler->clear($this->testQueue);
$this->assertTrue($result);
// Verify queue is empty
- $job = $this->handler->pop('test-queue', ['default']);
+ $job = $this->handler->pop($this->testQueue, ['default']);
$this->assertNull($job);
}
+ public function testClearNonexistentQueueKeepsHandlerUsable(): void
+ {
+ $this->assertTrue($this->handler->clear($this->testQueue));
+
+ $result = $this->handler->push($this->testQueue, 'success', ['message' => 'Still usable']);
+ $this->assertTrue($result->getStatus());
+
+ $job = $this->handler->pop($this->testQueue, ['default']);
+ $this->assertInstanceOf(QueueJob::class, $job);
+ $this->assertSame('Still usable', $job->payload['data']['message']);
+ $this->handler->done($job);
+ }
+
public function testIncorrectJobHandler(): void
{
$this->expectException(QueueException::class);
- $this->handler->push('test-queue', 'nonexistent-job', []);
+ $this->handler->push($this->testQueue, 'nonexistent-job', []);
}
public function testIncorrectQueueFormat(): void
@@ -219,47 +224,42 @@ public function testIncorrectPriority(): void
{
$this->expectException(QueueException::class);
- $this->config->queuePriorities['test-queue'] = ['high', 'low'];
+ $this->config->queuePriorities[$this->testQueue] = ['high', 'low'];
- $this->handler->setPriority('medium')->push('test-queue', 'success', []);
+ $this->handler->setPriority('medium')->push($this->testQueue, 'success', []);
}
public function testCustomPriorityMapping(): void
{
// Define custom priorities for a queue
- $this->config->queuePriorities['custom-priority-queue'] = ['urgent', 'normal', 'low'];
+ $this->config->queuePriorities[$this->customPriorityQueue] = ['urgent', 'normal', 'low'];
// Test that we can push jobs with custom priorities
- $result1 = $this->handler->setPriority('urgent')->push('custom-priority-queue', 'success', ['priority' => 'urgent']);
- $result2 = $this->handler->setPriority('normal')->push('custom-priority-queue', 'success', ['priority' => 'normal']);
- $result3 = $this->handler->setPriority('low')->push('custom-priority-queue', 'success', ['priority' => 'low']);
+ $result1 = $this->handler->setPriority('urgent')->push($this->customPriorityQueue, 'success', ['priority' => 'urgent']);
+ $result2 = $this->handler->setPriority('normal')->push($this->customPriorityQueue, 'success', ['priority' => 'normal']);
+ $result3 = $this->handler->setPriority('low')->push($this->customPriorityQueue, 'success', ['priority' => 'low']);
$this->assertTrue($result1->getStatus());
$this->assertTrue($result2->getStatus());
$this->assertTrue($result3->getStatus());
- usleep(100_000);
-
// Should get urgent priority job first
- $job = $this->handler->pop('custom-priority-queue', ['urgent', 'normal', 'low']);
- if ($job !== null) {
- $this->assertSame('urgent', $job->payload['data']['priority']);
- $this->handler->done($job);
- }
+ $job = $this->popEventually($this->customPriorityQueue, ['urgent', 'normal', 'low']);
+ $this->assertInstanceOf(QueueJob::class, $job);
+ $this->assertSame('urgent', $job->payload['data']['priority']);
+ $this->handler->done($job);
// Then normal priority
- $job = $this->handler->pop('custom-priority-queue', ['urgent', 'normal', 'low']);
- if ($job !== null) {
- $this->assertSame('normal', $job->payload['data']['priority']);
- $this->handler->done($job);
- }
+ $job = $this->handler->pop($this->customPriorityQueue, ['urgent', 'normal', 'low']);
+ $this->assertInstanceOf(QueueJob::class, $job);
+ $this->assertSame('normal', $job->payload['data']['priority']);
+ $this->handler->done($job);
// Finally low priority
- $job = $this->handler->pop('custom-priority-queue', ['urgent', 'normal', 'low']);
- if ($job !== null) {
- $this->assertSame('low', $job->payload['data']['priority']);
- $this->handler->done($job);
- }
+ $job = $this->handler->pop($this->customPriorityQueue, ['urgent', 'normal', 'low']);
+ $this->assertInstanceOf(QueueJob::class, $job);
+ $this->assertSame('low', $job->payload['data']['priority']);
+ $this->handler->done($job);
}
public function testPriority(): void
@@ -279,15 +279,15 @@ public function testPriorityException(): void
public function testPopEmpty(): void
{
- $result = $this->handler->pop('empty-queue', ['default']);
+ $result = $this->handler->pop($this->emptyQueue, ['default']);
$this->assertNull($result);
}
public function testFailedAndKeepJob(): void
{
- $this->handler->push('test-queue', 'success', ['test' => 'data']);
- $queueJob = $this->handler->pop('test-queue', ['default']);
+ $this->handler->push($this->testQueue, 'success', ['test' => 'data']);
+ $queueJob = $this->handler->pop($this->testQueue, ['default']);
$this->assertInstanceOf(QueueJob::class, $queueJob);
@@ -297,15 +297,15 @@ public function testFailedAndKeepJob(): void
$this->assertTrue($result);
$this->seeInDatabase('queue_jobs_failed', [
- 'queue' => 'test-queue',
+ 'queue' => $this->testQueue,
'connection' => 'rabbitmq',
]);
}
public function testFailedAndDontKeepJob(): void
{
- $this->handler->push('test-queue', 'success', ['test' => 'data']);
- $queueJob = $this->handler->pop('test-queue', ['default']);
+ $this->handler->push($this->testQueue, 'success', ['test' => 'data']);
+ $queueJob = $this->handler->pop($this->testQueue, ['default']);
$this->assertInstanceOf(QueueJob::class, $queueJob);
@@ -315,15 +315,15 @@ public function testFailedAndDontKeepJob(): void
$this->assertTrue($result);
$this->dontSeeInDatabase('queue_jobs_failed', [
- 'queue' => 'test-queue',
+ 'queue' => $this->testQueue,
'connection' => 'rabbitmq',
]);
}
public function testDone(): void
{
- $this->handler->push('test-queue', 'success', ['test' => 'data']);
- $queueJob = $this->handler->pop('test-queue', ['default']);
+ $this->handler->push($this->testQueue, 'success', ['test' => 'data']);
+ $queueJob = $this->handler->pop($this->testQueue, ['default']);
$this->assertInstanceOf(QueueJob::class, $queueJob);
@@ -335,13 +335,11 @@ public function testDone(): void
public function testClearAll(): void
{
- $this->handler->push('test-queue-1', 'success', ['test' => 'data1']);
- $this->handler->push('test-queue-2', 'success', ['test' => 'data2']);
-
- usleep(100_000);
+ $this->handler->push($this->testQueue1, 'success', ['test' => 'data1']);
+ $this->handler->push($this->testQueue2, 'success', ['test' => 'data2']);
- $job1 = $this->handler->pop('test-queue-1', ['default']);
- $job2 = $this->handler->pop('test-queue-2', ['default']);
+ $job1 = $this->popEventually($this->testQueue1, ['default']);
+ $job2 = $this->popEventually($this->testQueue2, ['default']);
$this->assertInstanceOf(QueueJob::class, $job1);
$this->assertInstanceOf(QueueJob::class, $job2);
@@ -356,15 +354,13 @@ public function testClearAll(): void
$channel->basic_nack($job2->amqpDeliveryTag, false, true);
}
- usleep(100_000);
-
// Clear all queues
$result = $this->handler->clear();
$this->assertTrue($result);
// Verify queues are empty by attempting to pop
- $jobAfter1 = $this->handler->pop('test-queue-1', ['default']);
- $jobAfter2 = $this->handler->pop('test-queue-2', ['default']);
+ $jobAfter1 = $this->handler->pop($this->testQueue1, ['default']);
+ $jobAfter2 = $this->handler->pop($this->testQueue2, ['default']);
$this->assertNull($jobAfter1);
$this->assertNull($jobAfter2);
@@ -385,4 +381,36 @@ private function isRabbitMQAvailable(): bool
{
return class_exists(AMQPConnectionFactory::class);
}
+
+ private function deleteDeclaredResources(): void
+ {
+ $channel = self::getPrivateProperty($this->handler, 'channel');
+
+ foreach (array_keys(self::getPrivateProperty($this->handler, 'declaredQueues')) as $queue) {
+ $channel->queue_delete($queue);
+ }
+
+ foreach (array_keys(self::getPrivateProperty($this->handler, 'declaredExchanges')) as $exchange) {
+ $channel->exchange_delete($exchange);
+ }
+ }
+
+ /**
+ * @param list $priorities
+ */
+ private function popEventually(string $queue, array $priorities, float $timeout = 1.0): ?QueueJob
+ {
+ $deadline = microtime(true) + $timeout;
+
+ do {
+ $job = $this->handler->pop($queue, $priorities);
+ if ($job !== null) {
+ return $job;
+ }
+
+ usleep(10_000);
+ } while (microtime(true) < $deadline);
+
+ return null;
+ }
}
diff --git a/tests/_support/CLITestCase.php b/tests/_support/CLITestCase.php
index f0d405d..f134062 100644
--- a/tests/_support/CLITestCase.php
+++ b/tests/_support/CLITestCase.php
@@ -14,6 +14,7 @@
namespace Tests\Support;
use CodeIgniter\CLI\CLI;
+use CodeIgniter\Test\Filters\CITestStreamFilter;
use CodeIgniter\Test\ReflectionHelper;
abstract class CLITestCase extends TestCase
@@ -22,6 +23,14 @@ abstract class CLITestCase extends TestCase
private array $lines = [];
+ protected function tearDown(): void
+ {
+ CITestStreamFilter::removeErrorFilter();
+ CITestStreamFilter::removeOutputFilter();
+
+ parent::tearDown();
+ }
+
protected function parseOutput(string $output): string
{
$this->lines = [];
diff --git a/tests/_support/Database/Seeds/TestRedisQueueSeeder.php b/tests/_support/Database/Seeds/TestRedisQueueSeeder.php
index 4fe4233..d14d86f 100644
--- a/tests/_support/Database/Seeds/TestRedisQueueSeeder.php
+++ b/tests/_support/Database/Seeds/TestRedisQueueSeeder.php
@@ -40,7 +40,10 @@ public function run(): void
throw new CriticalError('Queue: RedisException occurred with message (' . $e->getMessage() . ').');
}
- $redis->flushDB();
+ $keys = $redis->keys('queues:*');
+ if ($keys !== []) {
+ $redis->del($keys);
+ }
$jobQueue = new QueueJob([
'id' => '1234567890123456',