feat: use coroutines delay system
This commit is contained in:
parent
31d4932fdf
commit
f02bf6ac5a
1 changed files with 20 additions and 15 deletions
|
@ -26,6 +26,7 @@ class ScheduleExecutions extends ScheduleBase
|
||||||
$queue = $pools->get('queue')->pop();
|
$queue = $pools->get('queue')->pop();
|
||||||
$connection = $queue->getResource();
|
$connection = $queue->getResource();
|
||||||
$queueForFunctions = new Func($connection);
|
$queueForFunctions = new Func($connection);
|
||||||
|
$intervalEnd = (new \DateTime())->modify('+' . self::ENQUEUE_TIMER . ' seconds');
|
||||||
|
|
||||||
foreach ($this->schedules as $schedule) {
|
foreach ($this->schedules as $schedule) {
|
||||||
if (!$schedule['active']) {
|
if (!$schedule['active']) {
|
||||||
|
@ -38,13 +39,16 @@ class ScheduleExecutions extends ScheduleBase
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
$now = new \DateTime();
|
|
||||||
$scheduledAt = new \DateTime($schedule['schedule']);
|
$scheduledAt = new \DateTime($schedule['schedule']);
|
||||||
|
if ($scheduledAt <= $intervalEnd) {
|
||||||
if ($scheduledAt > $now) {
|
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
$delay = $scheduledAt->getTimestamp() - (new \DateTime())->getTimestamp();
|
||||||
|
|
||||||
|
\go(function () use ($queueForFunctions, $schedule, $delay) {
|
||||||
|
\sleep($delay);
|
||||||
|
|
||||||
$queueForFunctions
|
$queueForFunctions
|
||||||
->setType('schedule')
|
->setType('schedule')
|
||||||
// Set functionId instead of function as we don't have $dbForProject
|
// Set functionId instead of function as we don't have $dbForProject
|
||||||
|
@ -57,6 +61,7 @@ class ScheduleExecutions extends ScheduleBase
|
||||||
->setBody($schedule['data']['body'] ?? '')
|
->setBody($schedule['data']['body'] ?? '')
|
||||||
->setProject($schedule['project'])
|
->setProject($schedule['project'])
|
||||||
->trigger();
|
->trigger();
|
||||||
|
});
|
||||||
|
|
||||||
$dbForConsole->deleteDocument(
|
$dbForConsole->deleteDocument(
|
||||||
'schedules',
|
'schedules',
|
||||||
|
|
Loading…
Reference in a new issue