. * * @author Daniil Gentili * @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\Containers; use Amp\Sync\LocalMutex; use danog\AsyncOrm\DbArray; use danog\AsyncOrm\DbArrayBuilder; use danog\AsyncOrm\DbObject; use Revolt\EventLoop; use Traversable; use function Amp\async; /** * @template TKey as array-key * @template TValue as DbObject * @internal */ final class ObjectContainer { /** * @var array> */ private array $cache = []; /** * Cache cleanup watcher ID. */ private ?string $cacheCleanupId = null; private LocalMutex $mutex; public function __construct( /** @var DbArray */ public DbArray $inner, public DbArrayBuilder $config, public int $cacheTtl, ) { $this->mutex = new LocalMutex; } public function __sleep() { return ['inner']; } public function __wakeup(): void { $this->mutex = new LocalMutex; } public function startCacheCleanupLoop(): void { if ($this->cacheCleanupId !== null) { EventLoop::cancel($this->cacheCleanupId); } $this->cacheCleanupId = EventLoop::repeat( \max(1, $this->cacheTtl / 5), fn () => $this->flushCache(), ); } public function stopCacheCleanupLoop(): void { if ($this->cacheCleanupId !== null) { EventLoop::cancel($this->cacheCleanupId); $this->cacheCleanupId = null; } } /** * @param TKey $index * @return ?TValue */ public function get(string|int $index): mixed { if (isset($this->cache[$index])) { $obj = $this->cache[$index]; $ref = $obj->get(); if ($ref !== null) { $obj->ttl = \time() + $this->cacheTtl; return $ref; } // @codeCoverageIgnoreStart unset($this->cache[$index]); // @codeCoverageIgnoreEnd } $result = $this->inner->get($index); if (isset($this->cache[$index])) { return $this->cache[$index]->get(); } if ($result === null) { return null; } $this->cache[$index] = new ObjectReference( $result, \time() + $this->cacheTtl, $f = async( $result->initDb(...), $this, $index, $this->config ) ); $f->await(); return $result; } /** * @param TKey $key * @param TValue $value */ public function set(string|int $key, DbObject $value): void { if (isset($this->cache[$key]) && $this->cache[$key]->get() === $value) { return; } $value->initDb( $this, $key, $this->config ); $this->cache[$key] = new ObjectReference( $value, \time() + $this->cacheTtl, null ); $value->save(); } /** @param TKey $key */ public function unset(string|int $key): void { unset($this->cache[$key]); $this->inner->unset($key); } /** @return Traversable */ public function getIterator(): Traversable { foreach ($this->inner->getIterator() as $key => $value) { if (isset($this->cache[$key])) { $obj = $this->cache[$key]; $ref = $obj->get(); if ($ref !== null) { $obj->ttl = \time() + $this->cacheTtl; yield $key => $ref; continue; } } $this->cache[$key] = new ObjectReference( $value, \time() + $this->cacheTtl, $f = async( $value->initDb(...), $this, $key, $this->config ) ); $f->await(); yield $key => $value; } } public function count(): int { return $this->inner->count(); } public function clear(): void { $lock = $this->mutex->acquire(); $this->cache = []; $lock->release(); $this->inner->clear(); } /** * Flush all flushable keys. */ public function flushCache(): void { $lock = $this->mutex->acquire(); try { $now = \time(); $new = []; foreach ($this->cache as $key => $value) { if ($value->ttl <= $now) { $value->obj = null; } if ($value->get() !== null) { $new[$key] = $value; } } $this->cache = $new; } finally { EventLoop::queue($lock->release(...)); } } }