[CORE][Queue] Fix some minor issues

This commit is contained in:
Diogo Peralta Cordeiro 2022-03-05 14:22:44 +00:00
parent 626b4263f1
commit 6fa5ec3218
Signed by: diogo
GPG Key ID: 18D2D35001FBFAB0
3 changed files with 12 additions and 11 deletions

View File

@ -27,15 +27,10 @@ namespace App\Core\Queue;
class Message class Message
{ {
public $content;
public string $queue;
/** /**
* @codeCoverageIgnore * @codeCoverageIgnore
*/ */
public function __construct($content, string $queue) public function __construct(public mixed $payload, public string $queue)
{ {
$this->queue = $queue;
$this->content = $content;
} }
} }

View File

@ -26,6 +26,7 @@ declare(strict_types = 1);
namespace App\Core\Queue; namespace App\Core\Queue;
use App\Core\Event; use App\Core\Event;
use App\Util\Formatting;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface; use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
class MessageHandler implements MessageHandlerInterface class MessageHandler implements MessageHandlerInterface
@ -35,6 +36,11 @@ class MessageHandler implements MessageHandlerInterface
*/ */
public function __invoke(Message $message) public function __invoke(Message $message)
{ {
Event::handle($message->queue, [$message->content]); $retry_args = [];
$args = \is_array($message->payload) ? $message->payload : [$message->payload];
Event::handle(
name: 'Queue' . Formatting::snakeCaseToCamelCase($message->queue),
args: [...$args, &$retry_args],
);
} }
} }

View File

@ -48,12 +48,12 @@ abstract class Queue
* *
* @codeCoverageIgnore * @codeCoverageIgnore
*/ */
public static function enqueue($message, string $queue, bool $high = false, array $stamps = []) public static function enqueue($payload, string $queue, bool $priority = false, array $stamps = [])
{ {
if ($high) { if ($priority) {
self::$message_bus->dispatch(new MessageHigh($message, $queue), $stamps); self::$message_bus->dispatch(new MessageHigh($payload, $queue), $stamps);
} else { } else {
self::$message_bus->dispatch(new MessageLow($message, $queue), $stamps); self::$message_bus->dispatch(new MessageLow($payload, $queue), $stamps);
} }
} }
} }