swoole server, queue consumer with redis, ready for testing and discussions

Author: pinclau

bin/messenger.php

@@ -22,6 +22,7 @@
     \Netglue\PsrContainer\Messenger\ConfigProvider::class,
 
     // Default App module config
+    \Queue\App\ConfigProvider::class,
     \Queue\Swoole\ConfigProvider::class,
 
     // Load application config in a pre-defined order in such a way that local settings

config/config.php

@@ -0,0 +1,95 @@
+<?php
+
+namespace Queue\App;
+
+use Laminas\ServiceManager\Factory\InvokableFactory;
+use Netglue\PsrContainer\Messenger\Container\MessageBusStaticFactory;
+use Netglue\PsrContainer\Messenger\Container\Middleware\BusNameStampMiddlewareStaticFactory;
+use Netglue\PsrContainer\Messenger\Container\Middleware\MessageHandlerMiddlewareStaticFactory;
+use Netglue\PsrContainer\Messenger\Container\Middleware\MessageSenderMiddlewareStaticFactory;
+use Netglue\PsrContainer\Messenger\HandlerLocator\OneToManyFqcnContainerHandlerLocator;
+use Queue\App\Message\ExampleMessage;
+use Queue\App\Message\ExampleMessageHandler;
+use Queue\App\Message\ExampleMessageHandlerFactory;
+use Symfony\Component\Messenger\MessageBusInterface;
+
+class ConfigProvider
+{
+    public function __invoke()
+    {
+        return [
+            "dependencies" => $this->getDependencies(),
+            'symfony' => [
+                'messenger' => [
+                    'buses' => $this->busConfig(),
+                ],
+            ],
+        ];
+    }
+
+
+    private function getDependencies()
+    {
+        return [
+            "factories" => [
+                "message_bus" => [MessageBusStaticFactory::class, "message_bus"],
+                "message_bus_stamp_middleware" => [BusNameStampMiddlewareStaticFactory::class, "message_bus"],
+                "message_bus_sender_middleware" => [MessageSenderMiddlewareStaticFactory::class, "message_bus"],
+                "message_bus_handler_middleware" => [MessageHandlerMiddlewareStaticFactory::class, "message_bus"],
+                ExampleMessageHandler::class => ExampleMessageHandlerFactory::class
+            ],
+            "aliases" => [
+               MessageBusInterface::class => "message_bus"
+            ]
+        ];
+    }
+
+    private function busConfig()
+    {
+        return [
+            "message_bus" => [
+                'allows_zero_handlers' => false, // Means that it's an error if no handlers are defined for a given message
+
+                /**
+                 * Each bus needs middleware to do anything useful.
+                 *
+                 * Below is a minimal configuration to handle messages
+                 */
+                'middleware' => [
+                    // … Middleware that inspects the message before it has been sent to a transport would go here.
+                    "message_bus_stamp_middleware",
+                    'message_bus_sender_middleware', // Sends messages via a transport if configured.
+                    'message_bus_handler_middleware', // Executes the handlers configured for the message
+                ],
+
+                /**
+                 * Map messages to one or more handlers:
+                 *
+                 * Two locators are shipped, 1 message type to 1 handler and 1 message type to many handlers.
+                 * Both locators operate on the basis that handlers are available in the container.
+                 *
+                 */
+                'handler_locator' => OneToManyFqcnContainerHandlerLocator::class,
+                'handlers' => [
+                    ExampleMessage::class => [ExampleMessageHandler::class],
+                ],
+
+                /**
+                 * Routes define which transport(s) that messages dispatched on this bus should be sent with.
+                 *
+                 * The * wildcard applies to all messages.
+                 * The transport for each route must be an array of one or more transport identifiers. Each transport
+                 * is retrieved from the DI container by this value.
+                 *
+                 * An empty routes definition would mean that messages would be handled immediately and synchronously,
+                 * i.e. no transport would be used.
+                 *
+                 * Route specific messages to specific transports by using the message name as the key.
+                 */
+                'routes' => [
+                    ExampleMessage::class => ["redis"],
+                ],
+            ]
+        ];
+    }
+}

src/App/ConfigProvider.php

@@ -0,0 +1,18 @@
+<?php
+
+namespace Queue\App\Message;
+
+class ExampleMessage
+{
+    public function __construct(
+        private array $payload,
+    )
+    {
+    }
+
+    public function getPayload(): array
+    {
+        return $this->payload;
+    }
+
+}

src/App/Message/ExampleMessage.php

@@ -0,0 +1,21 @@
+<?php
+
+namespace Queue\App\Message;
+
+use Dot\Log\Logger;
+use Psr\Container\ContainerInterface;
+class ExampleMessageHandler
+{
+    public function __construct(private readonly ContainerInterface $container)
+    {
+    }
+
+    public function __invoke(ExampleMessage $message)
+    {
+
+        /** @var Logger $logger */
+        $logger = $this->container->get("dot-log.queue-log");
+
+        $logger->info("message: " . $message->getPayload()['foo'] ?? null);
+    }
+}

src/App/Message/ExampleMessageHandler.php

@@ -0,0 +1,14 @@
+<?php
+
+namespace Queue\App\Message;
+
+use Psr\Container\ContainerInterface;
+
+class ExampleMessageHandlerFactory
+{
+    public function __invoke(ContainerInterface $container)
+    {
+        return new ExampleMessageHandler($container);
+    }
+
+}

src/App/Message/ExampleMessageHandlerFactory.php

@@ -70,7 +70,8 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         $pidManager = $this->pidManager;
 
-        $server->on('start', function () use ($server, $pidManager, $processName) {
+        $server->on('start', function () use ($server, $pidManager, $processName, $output) {
+            $output->writeln('<info>Server started...</info>');
             $pidManager->write($server->master_pid, $server->manager_pid);
 
             swoole_set_process_name(sprintf('%s-master', $processName));

src/Swoole/Command/StartCommand.php

@@ -4,7 +4,10 @@
 
 use Psr\Container\ContainerInterface;
 
+use Queue\App\Message\ExampleMessage;
 use Swoole\Server as TCPSwooleServer;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Symfony\Component\Messenger\Stamp\DelayStamp;
 
 class TCPServerDelegator
 {
@@ -13,14 +16,23 @@ public function __invoke(ContainerInterface $container, string $serviceName, cal
         /** @var TCPSwooleServer $server */
         $server = $callback();
 
+        /** @var MessageBusInterface $bus */
+        $bus = $container->get(MessageBusInterface::class);
+
         $logger = $container->get("dot-log.queue-log");
 
         $server->on('Connect', function ($server, $fd) {
             echo "Client: Connect.\n";
         });
 
         // Register the function for the event `receive`
-        $server->on('receive', function ($server, $fd, $from_id, $data) use ($logger) {
+        $server->on('receive', function ($server, $fd, $from_id, $data) use ($logger, $bus) {
+
+            $bus->dispatch(new ExampleMessage(["foo" => $data]));
+            $bus->dispatch(new ExampleMessage(["foo" => "with 5 seconds delay"]), [
+                new DelayStamp(5000)
+            ]);
+
             $server->send($fd, "Server: {$data}");
             $logger->notice("Request received on receive", [
                 'fd' => $fd,