Make app work

src/App.php

@@ -4,6 +4,9 @@
 
 namespace Mammatus\Queue;
 
+use Interop\Queue\Consumer;
+use Interop\Queue\Context;
+use Mammatus\LifeCycleEvents\Shutdown;
 use Mammatus\Queue\Contracts\Worker;
 use Mammatus\Queue\Contracts\Worker as WorkerContract;
 use Mammatus\Queue\Generated\AbstractList_;
@@ -12,21 +15,30 @@
 use React\EventLoop\Loop;
 use RuntimeException;
 use Throwable;
+use WyriHaximus\Broadcast\Contracts\Listener;
 use WyriHaximus\PSR3\ContextLogger\ContextLogger;
 
 use function React\Async\async;
 use function React\Async\await;
 use function React\Promise\all;
 use function WyriHaximus\React\futurePromise;
 
-final class App extends AbstractList_
+final class App extends AbstractList_ implements Listener
 {
+    private bool $running = true;
+
     public function __construct(
         private readonly ContainerInterface $container,
+        private readonly Context $context,
         private readonly LoggerInterface $logger,
     ) {
     }
 
+    public function stop(Shutdown $event): void
+    {
+        $this->running = false;
+    }
+
     public function run(string $className): int
     {
         $promises = [];
@@ -35,22 +47,33 @@
                 continue;
             }
 
-            for ($i = 0; $i < $worker->concurrency; $i++) {
-                $this->logger->info('Starting consumer ' . $i . ' of ' . $worker->concurrency . ' for ' . $worker->class);
-                $promises[] = async(fn () => $this->consume($worker))();
-            }
+            $promises[] = async(fn () => $this->setupConsumer($worker))();
         }
 
         await(all($promises));
 
         return 0;
     }
 
-    private function consume(\Mammatus\Queue\Worker $worker): void
+    private function setupConsumer(\Mammatus\Queue\Worker $worker): int
     {
         $consumer       = $this->context->createConsumer(new Queue($worker->queue));
         $workerInstance = $this->container->get($worker->class);
         assert($workerInstance instanceof WorkerContract);
+
+        $promises = [];
+        for ($i = 0; $i < $worker->concurrency; $i++) {
+            $this->logger->info('Starting consumer ' . $i . ' of ' . $worker->concurrency . ' for ' . $worker->class);
+            $promises[] = async(fn () => $this->consume($consumer, $workerInstance))();
+        }
+
+        await(all($promises));
+
+        return 0;
+    }
+
+    private function consume(Consumer $consumer, WorkerContract $worker): void
+    {
         while ($this->running) {
             $message = $consumer->receiveNoWait();
             if ($message === null) {
@@ -59,9 +82,9 @@
             }
 
             try {
-                $workerInstance->perform($message);
+                $worker->perform($message);
                 $consumer->acknowledge($message);
             } catch (Throwable) {
                 $consumer->reject($message);
             }
         }