diff --git a/app/worker.php b/app/worker.php index 99f4154c98..32ce35e033 100644 --- a/app/worker.php +++ b/app/worker.php @@ -16,7 +16,6 @@ use Appwrite\Event\Phone; use Appwrite\Event\Usage; use Appwrite\Event\UsageDump; use Appwrite\Platform\Appwrite; -use Appwrite\Utopia\Pools\Connections; use Swoole\Runtime; use Utopia\App; use Utopia\Cache\Adapter\Sharding; @@ -37,32 +36,25 @@ use Utopia\Pools\Group; use Utopia\Queue\Connection; Authorization::disable(); -Runtime::enableCoroutine(); +Runtime::enableCoroutine(SWOOLE_HOOK_ALL); Server::setResource('register', fn () => $register); -Server::setResource('connections', function () { - return new Connections(); -}); - -Server::setResource('dbForConsole', function (Cache $cache, Registry $register, Connections $connections) { +Server::setResource('dbForConsole', function (Cache $cache, Registry $register) { $pools = $register->get('pools'); - - $connection = $pools + $database = $pools ->get('console') - ->pop(); - - $connections->add($connection); - - $database = $connection->getResource(); + ->pop() + ->getResource() + ; $adapter = new Database($database, $cache); $adapter->setNamespace('_console'); return $adapter; -}, ['cache', 'register', 'connections']); +}, ['cache', 'register']); -Server::setResource('dbForProject', function (Cache $cache, Registry $register, Message $message, Database $dbForConsole, Connections $connections) { +Server::setResource('dbForProject', function (Cache $cache, Registry $register, Message $message, Database $dbForConsole) { $payload = $message->getPayload() ?? []; $project = new Document($payload['project'] ?? []); @@ -71,19 +63,16 @@ Server::setResource('dbForProject', function (Cache $cache, Registry $register, } $pools = $register->get('pools'); - - $connection = $pools + $database = $pools ->get($project->getAttribute('database')) - ->pop(); - - $database = $connection->getResource(); - - $connections->add($connection); + ->pop() + ->getResource() + ; $adapter = new Database($database, $cache); $adapter->setNamespace('_' . $project->getInternalId()); return $adapter; -}, ['cache', 'register', 'message', 'dbForConsole', 'connections']); +}, ['cache', 'register', 'message', 'dbForConsole']); Server::setResource('project', function (Message $message, Database $dbForConsole) { $payload = $message->getPayload() ?? []; @@ -92,14 +81,14 @@ Server::setResource('project', function (Message $message, Database $dbForConsol if ($project->getId() === 'console') { return $project; } - return $dbForConsole->getDocument('projects', $project->getId()); + ; }, ['message', 'dbForConsole']); -Server::setResource('getProjectDB', function (Group $pools, Database $dbForConsole, Cache $cache, Connections $connections) { +Server::setResource('getProjectDB', function (Group $pools, Database $dbForConsole, $cache) { $databases = []; // TODO: @Meldiron This should probably be responsibility of utopia-php/pools - return function (Document $project) use ($pools, $dbForConsole, $cache, $connections, &$databases): Database { + return function (Document $project) use ($pools, $dbForConsole, $cache, &$databases): Database { if ($project->isEmpty() || $project->getId() === 'console') { return $dbForConsole; } @@ -112,13 +101,10 @@ Server::setResource('getProjectDB', function (Group $pools, Database $dbForConso return $database; } - $dbConnection = $pools + $dbAdapter = $pools ->get($databaseName) - ->pop(); - - $dbAdapter = $dbConnection->getResource(); - - $connections->add($dbConnection); + ->pop() + ->getResource(); $database = new Database($dbAdapter, $cache); @@ -128,7 +114,7 @@ Server::setResource('getProjectDB', function (Group $pools, Database $dbForConso return $database; }; -}, ['pools', 'dbForConsole', 'cache', 'connections']); +}, ['pools', 'dbForConsole', 'cache']); Server::setResource('abuseRetention', function () { return DateTime::addSeconds(new \DateTime(), -1 * App::getEnv('_APP_MAINTENANCE_RETENTION_ABUSE', 86400)); @@ -142,104 +128,82 @@ Server::setResource('executionRetention', function () { return DateTime::addSeconds(new \DateTime(), -1 * App::getEnv('_APP_MAINTENANCE_RETENTION_EXECUTION', 1209600)); }); -Server::setResource('cache', function (Registry $register, Connections $connections) { +Server::setResource('cache', function (Registry $register) { $pools = $register->get('pools'); $list = Config::getParam('pools-cache', []); $adapters = []; foreach ($list as $value) { - $connection = $pools + $adapters[] = $pools ->get($value) - ->pop(); - - $connections->add($connection); - - $adapters[] = $connection->getResource(); + ->pop() + ->getResource() + ; } return new Cache(new Sharding($adapters)); -}, ['register', 'connections']); - +}, ['register']); Server::setResource('log', fn() => new Log()); - Server::setResource('queueForUsage', function (Connection $queue) { return new Usage($queue); }, ['queue']); - Server::setResource('queueForUsageDump', function (Connection $queue) { return new UsageDump($queue); }, ['queue']); - Server::setResource('queue', function (Group $pools) { return $pools->get('queue')->pop()->getResource(); }, ['pools']); - Server::setResource('queueForDatabase', function (Connection $queue) { return new EventDatabase($queue); }, ['queue']); - Server::setResource('queueForMessaging', function (Connection $queue) { return new Phone($queue); }, ['queue']); - Server::setResource('queueForMails', function (Connection $queue) { return new Mail($queue); }, ['queue']); - Server::setResource('queueForBuilds', function (Connection $queue) { return new Build($queue); }, ['queue']); - Server::setResource('queueForDeletes', function (Connection $queue) { return new Delete($queue); }, ['queue']); - Server::setResource('queueForEvents', function (Connection $queue) { return new Event($queue); }, ['queue']); - Server::setResource('queueForAudits', function (Connection $queue) { return new Audit($queue); }, ['queue']); - Server::setResource('queueForFunctions', function (Connection $queue) { return new Func($queue); }, ['queue']); - Server::setResource('queueForCertificates', function (Connection $queue) { return new Certificate($queue); }, ['queue']); - Server::setResource('queueForMigrations', function (Connection $queue) { return new Migration($queue); }, ['queue']); - Server::setResource('logger', function (Registry $register) { return $register->get('logger'); }, ['register']); - Server::setResource('pools', function (Registry $register) { return $register->get('pools'); }, ['register']); - Server::setResource('getFunctionsDevice', function () { return function (string $projectId) { return getDevice(APP_STORAGE_FUNCTIONS . '/app-' . $projectId); }; }); - Server::setResource('getFilesDevice', function () { return function (string $projectId) { return getDevice(APP_STORAGE_UPLOADS . '/app-' . $projectId); }; }); - Server::setResource('getBuildsDevice', function () { return function (string $projectId) { return getDevice(APP_STORAGE_BUILDS . '/app-' . $projectId); }; }); - Server::setResource('getCacheDevice', function () { return function (string $projectId) { return getDevice(APP_STORAGE_CACHE . '/app-' . $projectId); @@ -290,9 +254,9 @@ $worker = $platform->getWorker(); $worker ->shutdown() - ->inject('connections') - ->action(function (Connections $connections) { - $connections->reclaim(); + ->inject('pools') + ->action(function (Group $pools) { + $pools->reclaim(); }); $worker @@ -301,10 +265,7 @@ $worker ->inject('logger') ->inject('log') ->inject('project') - ->inject('connections') - ->action(function (Throwable $error, ?Logger $logger, Log $log, Document $project, Connections $connections) { - $connections->reclaim(); - + ->action(function (Throwable $error, ?Logger $logger, Log $log, Document $project) { $version = App::getEnv('_APP_VERSION', 'UNKNOWN'); if ($error instanceof PDOException) {