2022-11-10 06:01:43 +13:00
|
|
|
<?php
|
|
|
|
|
|
|
|
require_once __DIR__ . '/init.php';
|
|
|
|
|
2022-11-16 07:13:17 +13:00
|
|
|
use Appwrite\Event\Func;
|
2022-12-13 20:35:05 +13:00
|
|
|
use Appwrite\Event\Usage;
|
2022-11-10 06:01:43 +13:00
|
|
|
use Swoole\Runtime;
|
|
|
|
use Utopia\App;
|
|
|
|
use Utopia\Cache\Adapter\Sharding;
|
|
|
|
use Utopia\Cache\Cache;
|
2022-11-16 23:40:41 +13:00
|
|
|
use Utopia\CLI\Console;
|
2022-11-10 06:01:43 +13:00
|
|
|
use Utopia\Config\Config;
|
|
|
|
use Utopia\Database\Database;
|
2022-11-13 03:35:42 +13:00
|
|
|
use Utopia\Database\Document;
|
2022-11-16 23:40:41 +13:00
|
|
|
use Utopia\Queue\Adapter\Swoole;
|
2022-11-13 03:35:42 +13:00
|
|
|
use Utopia\Queue\Message;
|
2022-11-10 06:01:43 +13:00
|
|
|
use Utopia\Queue\Server;
|
|
|
|
use Utopia\Registry\Registry;
|
2022-11-16 23:40:41 +13:00
|
|
|
use Utopia\Logger\Log;
|
2022-11-17 02:34:11 +13:00
|
|
|
use Utopia\Logger\Logger;
|
2022-11-17 08:39:35 +13:00
|
|
|
use Utopia\Pools\Group;
|
2022-11-16 23:40:41 +13:00
|
|
|
|
|
|
|
Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
|
2022-11-10 06:01:43 +13:00
|
|
|
|
|
|
|
global $register;
|
|
|
|
|
|
|
|
Server::setResource('register', fn() => $register);
|
|
|
|
|
|
|
|
Server::setResource('dbForConsole', function (Cache $cache, Registry $register) {
|
|
|
|
$pools = $register->get('pools');
|
2022-11-16 17:17:46 +13:00
|
|
|
$database = $pools
|
2022-11-10 06:01:43 +13:00
|
|
|
->get('console')
|
|
|
|
->pop()
|
|
|
|
->getResource()
|
|
|
|
;
|
|
|
|
|
2022-11-16 17:17:46 +13:00
|
|
|
$adapter = new Database($database, $cache);
|
|
|
|
$adapter->setNamespace('console');
|
2022-11-10 06:01:43 +13:00
|
|
|
|
2022-11-16 17:17:46 +13:00
|
|
|
return $adapter;
|
2022-11-10 06:01:43 +13:00
|
|
|
}, ['cache', 'register']);
|
|
|
|
|
2022-11-13 03:35:42 +13:00
|
|
|
Server::setResource('dbForProject', function (Cache $cache, Registry $register, Message $message, Database $dbForConsole) {
|
2022-11-16 17:17:46 +13:00
|
|
|
$payload = $message->getPayload() ?? [];
|
|
|
|
$project = new Document($payload['project'] ?? []);
|
2022-11-13 03:35:42 +13:00
|
|
|
|
|
|
|
if ($project->isEmpty() || $project->getId() === 'console') {
|
|
|
|
return $dbForConsole;
|
|
|
|
}
|
|
|
|
|
|
|
|
$pools = $register->get('pools');
|
2022-11-16 17:17:46 +13:00
|
|
|
$database = $pools
|
2022-11-13 03:35:42 +13:00
|
|
|
->get($project->getAttribute('database'))
|
|
|
|
->pop()
|
|
|
|
->getResource()
|
|
|
|
;
|
|
|
|
|
2022-11-16 17:17:46 +13:00
|
|
|
$adapter = new Database($database, $cache);
|
|
|
|
$adapter->setNamespace('_' . $project->getInternalId());
|
|
|
|
return $adapter;
|
2022-11-13 03:35:42 +13:00
|
|
|
}, ['cache', 'register', 'message', 'dbForConsole']);
|
|
|
|
|
2022-11-10 06:01:43 +13:00
|
|
|
Server::setResource('cache', function (Registry $register) {
|
|
|
|
$pools = $register->get('pools');
|
|
|
|
$list = Config::getParam('pools-cache', []);
|
|
|
|
$adapters = [];
|
|
|
|
|
|
|
|
foreach ($list as $value) {
|
|
|
|
$adapters[] = $pools
|
|
|
|
->get($value)
|
|
|
|
->pop()
|
|
|
|
->getResource()
|
|
|
|
;
|
|
|
|
}
|
|
|
|
|
|
|
|
return new Cache(new Sharding($adapters));
|
|
|
|
}, ['register']);
|
|
|
|
|
2022-11-16 19:08:01 +13:00
|
|
|
Server::setResource('queueForFunctions', function (Registry $register) {
|
2022-11-16 07:13:17 +13:00
|
|
|
$pools = $register->get('pools');
|
2022-11-16 17:17:46 +13:00
|
|
|
return new Func(
|
|
|
|
$pools
|
|
|
|
->get('queue')
|
|
|
|
->pop()
|
|
|
|
->getResource()
|
2022-11-16 18:30:57 +13:00
|
|
|
);
|
2022-11-16 07:13:17 +13:00
|
|
|
}, ['register']);
|
|
|
|
|
2022-12-13 20:35:05 +13:00
|
|
|
Server::setResource('queueForUsage', function (Registry $register) {
|
|
|
|
$pools = $register->get('pools');
|
|
|
|
return new Usage(
|
|
|
|
$pools
|
|
|
|
->get('queue')
|
|
|
|
->pop()
|
|
|
|
->getResource()
|
|
|
|
);
|
2022-11-10 06:01:43 +13:00
|
|
|
}, ['register']);
|
|
|
|
|
2022-12-13 20:35:05 +13:00
|
|
|
Server::setResource('logger', function ($register) {
|
|
|
|
return $register->get('logger');
|
2022-11-16 17:17:46 +13:00
|
|
|
}, ['register']);
|
2022-11-10 06:01:43 +13:00
|
|
|
|
2022-11-17 08:39:35 +13:00
|
|
|
Server::setResource('pools', function ($register) {
|
|
|
|
return $register->get('pools');
|
|
|
|
}, ['register']);
|
|
|
|
|
2022-11-10 06:01:43 +13:00
|
|
|
$pools = $register->get('pools');
|
2022-11-16 05:03:42 +13:00
|
|
|
$connection = $pools->get('queue')->pop()->getResource();
|
2022-11-10 06:01:43 +13:00
|
|
|
$workerNumber = swoole_cpu_num() * intval(App::getEnv('_APP_WORKER_PER_CORE', 6));
|
2022-12-12 22:37:10 +13:00
|
|
|
|
2022-11-16 23:54:21 +13:00
|
|
|
if (empty(App::getEnv('QUEUE'))) {
|
2022-11-16 23:40:41 +13:00
|
|
|
throw new Exception('Please configure "QUEUE" environemnt variable.');
|
|
|
|
}
|
|
|
|
|
|
|
|
$adapter = new Swoole($connection, $workerNumber, App::getEnv('QUEUE'));
|
|
|
|
$server = new Server($adapter);
|
|
|
|
|
2022-11-17 08:39:35 +13:00
|
|
|
$server
|
|
|
|
->shutdown()
|
|
|
|
->inject('pools')
|
|
|
|
->action(function (Group $pools) {
|
|
|
|
$pools->reclaim();
|
|
|
|
});
|
|
|
|
|
2022-11-16 23:40:41 +13:00
|
|
|
$server
|
|
|
|
->error()
|
|
|
|
->inject('error')
|
|
|
|
->inject('logger')
|
2022-11-17 02:34:11 +13:00
|
|
|
->action(function (Throwable $error, Logger $logger) {
|
2022-11-16 23:40:41 +13:00
|
|
|
$version = App::getEnv('_APP_VERSION', 'UNKNOWN');
|
|
|
|
|
|
|
|
if ($error instanceof PDOException) {
|
|
|
|
throw $error;
|
|
|
|
}
|
|
|
|
|
|
|
|
if ($error->getCode() >= 500 || $error->getCode() === 0) {
|
|
|
|
$log = new Log();
|
|
|
|
|
|
|
|
$log->setNamespace("appwrite-worker");
|
|
|
|
$log->setServer(\gethostname());
|
|
|
|
$log->setVersion($version);
|
|
|
|
$log->setType(Log::TYPE_ERROR);
|
|
|
|
$log->setMessage($error->getMessage());
|
2022-11-17 02:34:11 +13:00
|
|
|
$log->setAction('appwrite-queue-' . App::getEnv('QUEUE'));
|
2022-11-16 23:40:41 +13:00
|
|
|
$log->addTag('verboseType', get_class($error));
|
|
|
|
$log->addTag('code', $error->getCode());
|
|
|
|
$log->addExtra('file', $error->getFile());
|
|
|
|
$log->addExtra('line', $error->getLine());
|
|
|
|
$log->addExtra('trace', $error->getTraceAsString());
|
|
|
|
$log->addExtra('detailedTrace', $error->getTrace());
|
|
|
|
$log->addExtra('roles', \Utopia\Database\Validator\Authorization::$roles);
|
|
|
|
|
|
|
|
$isProduction = App::getEnv('_APP_ENV', 'development') === 'production';
|
|
|
|
$log->setEnvironment($isProduction ? Log::ENVIRONMENT_PRODUCTION : Log::ENVIRONMENT_STAGING);
|
|
|
|
|
|
|
|
$logger->addLog($log);
|
|
|
|
}
|
|
|
|
|
|
|
|
Console::error('[Error] Type: ' . get_class($error));
|
|
|
|
Console::error('[Error] Message: ' . $error->getMessage());
|
|
|
|
Console::error('[Error] File: ' . $error->getFile());
|
|
|
|
Console::error('[Error] Line: ' . $error->getLine());
|
2022-11-16 23:54:21 +13:00
|
|
|
});
|