Skip the sleep when the job returns true

docs/getting-started.md

@@ -81,6 +81,33 @@ The job's own run time is subtracted from the sleep: if the job took 300ms and t
 is 500ms, the worker only sleeps 200ms. If the job took longer than the sleep timer,
 the next iteration starts immediately. Pass `0` to disable sleeping entirely.
 
+### Skip the sleep when there is work
+
+If the job returns `true`, the worker skips the sleep and starts the next iteration immediately.
+This is useful for queue consumers: as long as there are messages, they are processed without a pause,
+and the worker only sleeps once the queue is empty.
+
+```php
+use Patchlevel\Worker\DefaultWorker;
+
+$worker = DefaultWorker::create(
+    static function (callable $stop) use ($queue): bool {
+        $message = $queue->pop();
+
+        if ($message === null) {
+            return false; // nothing to do, sleep
+        }
+
+        handle($message);
+
+        return true; // there may be more, continue immediately
+    },
+);
+
+$worker->run(1000);
+```
+Any other return value, including no return value at all, keeps the regular sleep behaviour.
+
 ## Logging
 
 The worker logs its lifecycle (start, iteration timings, sleep, stop reason) to the given PSR-3 logger.

src/DefaultWorker.php

@@ -28,7 +28,7 @@
 
     private readonly ClockInterface $clock;
 
-    /** @param Closure(Closure):void $job */
+    /** @param Closure(Closure):(bool|void) $job return true if the job did work to skip the sleep */
     public function __construct(
         private readonly Closure $job,
         private readonly EventDispatcherInterface $eventDispatcher,
@@ -39,7 +39,7 @@
     }
 
     /** @param positive-int|0 $sleepTimer in milliseconds */
     public function run(int $sleepTimer = 1000): void
     {
         $this->logger?->debug('Worker starting');
 
@@ -53,7 +53,7 @@
 
                 $startTime = $this->milliseconds();
 
-                ($this->job)($this->stop(...));
+                $didWork = ($this->job)($this->stop(...)) === true;
 
                 $endTime = $this->milliseconds();
                 $ranTime = $endTime - $startTime;
@@ -63,17 +63,21 @@
                 $this->eventDispatcher->dispatch(new WorkerRunningEvent($this));
 
                 if ($this->shouldStop) {
                     break;
                 }
 
+                if ($didWork) {
+                    continue;
+                }
+
                 $sleepFor = max($sleepTimer - $ranTime, 0);
 
                 if ($sleepFor <= 0) {
                     continue;
                 }
 
                 $this->logger?->debug('Worker sleep for {sleepTimer}ms', ['sleepTimer' => $sleepFor]);
                 usleep($sleepFor * 1000);
             }
         } catch (Throwable $exception) {
             throw $exception;
@@ -95,7 +99,7 @@
     }
 
     /**
-     * @param Closure(Closure):void                                                                               $job
+     * @param Closure(Closure):(bool|void)                                                                        $job
      * @param array{runLimit?: (positive-int|null), memoryLimit?: (string|null), timeLimit?: (positive-int|null)} $options
      */
     public static function create(

tests/Unit/DefaultWorkerTest.php

@@ -22,6 +22,8 @@
 use Symfony\Component\EventDispatcher\EventDispatcher;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
+use function array_shift;
+
 #[CoversClass(DefaultWorker::class)]
 final class DefaultWorkerTest extends TestCase
 {
@@ -243,6 +245,45 @@ public function testRunWorkerNotSleeping(): void
         $worker->run(5);
     }
 
+    public function testRunWorkerSkipsSleepWhenJobDidWork(): void
+    {
+        $logger = $this->createMock(LoggerInterface::class);
+        $logger
+            ->expects($this->exactly(11))
+            ->method('debug')
+            ->willReturnCallback(
+                new ReturnCallback([
+                    [['Worker starting', []]],
+                    [['Worker starting job run', []]],
+                    [['Worker finished job run ({ranTime}ms)', ['ranTime' => 10]]],
+                    [['Worker starting job run', []]],
+                    [['Worker finished job run ({ranTime}ms)', ['ranTime' => 10]]],
+                    [['Worker sleep for {sleepTimer}ms', ['sleepTimer' => 190]]],
+                    [['Worker starting job run', []]],
+                    [['Worker finished job run ({ranTime}ms)', ['ranTime' => 10]]],
+                    [['Worker received stop signal', []]],
+                    [['Worker stopped', []]],
+                    [['Worker terminated', []]],
+                ]),
+            );
+
+        $eventDispatcher = new EventDispatcher();
+        $eventDispatcher->addSubscriber(new StopWorkerOnIterationLimitListener(3));
+
+        $jobResults = [true, false, true];
+
+        $worker = new DefaultWorker(
+            static function () use (&$jobResults): bool {
+                return (bool)array_shift($jobResults);
+            },
+            $eventDispatcher,
+            $logger,
+            new TestClock(tick: 10),
+        );
+
+        $worker->run(200);
+    }
+
     public function testDefaultCreate(): void
     {
         $calls = 0;