Class Yiisoft\Queue\Redis\QueueProvider

                public function __construct(
    private Redis $redis, //redis connection,
    private string $channelName = self::DEFAULT_CHANNEL_NAME,
) {}

            

                public function delete(string $id): void
{
    $this->checkConnection();
    $this->redis->zrem("$this->channelName.reserved", $id);
    $this->redis->zrem("$this->channelName.delayed", $id);
    $this->redis->hdel("$this->channelName.messages", $id);
    $this->redis->hdel("$this->channelName.attempts", $id);
}

            

                public function existInReserved(int $id): bool
{
    $this->checkConnection();
    $exist = $this->redis->hexists("$this->channelName.attempts", (string) $id);
    return is_bool($exist) ? $exist : false;
}

            

                public function existInWaiting(int $id): bool
{
    $this->checkConnection();
    $exist = $this->redis->hexists("$this->channelName.messages", (string) $id);
    return is_bool($exist) ? $exist : false;
}

            

                public function getChannelName(): string
{
    return $this->channelName;
}

            

                public function getId(): int
{
    $this->checkConnection();
    $id = $this->redis->incr("$this->channelName.message_id");
    if (is_int($id)) {
        return $id;
    }
    throw new RuntimeException('Unable to get message id.');
}

            

                public function getLastId(): int
{
    $this->checkConnection();
    $id = $this->redis->get("$this->channelName.message_id");
    return is_string($id) ? (int) $id : 0;
}

            

                public function pushMessage(string $message, array $meta = []): int
{
    $this->checkConnection();
    $id = $this->getId();
    $this->redis->hset("$this->channelName.messages", (string) $id, $message);
    /** @var array|bool|float|int|string|null $delay */
    $delay = $meta[DelayEnvelope::META_DELAY_SECONDS] ?? 0;
    $delay = is_int($delay) || is_float($delay) ? $delay : 0;
    if ($delay > 0) {
        $this->redis->zadd("$this->channelName.delayed", time() + $delay, $id);
    } else {
        $this->redis->lpush("$this->channelName.waiting", $id);
    }
    return $id;
}

            

                public function reserve(int $timeout = 0): ?Reserve
{
    $this->checkConnection();
    // Moves delayed and reserved jobs into waiting list with lock for one second
    try {
        if ($this->redis->set("$this->channelName.moving_lock", 'true', ['NX', 'EX', 1])) {
            $this->moveExpired("$this->channelName.delayed");
            $this->moveExpired("$this->channelName.reserved");
        }
    } finally {
        $this->redis->del("$this->channelName.moving_lock");
    }
    $result = $this->redis->brpop("$this->channelName.waiting", $timeout);
    if (null === $result || !isset($result[1])) {
        return null;
    }
    $id = $result[1];
    if (!is_string($id)) {
        return null;
    }
    $payload = $this->redis->hget("$this->channelName.messages", $id);
    if (!is_string($payload)) {
        return null;
    }
    $this->redis->zRem("$this->channelName.reserved", time(), $id);
    $this->redis->hincrby("$this->channelName.attempts", $id, 1);
    return new Reserve((int) $id, $payload);
}

            

                public function withChannelName(string $channelName): QueueProviderInterface
{
    if ($this->channelName === $channelName) {
        return $this;
    }
    return new self($this->redis, $channelName);
}