Elgato_dark/appwrite
1
0
Fork 0
mirror of https://github.com/appwrite/appwrite synced 2026-05-24 01:18:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

feat: implement scheduledAt in schedule

Browse source
This commit is contained in:
loks0n 2024-06-11 13:39:38 +01:00
parent b8b81a9bd1
commit 7e8f72d267
1 changed files with 25 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
src/Appwrite/Platform/Tasks/ScheduleFunctions.php
View file

@ -40,32 +40,39 @@ class ScheduleFunctions extends ScheduleBase
$delayedExecutions = []; // Group executions with same delay to share one coroutine $delayedExecutions = []; // Group executions with same delay to share one coroutine
foreach ($this->schedules as $key => $schedule) { foreach ($this->schedules as $scheduleKey => $schedule) {
$cron = new CronExpression($schedule['schedule']); // TODO: Allow schedule to be DateTime, like ScheduleMessaging.php if (CronExpression::isValidExpression($schedule['schedule'])) {
$nextDate = $cron->getNextRunDate(); $cron = new CronExpression($schedule['schedule']);
$next = DateTime::format($nextDate); $nextDate = $cron->getNextRunDate();
} else {
try {
$nextDate = new \DateTime($schedule['schedule']);
$schedule['delete'] = true;
} catch (\Exception) {
Console::error('Failed to parse schedule: ' . $schedule['schedule']);
continue;
}
}
$next = DateTime::format($nextDate);
$currentTick = $next < $timeFrame; $currentTick = $next < $timeFrame;
if (!$currentTick) { if (!$currentTick) {
continue; continue;
} }
$total++; $total += 1;
$delay = $nextDate->getTimestamp() - \time(); // Time to wait from now until execution needs to be queued
$promiseStart = \time(); // in seconds
$executionStart = $nextDate->getTimestamp(); // in seconds
$delay = $executionStart - $promiseStart; // Time to wait from now until execution needs to be queued
if (!isset($delayedExecutions[$delay])) { if (!isset($delayedExecutions[$delay])) {
$delayedExecutions[$delay] = []; $delayedExecutions[$delay] = [];
} }
$delayedExecutions[$delay][] = $key; $delayedExecutions[$delay][] = $scheduleKey;
} }
foreach ($delayedExecutions as $delay => $scheduleKeys) { foreach ($delayedExecutions as $delay => $scheduleKeys) {
\go(function () use ($delay, $scheduleKeys, $pools) { \go(function () use ($delay, $scheduleKeys, $pools, $dbForConsole) {
\sleep($delay); // in seconds \sleep($delay); // in seconds
$queue = $pools->get('queue')->pop(); $queue = $pools->get('queue')->pop();
@ -76,7 +83,6 @@ class ScheduleFunctions extends ScheduleBase
if (!\array_key_exists($scheduleKey, $this->schedules)) { if (!\array_key_exists($scheduleKey, $this->schedules)) {
return; return;
} }
$schedule = $this->schedules[$scheduleKey]; $schedule = $this->schedules[$scheduleKey];
$queueForFunctions = new Func($connection); $queueForFunctions = new Func($connection);
@ -88,6 +94,13 @@ class ScheduleFunctions extends ScheduleBase
->setPath('/') ->setPath('/')
->setProject($schedule['project']) ->setProject($schedule['project'])
->trigger(); ->trigger();
if ($schedule['delete']) {
$dbForConsole->deleteDocument(
'schedules',
$schedule['$id'],
);
}
} }
$queue->reclaim(); $queue->reclaim();