Add Swoole coroutine concurrency

README.md

@@ -1,6 +1,5 @@
 # Utopia Queue
 
-[![Build Status](https://travis-ci.com/utopia-php/queue.svg?branch=main)](https://travis-ci.com/utopia-php/queue)
 ![Total Downloads](https://img.shields.io/packagist/dt/utopia-php/queue.svg)
 [![Discord](https://img.shields.io/discord/564160730845151244?label=discord)](https://appwrite.io/discord)
 

composer.json

@@ -21,7 +21,7 @@
     },
     "scripts":{
         "test": "phpunit",
-        "check": "vendor/bin/phpstan analyse",
+        "check": "vendor/bin/phpstan analyse --memory-limit=1G",
         "format": "vendor/bin/pint",
         "lint": "vendor/bin/pint --test"
     },

phpstan.neon

@@ -6,4 +6,4 @@ parameters:
         - tests
 
     scanDirectories:
-        - vendor/swoole
+        - vendor/swoole

src/Queue/Adapter.php

@@ -2,16 +2,20 @@
 
 namespace Utopia\Queue;
 
+use Utopia\DI\Container;
+
 abstract class Adapter
 {
-    public int $workerNum;
     public Queue $queue;
-    public string $namespace;
-    public Consumer $consumer;
+    protected ?Container $context = null;
 
-    public function __construct(int $workerNum, string $queue, string $namespace = 'utopia-queue')
-    {
-        $this->workerNum = $workerNum;
+    public function __construct(
+        public Consumer $consumer,
+        public int $workerNum,
+        string $queue,
+        public string $namespace = 'utopia-queue',
+        protected Container $resources = new Container(),
+    ) {
         $this->queue = new Queue($queue, $namespace);
     }
 
@@ -27,6 +31,36 @@ abstract public function start(): self;
      */
     abstract public function stop(): self;
 
+    public function consume(callable $messageCallback, callable $successCallback, callable $errorCallback): void
+    {
+        $this->consumer->consume(
+            $this->queue,
+            function (Message $message) use ($messageCallback) {
+                $this->context = new Container($this->resources());
+
+                return $messageCallback($message);
+            },
+            $successCallback,
+            function (?Message $message, \Throwable $error) use ($errorCallback) {
+                if ($message === null) {
+                    $this->context = new Container($this->resources());
+                }
+
+                $errorCallback($message, $error);
+            },
+        );
+    }
+
+    public function resources(): Container
+    {
+        return $this->resources;
+    }
+
+    public function context(): Container
+    {
+        return $this->context ??= new Container($this->resources());
+    }
+
     /**
      * Is called when a Worker starts.
      * @param callable $callback

src/Queue/Adapter/Swoole.php

@@ -3,12 +3,18 @@
 namespace Utopia\Queue\Adapter;
 
 use Swoole\Coroutine;
+use Swoole\Coroutine\Channel;
 use Swoole\Process;
+use Utopia\DI\Container;
 use Utopia\Queue\Adapter;
 use Utopia\Queue\Consumer;
+use Utopia\Queue\Error\ConsumerFailures;
+use Utopia\Queue\Message;
 
 class Swoole extends Adapter
 {
+    protected const string CONTEXT_KEY = '__utopia__';
+
     /** @var Process[] */
     protected array $workers = [];
 
@@ -23,9 +29,11 @@ public function __construct(
         int $workerNum,
         string $queue,
         string $namespace = 'utopia-queue',
+        protected int $maxCoroutines = 1,
+        Container $resources = new Container(),
     ) {
-        parent::__construct($workerNum, $queue, $namespace);
-        $this->consumer = $consumer;
+        parent::__construct($consumer, $workerNum, $queue, $namespace, $resources);
+        $this->maxCoroutines = \max(1, $maxCoroutines);
     }
 
     public function start(): self
@@ -71,6 +79,65 @@ protected function spawnWorker(int $workerId): void
         $this->workers[$pid] = $process;
     }
 
+    public function consume(callable $messageCallback, callable $successCallback, callable $errorCallback): void
+    {
+        $messageCallback = function (Message $message) use ($messageCallback) {
+            Coroutine::getContext()[self::CONTEXT_KEY] = new Container($this->resources());
+
+            return $messageCallback($message);
+        };
+
+        $errorCallback = function (?Message $message, \Throwable $error) use ($errorCallback) {
+            if ($message === null) {
+                Coroutine::getContext()[self::CONTEXT_KEY] = new Container($this->resources());
+            }
+
+            $errorCallback($message, $error);
+        };
+
+        $channel = new Channel($this->maxCoroutines);
+        $errors = [];
+
+        for ($i = 0; $i < $this->maxCoroutines; $i++) {
+            Coroutine::create(function () use ($messageCallback, $successCallback, $errorCallback, $channel, &$errors) {
+                try {
+                    $this->consumer->consume(
+                        $this->queue,
+                        $messageCallback,
+                        $successCallback,
+                        $errorCallback,
+                    );
+                } catch (\Throwable $error) {
+                    $errors[] = $error;
+                    $this->consumer->close();
+                    $channel->push(true);
+                    return;
+                }
+
+                $channel->push(true);
+            });
+        }
+
+        for ($i = 0; $i < $this->maxCoroutines; $i++) {
+            $channel->pop();
+        }
+
+        $channel->close();
+
+        if ($errors !== []) {
+            throw new ConsumerFailures($errors);
+        }
+    }
+
+    public function context(): Container
+    {
+        if (Coroutine::getCid() !== -1) {
+            return Coroutine::getContext()[self::CONTEXT_KEY] ?? $this->resources();
+        }
+
+        return $this->resources();
+    }
+
     protected function reap(): void
     {
         while (($ret = Process::wait(false)) !== false) {

src/Queue/Adapter/Workerman.php

@@ -2,6 +2,7 @@
 
 namespace Utopia\Queue\Adapter;
 
+use Utopia\DI\Container;
 use Utopia\Queue\Adapter;
 use Utopia\Queue\Consumer;
 use Workerman\Worker;
@@ -10,13 +11,17 @@ class Workerman extends Adapter
 {
     protected Worker $worker;
 
-    public function __construct(Consumer $consumer, int $workerNum, string $queue, string $namespace = 'utopia-queue')
-    {
-        parent::__construct($workerNum, $queue, $namespace);
+    public function __construct(
+        Consumer $consumer,
+        int $workerNum,
+        string $queue,
+        string $namespace = 'utopia-queue',
+        Container $resources = new Container(),
+    ) {
+        parent::__construct($consumer, $workerNum, $queue, $namespace, $resources);
 
         $this->worker = new Worker();
         $this->worker->count = $workerNum;
-        $this->consumer = $consumer;
     }
 
     public function start(): self

src/Queue/Error/ConsumerFailures.php

@@ -0,0 +1,25 @@
+<?php
+
+namespace Utopia\Queue\Error;
+
+class ConsumerFailures extends \RuntimeException
+{
+    /**
+     * @param \Throwable[] $errors
+     */
+    public function __construct(private array $errors)
+    {
+        parent::__construct(
+            'Queue consumers failed with ' . \count($errors) . ' error(s).',
+            previous: $errors[0] ?? null,
+        );
+    }
+
+    /**
+     * @return \Throwable[]
+     */
+    public function getErrors(): array
+    {
+        return $this->errors;
+    }
+}