Skip to content

Revert "feat: add workerStop handling" #49

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
merged 1 commit into from
Jun 24, 2025
Merged

Revert "feat: add workerStop handling" #49

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: 2 additions & 10 deletions src/Queue/Adapter/Swoole.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

namespace Utopia\Queue\Adapter;

use Swoole\Constant;
use Swoole\Process\Pool;
use Utopia\Queue\Adapter;
use Utopia\Queue\Consumer;
Expand All @@ -11,9 +10,6 @@ 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 @@ -31,16 +27,13 @@ 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(Constant::EVENT_WORKER_START, function (Pool $pool, string $workerId) use ($callback) {
$this->pool->on('WorkerStart', function (Pool $pool, string $workerId) use ($callback) {
call_user_func($callback, $workerId);
});

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

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

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

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

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

$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);
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) {
if ($this->job->getHook()) {
foreach ($this->initHooks as $hook) { // Global init hooks
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);
Expand All @@ -238,55 +264,29 @@ function (Message $message) {
}

foreach ($this->job->getGroups() as $group) {
foreach ($this->initHooks as $hook) { // Group init hooks
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()}");

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);
}
}
}
self::setResource('error', fn () => $th);

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);
}
foreach ($this->errorHooks as $hook) {
($hook->getAction())(...$this->getArguments($hook));
}
}
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,6 +318,31 @@ 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;
}
Comment on lines +327 to +344
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Incorrect log message in workerStop() method

The log message indicates the worker is "ready" when it should indicate the worker is stopping.

 public function workerStop(?callable $callback = null): self
 {
     try {
         $this->adapter->workerStop(function (string $workerId) use ($callback) {
-            Console::success("[Worker] Worker {$workerId} is ready!");
+            Console::success("[Worker] Worker {$workerId} is stopping!");
             if (!is_null($callback)) {
                 call_user_func($callback);
             }
         });
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
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;
}
public function workerStop(?callable $callback = null): self
{
try {
$this->adapter->workerStop(function (string $workerId) use ($callback) {
Console::success("[Worker] Worker {$workerId} is stopping!");
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;
}
🤖 Prompt for AI Agents 
In src/Queue/Server.php around lines 327 to 344, the log message inside the
workerStop() method incorrectly states the worker is "ready" when it should
indicate the worker is stopping. Update the Console::success log message to
reflect that the worker with the given ID is stopping, not ready.


/**
* Get Arguments
*
Expand Down