Moved all event-dispatching stuff to its own module

module/EventDispatcher/src/Async/Task.php

@@ -0,0 +1,33 @@
+<?php
+declare(strict_types=1);
+
+namespace Shlinkio\Shlink\EventDispatcher\Async;
+
+use Psr\Container\ContainerInterface;
+
+use function get_class;
+use function sprintf;
+
+class Task
+{
+    /** @var string */
+    private $regularListenerName;
+    /** @var object */
+    private $event;
+
+    public function __construct(string $regularListenerName, object $event)
+    {
+        $this->regularListenerName = $regularListenerName;
+        $this->event = $event;
+    }
+
+    public function __invoke(ContainerInterface $container)
+    {
+        ($container->get($this->regularListenerName))($this->event);
+    }
+
+    public function toString(): string
+    {
+        return sprintf('Listener -> "%s", Event -> "%s"', $this->regularListenerName, get_class($this->event));
+    }
+}

module/EventDispatcher/src/Async/TaskRunner.php

@@ -0,0 +1,55 @@
+<?php
+declare(strict_types=1);
+
+namespace Shlinkio\Shlink\EventDispatcher\Async;
+
+use Psr\Container\ContainerInterface;
+use Psr\Log\LoggerInterface;
+use Swoole\Http\Server as HttpServer;
+use Throwable;
+
+use function get_class;
+use function gettype;
+use function is_object;
+
+class TaskRunner
+{
+    /** @var LoggerInterface */
+    private $logger;
+    /** @var ContainerInterface */
+    private $container;
+
+    public function __construct(LoggerInterface $logger, ContainerInterface $container)
+    {
+        $this->logger = $logger;
+        $this->container = $container;
+    }
+
+    public function __invoke(HttpServer $server, int $taskId, int $fromId, $task): void
+    {
+        if (! $task instanceof Task) {
+            $this->logger->error('Invalid task provided to task worker: {type}', [
+                'type' => is_object($task) ? get_class($task) : gettype($task),
+            ]);
+            $server->finish('');
+            return;
+        }
+
+        $this->logger->notice('Starting work on task {taskId}: {task}', [
+            'taskId' => $taskId,
+            'task' => $task->toString(),
+        ]);
+
+        try {
+            $task($this->container);
+        } catch (Throwable $e) {
+            $this->logger->error('Error processing task {taskId}: {e}', [
+                'taskId' => $taskId,
+                'e' => $e,
+            ]);
+        } finally {
+            // Notify the server that processing of the task has finished:
+            $server->finish('');
+        }
+    }
+}

module/EventDispatcher/src/Async/TaskRunnerDelegator.php

@@ -0,0 +1,29 @@
+<?php
+declare(strict_types=1);
+
+namespace Shlinkio\Shlink\EventDispatcher\Async;
+
+use Interop\Container\ContainerInterface;
+use Psr\Log\LoggerInterface;
+use Swoole\Http\Server as HttpServer;
+use Zend\ServiceManager\Factory\DelegatorFactoryInterface;
+
+class TaskRunnerDelegator implements DelegatorFactoryInterface
+{
+    public function __invoke(
+        ContainerInterface $container,
+        $name,
+        callable $callback,
+        array $options = null
+    ): HttpServer {
+        $server = $callback();
+        $logger = $container->get(LoggerInterface::class);
+
+        $server->on('task', $container->get(TaskRunner::class));
+        $server->on('finish', function (HttpServer $server, int $taskId) use ($logger) {
+            $logger->notice('Task #{taskId} has finished processing', ['taskId' => $taskId]);
+        });
+
+        return $server;
+    }
+}

module/EventDispatcher/src/Async/TaskRunnerFactory.php

@@ -0,0 +1,17 @@
+<?php
+declare(strict_types=1);
+
+namespace Shlinkio\Shlink\EventDispatcher\Async;
+
+use Interop\Container\ContainerInterface;
+use Psr\Log\LoggerInterface;
+use Zend\ServiceManager\Factory\FactoryInterface;
+
+class TaskRunnerFactory implements FactoryInterface
+{
+    public function __invoke(ContainerInterface $container, $requestedName, array $options = null): TaskRunner
+    {
+        $logger = $container->get(LoggerInterface::class);
+        return new TaskRunner($logger, $container);
+    }
+}