Merge pull request #12 from MammatusPHP/cancel-all-promises-when-shutting-down

Cancel all promises when shutting down

Author: WyriHaximus

composer.lock

@@ -11,6 +11,7 @@
 use Mammatus\Queue\App\Queue;
 use Mammatus\Queue\Generated\AbstractList;
 use Psr\Log\LoggerInterface;
+use React\Promise\PromiseInterface;
 use Throwable;
 use WyriHaximus\Broadcast\Contracts\Listener;
 
@@ -21,6 +22,9 @@
 /** @implements Bootable<Queue> */
 final class App extends AbstractList implements Bootable, Listener
 {
+    /** @var array<PromiseInterface<mixed>> */
+    private array $promises = [];
+
     public function __construct(
         private readonly Consumer $consumer,
         private readonly LoggerInterface $logger,
@@ -29,22 +33,25 @@ public function __construct(
 
     public function stop(Shutdown $event): void
     {
+        foreach ($this->promises as $promise) {
+            $promise->cancel();
+        }
+
         $this->consumer->close();
     }
 
     public function boot(Argv $argv): ExitCode
     {
         try {
-            $promises = [];
             foreach ($this->workers() as $worker) {
                 if ($worker->class !== $argv->className) {
                     continue;
                 }
 
-                $promises[] = $this->consumer->setupConsumer($worker);
+                $this->promises[] = $this->consumer->setupConsumer($worker);
             }
 
-            await(all($promises));
+            await(all($this->promises));
 
             $exitCode = ExitCode::Success;
         } catch (Throwable $throwable) { /** @phpstan-ignore-line */

src/App.php

@@ -4,7 +4,7 @@
 
 namespace Mammatus\Queue;
 
-use Mammatus\LifeCycleEvents\Initialize;
+use Mammatus\LifeCycleEvents\Boot;
 use Mammatus\LifeCycleEvents\Shutdown;
 use Mammatus\Queue\Generated\AbstractList;
 use Psr\EventDispatcher\EventDispatcherInterface;
@@ -21,7 +21,7 @@ public function __construct(
     ) {
     }
 
-    public function start(Initialize $event): void
+    public function start(Boot $event): void
     {
         $this->logger->debug('Starting queue manager');
         $this->boot();

src/Manager.php

@@ -4,7 +4,7 @@
 
 namespace Mammatus\Tests\Queue;
 
-use Mammatus\LifeCycleEvents\Initialize;
+use Mammatus\LifeCycleEvents\Boot;
 use Mammatus\LifeCycleEvents\Shutdown;
 use Mammatus\Queue\BuildIn\Noop;
 use Mammatus\Queue\Contracts\Worker as WorkerContract;
@@ -47,7 +47,7 @@ public function runHappy(): void
             $eventDispatcher,
             $logger,
         );
-        $manager->start(new Initialize());
+        $manager->start(new Boot());
         await(sleep(9));
         $manager->stop(new Shutdown());
     }
@@ -76,7 +76,7 @@ public function runAngry(): void
             $eventDispatcher,
             $logger,
         );
-        $manager->start(new Initialize());
+        $manager->start(new Boot());
         await(sleep(9));
         $manager->stop(new Shutdown());
     }
@@ -111,6 +111,6 @@ public function notAnWorker(): void
             $eventDispatcher,
             $logger,
         );
-        $manager->start(new Initialize());
+        $manager->start(new Boot());
     }
 }