Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Setup Psalm error level to 1 #505

Merged
merged 14 commits into from
Dec 4, 2023
Prev Previous commit
Next Next commit
Fixed redis driver issues
s1lver committed Nov 24, 2023
commit f126104f7227243ba29d98d939b006073506c3df
10 changes: 5 additions & 5 deletions src/drivers/redis/InfoAction.php
Original file line number Diff line number Diff line change
@@ -38,20 +38,20 @@ public function run(): void
$delayed = $this->queue->redis->zcount("$prefix.delayed", '-inf', '+inf');
$reserved = $this->queue->redis->zcount("$prefix.reserved", '-inf', '+inf');
$total = $this->queue->redis->get("$prefix.message_id");
$done = $total - $waiting - $delayed - $reserved;
$done = (int)$total - (int)$waiting - (int)$delayed - (int)$reserved;

Console::output($this->format('Jobs', Console::FG_GREEN));

Console::stdout($this->format('- waiting: ', Console::FG_YELLOW));
Console::output($waiting);
Console::output((string)$waiting);

Console::stdout($this->format('- delayed: ', Console::FG_YELLOW));
Console::output($delayed);
Console::output((string)$delayed);

Console::stdout($this->format('- reserved: ', Console::FG_YELLOW));
Console::output($reserved);
Console::output((string)$reserved);

Console::stdout($this->format('- done: ', Console::FG_YELLOW));
Console::output($done);
Console::output((string)$done);
}
}
12 changes: 11 additions & 1 deletion src/drivers/redis/Queue.php
Original file line number Diff line number Diff line change
@@ -63,6 +63,10 @@ public function run(bool $repeat, int $timeout = 0): ?int
while ($canContinue()) {
if (($payload = $this->reserve($timeout)) !== null) {
[$id, $message, $ttr, $attempt] = $payload;
/**
* @psalm-var int|string $id
* @psalm-var string $message
*/
if ($this->handleMessage($id, $message, (int)$ttr, (int)$attempt)) {
$this->delete($id);
}
@@ -103,6 +107,7 @@ public function clear(): void
while (!$this->redis->set("$this->channel.moving_lock", true, 'NX')) {
usleep(10000);
}
/** @psalm-suppress MixedArgument */
$this->redis->executeCommand('DEL', $this->redis->keys("$this->channel.*"));
}

@@ -147,6 +152,7 @@ protected function reserve(int $timeout): ?array
if (!$timeout) {
$id = $this->redis->rpop("$this->channel.waiting");
} elseif ($result = $this->redis->brpop("$this->channel.waiting", $timeout)) {
/** @psalm-var array $result */
$id = $result[1];
}
if (!$id) {
@@ -158,7 +164,10 @@ protected function reserve(int $timeout): ?array
return null;
}

/** @psalm-suppress PossiblyUndefinedArrayOffset */
/**
* @psalm-suppress PossiblyUndefinedArrayOffset
* @psalm-var string $payload
*/
[$ttr, $message] = explode(';', $payload, 2);
$this->redis->zadd("$this->channel.reserved", time() + (int)$ttr, $id);
$attempt = $this->redis->hincrby("$this->channel.attempts", $id, 1);
@@ -201,6 +210,7 @@ protected function pushMessage(string $payload, int $ttr, int $delay, mixed $pri
throw new NotSupportedException('Job priority is not supported in the driver.');
}

/** @var string|int $id */
$id = $this->redis->incr("$this->channel.message_id");
$this->redis->hset("$this->channel.messages", $id, "$ttr;$payload");
if (!$delay) {