Skip to content

feat: add workerStop handling #45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
basert merged 1 commit into from
May 30, 2025
Merged

feat: add workerStop handling #45

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions src/Queue/Adapter/Swoole.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

namespace Utopia\Queue\Adapter;

use Swoole\Constant;
use Swoole\Process\Pool;
use Utopia\Queue\Adapter;
use Utopia\Queue\Consumer;
Expand All @@ -10,6 +11,9 @@ class Swoole extends Adapter
{
protected Pool $pool;

/** @var callable */
private $onStop;

public function __construct(Consumer $consumer, int $workerNum, string $queue, string $namespace = 'utopia-queue')
{
parent::__construct($workerNum, $queue, $namespace);
Expand All @@ -27,13 +31,16 @@ public function start(): self

public function stop(): self
{
if ($this->onStop) {
call_user_func($this->onStop);
}
$this->pool->shutdown();
return $this;
}

public function workerStart(callable $callback): self
{
$this->pool->on('WorkerStart', function (Pool $pool, string $workerId) use ($callback) {
$this->pool->on(Constant::EVENT_WORKER_START, function (Pool $pool, string $workerId) use ($callback) {
call_user_func($callback, $workerId);
});

Expand All @@ -42,7 +49,8 @@ public function workerStart(callable $callback): self

public function workerStop(callable $callback): self
{
$this->pool->on('WorkerStart', function (Pool $pool, string $workerId) use ($callback) {
$this->onStop = $callback;
$this->pool->on(Constant::EVENT_WORKER_STOP, function (Pool $pool, string $workerId) use ($callback) {
call_user_func($callback, $workerId);
});

Expand Down
1 change: 1 addition & 0 deletions src/Queue/Broker/AMQP.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,7 @@ public function consume(Queue $queue, callable $messageCallback, callable $succe

public function close(): void
{
$this->channel?->stopConsume();
$this->channel?->getConnection()?->close();
}

Expand Down
125 changes: 50 additions & 75 deletions src/Queue/Server.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,45 +217,19 @@ public function start(): self
call_user_func_array($this->workerStartHook->getAction(), $this->getArguments($this->workerStartHook));
}

while (true) {
$this->adapter->consumer->consume(
$this->adapter->queue,
function (Message $message) {
$receivedAtTimestamp = microtime(true);
Console::info("[Job] Received Job ({$message->getPid()}).");
try {
$waitDuration = microtime(true) - $message->getTimestamp();
$this->jobWaitTime->record($waitDuration);

$this->resources = [];
self::setResource('message', fn () => $message);
if ($this->job->getHook()) {
foreach ($this->initHooks as $hook) { // Global init hooks
if (in_array('*', $hook->getGroups())) {
$arguments = $this->getArguments($hook, $message->getPayload());
\call_user_func_array($hook->getAction(), $arguments);
}
}
}

foreach ($this->job->getGroups() as $group) {
foreach ($this->initHooks as $hook) { // Group init hooks
if (in_array($group, $hook->getGroups())) {
$arguments = $this->getArguments($hook, $message->getPayload());
\call_user_func_array($hook->getAction(), $arguments);
}
}
}

return \call_user_func_array($this->job->getAction(), $this->getArguments($this->job, $message->getPayload()));
} finally {
$processDuration = microtime(true) - $receivedAtTimestamp;
$this->processDuration->record($processDuration);
}
},
function (Message $message) {
$this->adapter->consumer->consume(
$this->adapter->queue,
function (Message $message) {
$receivedAtTimestamp = microtime(true);
Console::info("[Job] Received Job ({$message->getPid()}).");
try {
$waitDuration = microtime(true) - $message->getTimestamp();
$this->jobWaitTime->record($waitDuration);

$this->resources = [];
self::setResource('message', fn () => $message);
if ($this->job->getHook()) {
foreach ($this->shutdownHooks as $hook) { // Global init hooks
foreach ($this->initHooks as $hook) { // Global init hooks
if (in_array('*', $hook->getGroups())) {
$arguments = $this->getArguments($hook, $message->getPayload());
\call_user_func_array($hook->getAction(), $arguments);
Expand All @@ -264,29 +238,55 @@ function (Message $message) {
}

foreach ($this->job->getGroups() as $group) {
foreach ($this->shutdownHooks as $hook) { // Group init hooks
foreach ($this->initHooks as $hook) { // Group init hooks
if (in_array($group, $hook->getGroups())) {
$arguments = $this->getArguments($hook, $message->getPayload());
\call_user_func_array($hook->getAction(), $arguments);
}
}
}
Console::success("[Job] ({$message->getPid()}) successfully run.");
},
function (?Message $message, Throwable $th) {
Console::error("[Job] ({$message?->getPid()}) failed to run.");
Console::error("[Job] ({$message?->getPid()}) {$th->getMessage()}");

self::setResource('error', fn () => $th);
return \call_user_func_array($this->job->getAction(), $this->getArguments($this->job, $message->getPayload()));
} finally {
$processDuration = microtime(true) - $receivedAtTimestamp;
$this->processDuration->record($processDuration);
}
},
function (Message $message) {
if ($this->job->getHook()) {
foreach ($this->shutdownHooks as $hook) { // Global init hooks
if (in_array('*', $hook->getGroups())) {
$arguments = $this->getArguments($hook, $message->getPayload());
\call_user_func_array($hook->getAction(), $arguments);
}
}
}

foreach ($this->errorHooks as $hook) {
($hook->getAction())(...$this->getArguments($hook));
foreach ($this->job->getGroups() as $group) {
foreach ($this->shutdownHooks as $hook) { // Group init hooks
if (in_array($group, $hook->getGroups())) {
$arguments = $this->getArguments($hook, $message->getPayload());
\call_user_func_array($hook->getAction(), $arguments);
}
}
},
);
}
}
Console::success("[Job] ({$message->getPid()}) successfully run.");
},
function (?Message $message, Throwable $th) {
Console::error("[Job] ({$message?->getPid()}) failed to run.");
Console::error("[Job] ({$message?->getPid()}) {$th->getMessage()}");

self::setResource('error', fn () => $th);

foreach ($this->errorHooks as $hook) {
($hook->getAction())(...$this->getArguments($hook));
}
},
);
});

$this->adapter->workerStop(fn () => $this->adapter->consumer->close());

$this->adapter->start();
} catch (Throwable $error) {
self::setResource('error', fn () => $error);
Expand Down Expand Up @@ -318,31 +318,6 @@ public function getWorkerStart(): Hook
return $this->workerStartHook;
}

/**
* Is called when a Worker stops.
* @param callable|null $callback
* @return self
* @throws Exception
*/
public function workerStop(?callable $callback = null): self
{
try {
$this->adapter->workerStop(function (string $workerId) use ($callback) {
Console::success("[Worker] Worker {$workerId} is ready!");
if (!is_null($callback)) {
call_user_func($callback);
}
});
} catch (Throwable $error) {
self::setResource('error', fn () => $error);
foreach ($this->errorHooks as $hook) {
call_user_func_array($hook->getAction(), $this->getArguments($hook));
}
}

return $this;
}

/**
* Get Arguments
*
Expand Down