<?php
 namespace App\Cron\Job; class ReportsDataPopulate extends \App\Cron\Job\Utils\LongRunningCronJobAbstract { protected $category = 'operational'; private $stopAfterTransactionsCount = 30000; public function validateSettings(\App\Entity\Messages $msgs, array $fields) { if ($fields['scheduleOptions']['type']=='interval' ) { list($h, $m, $s) = explode(':', $fields['scheduleOptions']['repeatPeriod']); $repeatPeriod = $h*3600 + $m*60 + $s; if ($repeatPeriod<120) { $msgs->addWarningText("The process should not run more often than 2 minutes.", 'type'); } if ($repeatPeriod>7200) { $msgs->addWarningText("The process should not run more seldom than 2 hours.", 'type'); } } else { $msgs->addWarningText("The process should be set as `multiple times per day`.", 'type'); } } protected function executeJob() { if (!in_array(\App\ReportsData\Params::getInstance()->getStatus(), ['running', 'pending'])) { $currSts = \App\ReportsData\Params::getInstance()->getStatus(); if ($currSts=='blocked') { $blockedUntil = \App\ReportsData\Params::getInstance()->get('blockedUntil'); $blockingApp = \App\ReportsData\Params::getInstance()->get('blockingApp'); $this->logger->idle("The cron will not start - the current RD queue status is `{$currSts}` until {$blockedUntil} (blocking app: {$blockingApp})."); } else { $this->logger->idle("The cron will not start - the current RD queue status is `{$currSts}`."); } return ; } $bakCron = new \App\Cron\Job\TinaBackup(); if ($bakCron->isAnotherInstanceRunning()) { $this->logger->idle("The cron will not execute as there is a TINA backup cron in progress."); return ; } \App\ReportsData\Params::getAdmin()->saveStatusToRunning(); $stopProcess = false; $queueMng = \App\ReportsData\ChangesQueue\Manager::getInstance(); \DB::startTransaction(); $movedTransactionsList = $queueMng->moveDataFromFastQueueToMainQueue(); \DB::commitTransaction(); if ($movedTransactionsList===false) { $this->logger->warning( "A deadlock was found while copying the content from the fast queue to the main queue. This warning can be ignored, the next execution of the cron should run normally. The current process will stop now." ); $stopProcess = true; } if ($movedTransactionsList) { $movedRecordsLog = []; foreach ($movedTransactionsList as $row) { $movedRecordsLog[] = $row['entity'].' #'.$row['id']; } $this->logger->info( "Moved ".count($movedRecordsLog)." transactions from fast queue to main queue: " . implode(', ', $movedRecordsLog) ); } $processedTransCount = 0; if (!$stopProcess) { $totalRecordsInitialCount = $queueMng->getTotalCountMain(); if (!$totalRecordsInitialCount) { $this->logger->idle("The RD queue is empty."); $stopProcess = true; } else { $this->logger->info(sprintf("There are %s records in the RD queue.", $totalRecordsInitialCount)); } $totalTransCount = min($totalRecordsInitialCount, $this->stopAfterTransactionsCount); $executedTransactions = []; } $displayedFirstRecord = false; while (!$stopProcess) { if ($processedTransCount>=$this->stopAfterTransactionsCount) { $this->logger->info("The cron will stop now - it is set to automatically stop after {$this->stopAfterTransactionsCount} transactions."); $stopProcess = true; } $currentStatus = \App\ReportsData\Params::getInstance()->getStatus(); if (!in_array($currentStatus, ['running', 'pending'])) { $stopProcess = true; $this->logger->info("The cron will stop now - the RD queue status was meanwhile set to `{$currentStatus}`."); } if (!$stopProcess) { $this->updateStatus([ 'progress'=>$processedTransCount/$totalTransCount ]); try { \DB::startTransaction(); $oldestRecord = $queueMng->getOldestRecordMainQueue(); if (!$displayedFirstRecord) { $this->logger->info(sprintf("The first record in the queue is: %s #%s.", $oldestRecord['entity'], $oldestRecord['id'])); $displayedFirstRecord = true; } if (!$oldestRecord) { $stopProcess = true; $this->logger->info("The RD queue is now empty - the cron will stop."); } else { $queueMng->processRecord($oldestRecord['entity'], $oldestRecord['id']); } \DB::commitTransaction(); if ($oldestRecord) { $processedTransCount++; $executedTransactions[] = $oldestRecord['entity'].' #'.$oldestRecord['id']; } } catch (\DBException $e) { \DB::rollbackTransaction(); $stopProcess = true; $this->logger->critical('Found a DB error while processing: '.$e->getMessage().'. The current process will stop.'); } if (!($processedTransCount%1000)) { $this->logger->info(sprintf('Processed %s records since current cron process execution.', $processedTransCount)); } } } if ($processedTransCount) { $this->logger->success( "Successfully processed {$processedTransCount} transactions." ); $this->logger->info( "Processed transactions: " . implode(', ', $executedTransactions) ); } $currSts = \App\ReportsData\Params::getInstance()->getStatus(); if ($currSts=='running') { \App\ReportsData\Params::getAdmin()->saveStatusToPending(); } } } 