* Removed "simple:" from command names

* Changed behaviour of dotenv option to simplify
* Added option to stop working when queue is empty

Author: pdscopes

src/Command/Symfony/Restart.php

@@ -37,7 +37,7 @@ public function __construct(CacheInterface $cache)
     protected function configure()
     {
         $this
-            ->setName('simple:task-worker:restart')
+            ->setName('task-worker:restart')
             ->setDescription('Inform all task worker to stop after current task is complete')
             ->setHelp('This command allows you to safely stop all task workers');
     }

src/Command/Symfony/Work.php

@@ -40,14 +40,16 @@ public function __construct(CacheInterface $cache)
     protected function configure()
     {
         $this
-            ->setName('simple:task-worker:work')
+            ->setName('task-worker:work')
             ->setDescription('Patiently wait for a task(s) to perform')
             ->setHelp('This command allows you to start a task worker')
-            ->addOption('env', 'e', InputOption::VALUE_REQUIRED, 'Dot env file directory', realpath(__DIR__ . '/../../../examples/'))
+            ->addOption('dotenv', 'e', InputOption::VALUE_REQUIRED, 'Load configuration from environment file', '.env')
             ->addOption(Worker::OPT_SLEEP, 's', InputOption::VALUE_REQUIRED, 'How long, in seconds, to sleep if not tasks are available', Worker::defaultOptions()[Worker::OPT_SLEEP])
             ->addOption(Worker::OPT_ATTEMPTS, 'a', InputOption::VALUE_REQUIRED, 'How long many attempts a task is allowed before being failed (zero is unlimited)', Worker::defaultOptions()[Worker::OPT_ATTEMPTS])
             ->addOption(Worker::OPT_ALIVE, 'l', InputOption::VALUE_REQUIRED, 'How long, in seconds, the worker will stay alive for (zero is unlimited)', Worker::defaultOptions()[Worker::OPT_ALIVE])
-            ->addOption(Worker::OPT_REST, 'r', InputOption::VALUE_REQUIRED, 'How long, in milliseconds, to rest between tasks', Worker::defaultOptions()[Worker::OPT_REST]);
+            ->addOption(Worker::OPT_REST, 'r', InputOption::VALUE_REQUIRED, 'How long, in milliseconds, to rest between tasks', Worker::defaultOptions()[Worker::OPT_REST])
+            ->addOption(Worker::OPT_MAX_TASKS, 'm', InputOption::VALUE_REQUIRED, 'How many tasks should be performed (zero is unlimited)', Worker::defaultOptions()[Worker::OPT_MAX_TASKS])
+            ->addOption(Worker::OPT_UNTIL_EMPTY, 'u', InputOption::VALUE_NONE, 'If the worker should stop when the queue is empty');
     }
 
     /**
@@ -68,7 +70,7 @@ protected function configure()
     protected function execute(InputInterface $input, OutputInterface $output)
     {
 
-        $dotenv = new Dotenv($input->getOption('env'));
+        $dotenv = new Dotenv(getcwd(), $input->getOption('dotenv'));
         $dotenv->load();
 
         $logger = new ConsoleLogger($output);

src/Worker.php

@@ -30,18 +30,21 @@ class Worker
     const OPT_REST = 'rest';
     /** Option to set the maximum number of tasks a worker should perform before stopping (zero is unlimited). */
     const OPT_MAX_TASKS = 'max-tasks';
+    /** Option to set the worker to stop when the queue it empty (boolean 1 or 0). */
+    const OPT_UNTIL_EMPTY = 'until-empty';
 
     /**
      * @return array
      */
     public static function defaultOptions() : array
     {
         return [
-            self::OPT_SLEEP     => 3,
-            self::OPT_ATTEMPTS  => 0,
-            self::OPT_ALIVE     => 0,
-            self::OPT_REST      => 50,
-            self::OPT_MAX_TASKS => 0,
+            self::OPT_SLEEP       => 3,
+            self::OPT_ATTEMPTS    => 0,
+            self::OPT_ALIVE       => 0,
+            self::OPT_REST        => 50,
+            self::OPT_MAX_TASKS   => 0,
+            self::OPT_UNTIL_EMPTY => 0,
         ];
     }
 
@@ -156,12 +159,19 @@ public function run(): int
                     if ($this->opt(self::OPT_REST) > 0) {
                         usleep(min(1,$this->opt(self::OPT_REST)));
                     }
-                } else {
-                    // Patiently wait for next task to arrive
+                }
+
+                // Should we continue working?
+                if (!$this->shouldContinueWorking($task)) {
+                    break;
+                }
+
+                // Patiently wait for next task to arrive
+                if ($task === null) {
                     sleep(min(1, $this->opt(self::OPT_SLEEP)));
                 }
             }
-            while ($this->shouldContinueWorking());
+            while (true);
         }
         catch (\Throwable $throwable) {
             $this->logger->critical($throwable->getMessage(), ['trace' => $throwable->getTrace()]);
@@ -193,17 +203,20 @@ protected function prepare(Task $task) : Task
     }
 
     /**
+     * @param Task $task
      * @return bool
      */
-    protected function shouldContinueWorking(): bool
+    protected function shouldContinueWorking(Task $task = null): bool
     {
-        $optMaxTasks = $this->opt(self::OPT_MAX_TASKS);
-        $optAlive    = $this->opt(self::OPT_ALIVE);
-        $restartTime = $this->cache->get(sha1(self::CACHE_RESTART), 0);
+        $optMaxTasks   = $this->opt(self::OPT_MAX_TASKS);
+        $optAlive      = $this->opt(self::OPT_ALIVE);
+        $restartTime   = $this->cache->get(sha1(self::CACHE_RESTART), 0);
+        $optUntilEmpty = $this->opt(self::OPT_UNTIL_EMPTY);
 
         return
             ($optMaxTasks <= 0 || $optMaxTasks > $this->taskCount) &&
             ($optAlive <= 0 || $optAlive > round(time() - $this->startTime)) &&
-            ($restartTime === 0 || $restartTime <= $this->startTime);
+            ($restartTime === 0 || $restartTime <= $this->startTime) &&
+            (!$optUntilEmpty || $task !== null);
     }
 }