Bugfixes and added a way to neatly stop the worker-tenders

Author: webdevvie

Command/AbstractWorker.php

@@ -94,7 +94,7 @@ abstract class AbstractWorker extends ContainerAwareCommand
      */
     protected function addDefaultConfiguration()
     {
-        $this->addOption('use-tube', null, InputOption::VALUE_OPTIONAL, 'What tube to send to');
+        $this->addOption('use-tube', null, InputOption::VALUE_OPTIONAL, 'What tube to work with');
     }
 
     /**

Command/StopWorkerTenderCommand.php

@@ -0,0 +1,93 @@
+<?php
+
+namespace Webdevvie\PheanstalkTaskQueueBundle\Command;
+
+use Webdevvie\PheanstalkTaskQueueBundle\Command\AbstractWorker;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+use Webdevvie\PheanstalkTaskQueueBundle\Entity\Task;
+
+/**
+ * Class StopWorkerTenderCommand
+ * Sends a signal to the worker tender to stop
+ *
+ * @package Webdevvie\PheanstalkTaskQueueBundle\Command
+ * @author John Bakker <me@johnbakker.name>
+ */
+class StopWorkerTenderCommand extends AbstractWorker
+{
+    /**
+     * {@inheritDoc}
+     *
+     * @return void
+     */
+    protected function configure()
+    {
+        $this->setName('taskqueue:stop-worker-tender')
+            ->setDescription('Signals the worker tender to stop and waits for it to stop');
+        $this->addDefaultConfiguration();
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * @param InputInterface $input
+     * @param OutputInterface $output
+     * @return void
+     * @throws \InvalidArgumentException
+     */
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $this->initialiseWorker($input, $output);
+        $processes = $this->findWorkerTenderProcessesForTube($this->tube);
+        if (count($processes) == 0) {
+            $output->writeln("<info>Found no processes</info>");
+            return;
+        }
+        $output->writeln('<info>Found ' . count($processes) . ' process(es)');
+        foreach ($processes as $process) {
+            $output->writeln('<info>sending kill to:</info> ' . $process);
+            exec("kill $process");
+        }
+        $output->write('<info>Waiting on processes to stop</info> ');
+        while (count($processes) > 0) {
+            $processes = $this->findWorkerTenderProcessesForTube($this->tube);
+            $output->write(".");
+            sleep(1);
+
+        }
+        $output->writeln("Done!");
+    }
+
+    /**
+     * Returns a list of processes that match the worker-tender
+     *
+     * @param string $tube
+     * @return array
+     */
+    private function findWorkerTenderProcessesForTube($tube)
+    {
+        $processes = array();
+        $command = "ps ax ";
+        $command .= "|grep php ";
+        $command .= "|grep -v stop ";
+        $command .= "|grep worker-tender ";
+        if ($tube !== '') {
+            $command .= "|grep " . escapeshellarg('use-tube=' . $tube);
+        }
+        $data = exec($command);
+        $lines = explode("\n", $data);
+        foreach ($lines as $line) {
+            $line = trim($line);
+            $parts = explode(" ", $line);
+            if ($parts[0] === '') {
+                continue;
+            }
+            $processes[] = intval($parts[0]);
+        }
+
+        return $processes;
+
+    }
+}

Command/Tender/ChildProcessContainer.php

@@ -197,7 +197,7 @@ public function isCleanedUp()
      */
     public function getOutputLinesAvailable()
     {
-        $outputLines = [];
+        $outputLines = array();
         return $outputLines;
     }
 
@@ -301,9 +301,9 @@ private function sendTERMSignal()
      */
     public function getReadyForBed()
     {
-        $busyStatusses = [ChildProcessContainer::STATUS_SLEEPY,
+        $busyStatusses = array(ChildProcessContainer::STATUS_SLEEPY,
             ChildProcessContainer::STATUS_BUSY_BUT_SLEEPY
-        ];
+        );
         if (in_array($this->status, $busyStatusses)) {
             //ready for bed
             return;

Command/WorkerTenderCommand.php

@@ -18,7 +18,7 @@
  * one time per "tube"* You can supply several parameters to this command. Please refer to those parameters
  * for more information about customising your worker-tender experience
  *
- * @author John Bakker <me@johnbakker.name
+ * @author John Bakker <me@johnbakker.name>
  */
 class WorkerTenderCommand extends AbstractWorker
 {
@@ -66,7 +66,7 @@ protected function configure()
                 'Max time to keep a worker in seconds',
                 6000000
             );
-        $this->addDefaultConfiguration();
+        $this->addOption('use-tube', null, InputOption::VALUE_REQUIRED, 'What tube to work with');
     }
 
     /**
@@ -235,8 +235,10 @@ public function moreOrLess($total, $busy, $available)
      */
     private function findDisposableWorkers()
     {
+        $disposableStatusses = array(ChildProcessContainer::STATUS_READY, ChildProcessContainer::STATUS_ALIVE);
         foreach ($this->family as &$child) {
-            if (in_array($child->status, array(ChildProcessContainer::STATUS_READY, ChildProcessContainer::STATUS_ALIVE))) {
+
+            if (in_array($child->status, $disposableStatusses)) {
                 if ($child->getAge() < 10) {
                     //less than ten seconds old keep it alive a bit to let its do its job
                     continue;

DependencyInjection/Configuration.php

@@ -7,8 +7,6 @@
 
 /**
  * This is the class that validates and merges configuration from your app/config files
- *
- * To learn more see {@link http://symfony.com/doc/current/cookbook/bundles/extension.html#cookbook-bundles-extension-config-class}
  */
 class Configuration implements ConfigurationInterface
 {

Service/TaskQueueService.php

@@ -86,8 +86,7 @@ public function getDefaultTube()
     public function getStatusOfTaskWithId($taskId)
     {
         $task = $this->taskRepo->find($taskId);
-        if(!($task instanceof Task))
-        {
+        if (!($task instanceof Task)) {
             //the task was not found
             return Task::STATUS_GONE;
         }
@@ -169,7 +168,7 @@ public function cleanUpTasks($timePeriod)
             t.created <= :older'
         );
         $query->setParameter('status', array(Task::STATUS_DONE));
-        $query->setParameter('older', date("Y-m-d H:i:s", time()-$timePeriod));
+        $query->setParameter('older', date("Y-m-d H:i:s", time() - $timePeriod));
         $query->execute();
     }
 
@@ -206,7 +205,9 @@ public function reserveTask($tube = null)
             throw new TaskQueueServiceException("Invalid format in TaskQueue {$tube} ");
         } catch (\ReflectionException $exception) {
             $this->beanstalk->delete($inTask);
-            throw new TaskQueueServiceException("Invalid format in TaskQueue {$tube} class ".$parts[0].' is unknown');
+            throw new TaskQueueServiceException(
+                "Invalid format in TaskQueue {$tube} class " . $parts[0] . ' is unknown'
+            );
         }
         if (!($taskObject instanceof TaskDescriptionInterface)) {
             $this->beanstalk->delete($inTask);
@@ -254,6 +255,8 @@ public function markFailed(WorkPackage $task)
     }
 
     /**
+     * Updates the task status
+     *
      * @param WorkPackage $task
      * @param string $status
      * @return void