Skip to content

Commit cd4248a

Browse files
loks0nloks0n
authored
Merge pull request #84 from utopia-php/swoole-concurrency
Add Swoole coroutine concurrency
2 parents 954cd37 + 88a5431 commit cd4248a

12 files changed

Lines changed: 393 additions & 97 deletions

File tree

‎.travis.yml‎

Lines changed: 0 additions & 17 deletions
This file was deleted.

‎README.md‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
# Utopia Queue
22

3-
[![Build Status](https://travis-ci.com/utopia-php/queue.svg?branch=main)](https://travis-ci.com/utopia-php/queue)
43
![Total Downloads](https://img.shields.io/packagist/dt/utopia-php/queue.svg)
54
[![Discord](https://img.shields.io/discord/564160730845151244?label=discord)](https://appwrite.io/discord)
65

‎composer.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
},
2222
"scripts":{
2323
"test": "phpunit",
24-
"check": "vendor/bin/phpstan analyse",
24+
"check": "vendor/bin/phpstan analyse --memory-limit=1G",
2525
"format": "vendor/bin/pint",
2626
"lint": "vendor/bin/pint --test"
2727
},

‎phpstan.neon‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,4 @@ parameters:
66
- tests
77

88
scanDirectories:
9-
- vendor/swoole
9+
- vendor/swoole

‎src/Queue/Adapter.php‎

Lines changed: 40 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,20 @@
22

33
namespace Utopia\Queue;
44

5+
use Utopia\DI\Container;
6+
57
abstract class Adapter
68
{
7-
public int $workerNum;
89
public Queue $queue;
9-
public string $namespace;
10-
public Consumer $consumer;
10+
protected ?Container $context = null;
1111

12-
public function __construct(int $workerNum, string $queue, string $namespace = 'utopia-queue')
13-
{
14-
$this->workerNum = $workerNum;
12+
public function __construct(
13+
public Consumer $consumer,
14+
public int $workerNum,
15+
string $queue,
16+
public string $namespace = 'utopia-queue',
17+
protected Container $resources = new Container(),
18+
) {
1519
$this->queue = new Queue($queue, $namespace);
1620
}
1721

@@ -27,6 +31,36 @@ abstract public function start(): self;
2731
*/
2832
abstract public function stop(): self;
2933

34+
public function consume(callable $messageCallback, callable $successCallback, callable $errorCallback): void
35+
{
36+
$this->consumer->consume(
37+
$this->queue,
38+
function (Message $message) use ($messageCallback) {
39+
$this->context = new Container($this->resources());
40+
41+
return $messageCallback($message);
42+
},
43+
$successCallback,
44+
function (?Message $message, \Throwable $error) use ($errorCallback) {
45+
if ($message === null) {
46+
$this->context = new Container($this->resources());
47+
}
48+
49+
$errorCallback($message, $error);
50+
},
51+
);
52+
}
53+
54+
public function resources(): Container
55+
{
56+
return $this->resources;
57+
}
58+
59+
public function context(): Container
60+
{
61+
return $this->context ??= new Container($this->resources());
62+
}
63+
3064
/**
3165
* Is called when a Worker starts.
3266
* @param callable $callback

‎src/Queue/Adapter/Swoole.php‎

Lines changed: 69 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,18 @@
33
namespace Utopia\Queue\Adapter;
44

55
use Swoole\Coroutine;
6+
use Swoole\Coroutine\Channel;
67
use Swoole\Process;
8+
use Utopia\DI\Container;
79
use Utopia\Queue\Adapter;
810
use Utopia\Queue\Consumer;
11+
use Utopia\Queue\Error\ConsumerFailures;
12+
use Utopia\Queue\Message;
913

1014
class Swoole extends Adapter
1115
{
16+
protected const string CONTEXT_KEY = '__utopia__';
17+
1218
/** @var Process[] */
1319
protected array $workers = [];
1420

@@ -23,9 +29,11 @@ public function __construct(
2329
int $workerNum,
2430
string $queue,
2531
string $namespace = 'utopia-queue',
32+
protected int $maxCoroutines = 1,
33+
Container $resources = new Container(),
2634
) {
27-
parent::__construct($workerNum, $queue, $namespace);
28-
$this->consumer = $consumer;
35+
parent::__construct($consumer, $workerNum, $queue, $namespace, $resources);
36+
$this->maxCoroutines = \max(1, $maxCoroutines);
2937
}
3038

3139
public function start(): self
@@ -71,6 +79,65 @@ protected function spawnWorker(int $workerId): void
7179
$this->workers[$pid] = $process;
7280
}
7381

82+
public function consume(callable $messageCallback, callable $successCallback, callable $errorCallback): void
83+
{
84+
$messageCallback = function (Message $message) use ($messageCallback) {
85+
Coroutine::getContext()[self::CONTEXT_KEY] = new Container($this->resources());
86+
87+
return $messageCallback($message);
88+
};
89+
90+
$errorCallback = function (?Message $message, \Throwable $error) use ($errorCallback) {
91+
if ($message === null) {
92+
Coroutine::getContext()[self::CONTEXT_KEY] = new Container($this->resources());
93+
}
94+
95+
$errorCallback($message, $error);
96+
};
97+
98+
$channel = new Channel($this->maxCoroutines);
99+
$errors = [];
100+
101+
for ($i = 0; $i < $this->maxCoroutines; $i++) {
102+
Coroutine::create(function () use ($messageCallback, $successCallback, $errorCallback, $channel, &$errors) {
103+
try {
104+
$this->consumer->consume(
105+
$this->queue,
106+
$messageCallback,
107+
$successCallback,
108+
$errorCallback,
109+
);
110+
} catch (\Throwable $error) {
111+
$errors[] = $error;
112+
$this->consumer->close();
113+
$channel->push(true);
114+
return;
115+
}
116+
117+
$channel->push(true);
118+
});
119+
}
120+
121+
for ($i = 0; $i < $this->maxCoroutines; $i++) {
122+
$channel->pop();
123+
}
124+
125+
$channel->close();
126+
127+
if ($errors !== []) {
128+
throw new ConsumerFailures($errors);
129+
}
130+
}
131+
132+
public function context(): Container
133+
{
134+
if (Coroutine::getCid() !== -1) {
135+
return Coroutine::getContext()[self::CONTEXT_KEY] ?? $this->resources();
136+
}
137+
138+
return $this->resources();
139+
}
140+
74141
protected function reap(): void
75142
{
76143
while (($ret = Process::wait(false)) !== false) {

‎src/Queue/Adapter/Workerman.php‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
namespace Utopia\Queue\Adapter;
44

5+
use Utopia\DI\Container;
56
use Utopia\Queue\Adapter;
67
use Utopia\Queue\Consumer;
78
use Workerman\Worker;
@@ -10,13 +11,17 @@ class Workerman extends Adapter
1011
{
1112
protected Worker $worker;
1213

13-
public function __construct(Consumer $consumer, int $workerNum, string $queue, string $namespace = 'utopia-queue')
14-
{
15-
parent::__construct($workerNum, $queue, $namespace);
14+
public function __construct(
15+
Consumer $consumer,
16+
int $workerNum,
17+
string $queue,
18+
string $namespace = 'utopia-queue',
19+
Container $resources = new Container(),
20+
) {
21+
parent::__construct($consumer, $workerNum, $queue, $namespace, $resources);
1622

1723
$this->worker = new Worker();
1824
$this->worker->count = $workerNum;
19-
$this->consumer = $consumer;
2025
}
2126

2227
public function start(): self

‎src/Queue/Error/ConsumerFailures.php‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
<?php
2+
3+
namespace Utopia\Queue\Error;
4+
5+
class ConsumerFailures extends \RuntimeException
6+
{
7+
/**
8+
* @param \Throwable[] $errors
9+
*/
10+
public function __construct(private array $errors)
11+
{
12+
parent::__construct(
13+
'Queue consumers failed with ' . \count($errors) . ' error(s).',
14+
previous: $errors[0] ?? null,
15+
);
16+
}
17+
18+
/**
19+
* @return \Throwable[]
20+
*/
21+
public function getErrors(): array
22+
{
23+
return $this->errors;
24+
}
25+
}

0 commit comments

Comments
 (0)