* @author Alexander Pankratov * @copyright 2016-2024 Daniil Gentili * @copyright 2016-2024 Alexander Pankratov * @license https://opensource.org/license/apache-2-0 Apache 2.0 * @link https://github.com/danog/AsyncOrm AsyncOrm documentation */ namespace danog\AsyncOrm\Internal\Driver; use Amp\Redis\Connection\ReconnectingRedisLink; use Amp\Redis\RedisClient; use Amp\Sync\LocalKeyedMutex; use danog\AsyncOrm\DbArrayBuilder; use danog\AsyncOrm\Driver\DriverArray; use danog\AsyncOrm\Internal\Serializer\BoolString; use danog\AsyncOrm\Internal\Serializer\FloatString; use danog\AsyncOrm\Internal\Serializer\IntString; use danog\AsyncOrm\Internal\Serializer\Passthrough; use danog\AsyncOrm\KeyType; use danog\AsyncOrm\Serializer; use danog\AsyncOrm\Settings\RedisSettings; use danog\AsyncOrm\ValueType; use Revolt\EventLoop; use function Amp\Redis\createRedisConnector; /** * Redis database backend. * * @internal * * @template TKey as array-key * @template TValue * @extends DriverArray */ final class RedisArray extends DriverArray { /** @var array */ private static array $connections = []; private static ?LocalKeyedMutex $mutex = null; private readonly RedisClient $db; private readonly bool $castToInt; /** * @api * @param Serializer $serializer */ public function __construct(DbArrayBuilder $config, Serializer $serializer) { /** @var Serializer */ $serializer = match ($config->valueType) { ValueType::INT => new IntString, ValueType::FLOAT => new FloatString, ValueType::BOOL => new BoolString, ValueType::SCALAR, ValueType::OBJECT => $serializer, default => new Passthrough }; $this->castToInt = $config->keyType === KeyType::INT; parent::__construct($config, $serializer); self::$mutex ??= new LocalKeyedMutex; \assert($config->settings instanceof RedisSettings); $dbKey = $config->settings->getDbIdentifier(); $lock = self::$mutex->acquire($dbKey); try { if (!isset(self::$connections[$dbKey])) { self::$connections[$dbKey] = new RedisClient(new ReconnectingRedisLink(createRedisConnector($config->settings->config))); self::$connections[$dbKey]->ping(); } } finally { EventLoop::queue($lock->release(...)); } $this->db = self::$connections[$dbKey]; } #[\Override] public function set(string|int $key, mixed $value): void { /** @var string */ $value = $this->serializer->serialize($value); $this->db->set($this->config->table.':'.(string) $key, $value); } #[\Override] public function get(string|int $key): mixed { $key = (string) $key; $value = $this->db->get($this->config->table.':'.$key); if ($value !== null) { /** @var TValue */ $value = $this->serializer->deserialize($value); } return $value; } #[\Override] public function unset(string|int $key): void { $this->db->delete($this->config->table.':'.(string) $key); } /** * Get iterator. * * @return \Traversable */ #[\Override] public function getIterator(): \Traversable { $request = $this->db->scan($this->config->table.':*'); $len = \strlen($this->config->table)+1; foreach ($request as $key) { $sub = \substr($key, $len); if ($this->castToInt) { $sub = (int) $sub; } yield $sub => $this->serializer->deserialize($this->db->get($key)); } } /** * Count elements. * * @api * * @link https://php.net/manual/en/arrayiterator.count.php * @return int The number of elements or public properties in the associated * array or object, respectively. */ #[\Override] public function count(): int { return \iterator_count($this->db->scan($this->config->table.':*')); } /** * Clear all elements. */ #[\Override] public function clear(): void { $request = $this->db->scan($this->config->table.':*'); $keys = []; foreach ($request as $key) { $keys[] = $key; if (\count($keys) === 10) { // @codeCoverageIgnoreStart $this->db->delete(...$keys); $keys = []; // @codeCoverageIgnoreEnd } } if ($keys) { $this->db->delete(...$keys); } } }