mmucklo /
DtcQueueBundle
Complex classes like Loop often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes. You can also have a look at the cohesion graph to spot any un-connected, or weakly-connected components.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
While breaking up the class, it is a good idea to analyze how other classes use Loop, and based on these observations, apply Extract Interface, too.
| 1 | <?php |
||
| 17 | class Loop |
||
| 18 | { |
||
| 19 | /** @var Run $run */ |
||
| 20 | protected $run; |
||
| 21 | |||
| 22 | /** @var OutputInterface */ |
||
| 23 | protected $output; |
||
| 24 | |||
| 25 | /** @var LoggerInterface */ |
||
| 26 | protected $logger; |
||
| 27 | |||
| 28 | /** @var WorkerManager */ |
||
| 29 | protected $workerManager; |
||
| 30 | |||
| 31 | /** @var JobManagerInterface */ |
||
| 32 | protected $jobManager; |
||
| 33 | |||
| 34 | /** @var RunManager */ |
||
| 35 | protected $runManager; |
||
| 36 | |||
| 37 | /** @var int */ |
||
| 38 | protected $processTimeout; |
||
| 39 | |||
| 40 | 3 | public function __construct( |
|
| 49 | |||
| 50 | /** |
||
| 51 | * @return int |
||
| 52 | */ |
||
| 53 | public function getProcessTimeout() |
||
| 57 | |||
| 58 | /** |
||
| 59 | * @param int $processTimeout |
||
| 60 | */ |
||
| 61 | public function setProcessTimeout(int $processTimeout) |
||
| 65 | |||
| 66 | public function setLogger(LoggerInterface $logger) |
||
| 70 | |||
| 71 | 1 | public function setOutput(OutputInterface $output) |
|
| 75 | |||
| 76 | /** |
||
| 77 | * The current (last) run object. |
||
| 78 | * |
||
| 79 | * @return Run|null |
||
| 80 | */ |
||
| 81 | 3 | public function getRun() |
|
| 85 | |||
| 86 | /** |
||
| 87 | * @param float $start |
||
| 88 | */ |
||
| 89 | 2 | public function runJobById($start, $jobId) |
|
| 113 | |||
| 114 | /** |
||
| 115 | * @param float $start |
||
| 116 | * @param int $nanoSleep |
||
| 117 | * @param null|int $maxCount |
||
| 118 | * @param null|int $duration |
||
| 119 | */ |
||
| 120 | 3 | public function runLoop($start, $workerName, $methodName, $maxCount, $duration = null, $nanoSleep = 500000000) |
|
| 121 | { |
||
| 122 | 3 | $this->checkParameters($nanoSleep, $maxCount, $duration); |
|
| 123 | 3 | $this->workerManager->setLoggingFunc([$this, 'log']); |
|
| 124 | 3 | $this->runStart($start, $maxCount, $duration); |
|
| 125 | try { |
||
| 126 | 3 | $this->log('info', 'Staring up a new job...'); |
|
| 127 | |||
| 128 | 3 | $endTime = $this->getEndTime($duration); |
|
| 129 | 3 | $currentJob = 1; |
|
| 130 | 3 | $noMoreJobsToRun = false; |
|
| 131 | do { |
||
| 132 | 3 | $this->recordHeartbeat($start); |
|
| 133 | 3 | $job = $this->workerManager->run($workerName, $methodName, true, $this->run->getId()); |
|
| 134 | 3 | $this->runCurrentJob($job, $noMoreJobsToRun, $currentJob, $duration, $nanoSleep); |
|
| 135 | 3 | } while (!$this->isFinished($maxCount, $endTime, $currentJob, $noMoreJobsToRun)); |
|
| 136 | 3 | } catch (\Exception $e) { |
|
| 137 | // Uncaught error: possibly with QueueBundle itself |
||
| 138 | $this->log('critical', $e->getMessage(), $e->getTrace()); |
||
| 139 | } |
||
| 140 | 3 | $this->runStop($start); |
|
| 141 | |||
| 142 | 3 | return 0; |
|
| 143 | } |
||
| 144 | |||
| 145 | /** |
||
| 146 | * @param int $nanoSleep |
||
| 147 | * @param null|int $maxCount |
||
| 148 | * @param null|int $duration |
||
| 149 | * |
||
| 150 | * @throws \Exception |
||
| 151 | */ |
||
| 152 | 3 | private function checkParameters(&$nanoSleep, &$maxCount, &$duration) |
|
| 168 | |||
| 169 | /** |
||
| 170 | * @param int|null $duration |
||
| 171 | * |
||
| 172 | * @return null|\DateTime |
||
| 173 | */ |
||
| 174 | 3 | protected function getEndTime($duration) |
|
| 175 | { |
||
| 176 | 3 | $endTime = null; |
|
| 177 | 3 | if (null !== $duration) { |
|
| 178 | 2 | $interval = new \DateInterval("PT${duration}S"); |
|
| 179 | 2 | $endTime = clone $this->run->getStartedAt(); |
|
| 180 | 2 | $endTime->add($interval); |
|
| 181 | 2 | } |
|
| 182 | |||
| 183 | return $endTime; |
||
| 184 | } |
||
| 185 | |||
| 186 | /** |
||
| 187 | * @param Job|null $job |
||
| 188 | * @param bool $noMoreJobsToRun |
||
| 189 | * @param int $currentJob |
||
| 190 | * @param int|null $duration |
||
| 191 | * @param int $nanoSleep |
||
| 192 | */ |
||
| 193 | protected function runCurrentJob($job, &$noMoreJobsToRun, &$currentJob, $duration, $nanoSleep) |
||
| 194 | { |
||
| 195 | 3 | if (null !== $job) { |
|
| 196 | 3 | $noMoreJobsToRun = false; |
|
| 197 | 3 | $this->reportJob($job); |
|
| 198 | 3 | $this->updateProcessed($currentJob); |
|
| 199 | 3 | ++$currentJob; |
|
| 200 | 3 | } else { |
|
| 201 | 3 | if (!$noMoreJobsToRun) { |
|
| 202 | 3 | $this->log('info', 'No more jobs to run ('.($currentJob - 1).' processed so far).'); |
|
| 203 | 3 | $noMoreJobsToRun = true; |
|
| 204 | 3 | } |
|
| 205 | 3 | if (null !== $duration) { |
|
| 206 | 2 | $nanoSleepTime = function_exists('random_int') ? random_int(0, $nanoSleep) : mt_rand(0, $nanoSleep); |
|
| 207 | 2 | time_nanosleep(0, $nanoSleepTime); |
|
| 208 | 2 | } |
|
| 209 | } |
||
| 210 | 3 | } |
|
| 211 | |||
| 212 | /** |
||
| 213 | * @param $maxCount |
||
| 214 | * @param $duration |
||
| 215 | * @param $processTimeout |
||
| 216 | */ |
||
| 217 | public function checkMaxCountDuration(&$maxCount, &$duration, &$processTimeout) |
||
| 218 | { |
||
| 219 | 1 | if (null !== $duration && null !== $processTimeout && $duration >= $processTimeout) { |
|
| 220 | $this->log('info', "duration ($duration) >= to process timeout ($processTimeout), so doubling process timeout to: ".(2 * $processTimeout)); |
||
| 221 | $processTimeout *= 2; |
||
| 222 | } |
||
| 223 | |||
| 224 | 1 | if (null === $maxCount && null === $duration) { |
|
| 225 | 1 | $maxCount = 1; |
|
| 226 | 1 | } |
|
| 227 | 1 | } |
|
| 228 | |||
| 229 | /** |
||
| 230 | * Determine if the run loop is finished. |
||
| 231 | * |
||
| 232 | * @param int|null $maxCount |
||
| 233 | * @param int $currentJob |
||
| 234 | * @param \DateTime|null $endTime |
||
| 235 | * @param bool $noMoreJobsToRun |
||
| 236 | * |
||
| 237 | * @return bool |
||
| 238 | */ |
||
| 239 | protected function isFinished($maxCount, $endTime, $currentJob, $noMoreJobsToRun) |
||
| 240 | { |
||
| 241 | 3 | if ((null === $maxCount || $currentJob <= $maxCount)) { |
|
| 242 | 3 | if (null === $endTime) { // This means that there is a $maxCount as we force one or the other to be not null |
|
| 243 | 3 | if ($noMoreJobsToRun) { |
|
| 244 | 3 | return true; |
|
| 245 | } |
||
| 246 | |||
| 247 | 3 | return false; |
|
| 248 | } |
||
| 249 | 2 | $now = new \DateTime(); |
|
| 250 | 2 | if ($endTime > $now) { |
|
| 251 | 2 | return false; |
|
| 252 | } |
||
| 253 | 2 | } |
|
| 254 | |||
| 255 | 3 | return true; |
|
| 256 | } |
||
| 257 | |||
| 258 | /** |
||
| 259 | * @param float $start |
||
| 260 | */ |
||
| 261 | protected function recordHeartbeat($start) |
||
| 267 | |||
| 268 | protected function persistRun($action = 'persist') |
||
| 269 | { |
||
| 270 | 3 | if ($this->runManager instanceof BaseRunManager) { |
|
| 271 | 2 | $objectManager = $this->runManager->getObjectManager(); |
|
| 272 | 2 | $objectManager->$action($this->run); |
|
| 273 | 2 | $objectManager->flush(); |
|
| 274 | 2 | } |
|
| 275 | 3 | } |
|
| 276 | |||
| 277 | /** |
||
| 278 | * @param int $count |
||
| 279 | */ |
||
| 280 | protected function updateProcessed($count) |
||
| 285 | |||
| 286 | /** |
||
| 287 | * Sets up the runManager (document / entity persister) if appropriate. |
||
| 288 | * |
||
| 289 | * @param float $start |
||
| 290 | * @param int|null $maxCount |
||
| 291 | * @param int|null $duration |
||
| 292 | */ |
||
| 293 | protected function runStart($start, $maxCount = null, $duration = null) |
||
| 312 | |||
| 313 | /** |
||
| 314 | * @param int|null $start |
||
| 315 | */ |
||
| 316 | protected function runStop($start) |
||
| 327 | |||
| 328 | /** |
||
| 329 | * @param Job $job |
||
| 330 | */ |
||
| 331 | protected function reportJob(Job $job) |
||
| 343 | |||
| 344 | /** |
||
| 345 | * @param string $level |
||
| 346 | */ |
||
| 347 | public function log($level, $msg, array $context = []) |
||
| 364 | } |
||
| 365 |
