Lines 99.12% 113 / 114
Methods 94.11% 16 / 17
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 4 / 4 100.00% 1 / 1 3
 emit 100.00% 7 / 7 100.00% 1 / 1 1
 handle 100.00% 22 / 22 100.00% 1 / 1 6
 reportToTracy 100.00% 2 / 2 100.00% 1 / 1 3
 shutdownReason 100.00% 2 / 2 100.00% 1 / 1 2
 dispatch 100.00% 9 / 9 100.00% 1 / 1 5
 handleMessage 100.00% 18 / 18 100.00% 1 / 1 4
 retryMessage 100.00% 15 / 15 100.00% 1 / 1 8
 usesRetryTrait 100.00% 8 / 8 100.00% 1 / 1 4
 nextRetry 100.00% 2 / 2 100.00% 1 / 1 1
 hasHandlers 100.00% 1 / 1 100.00% 1 / 1 1
 registerHandler 100.00% 2 / 2 100.00% 1 / 1 1
 registerHandlers 100.00% 3 / 3 100.00% 1 / 1 2
 unregisterAllHandlers 100.00% 2 / 2 100.00% 1 / 1 1
 unregisterHandler 85.71% 6 / 7 0.00% 0 / 1 2.01
 messageLoggerContext 100.00% 8 / 8 100.00% 1 / 1 1
 log 100.00% 2 / 2 100.00% 1 / 1 2
22final class Dispatcher implements DispatcherInterface
23{
24    public const int DEFAULT_PRIORITY = 100;
25
26    /**
27     * All registered handlers
28     *
29     * @var array<string, array<int, HandlerInterface>>
30     */
31    private array $handlers = [];
32
33    private readonly DateTimeImmutable $startTime;
34
35    private readonly Randomizer $randomizer;
36
37    /**
38     * Create new Dispatcher
39     *
40     * @param Randomizer|null $randomizer Source of the retry jitter. Pass one with a seeded engine
41     *                                    (e.g. new Randomizer(new Mt19937(42))) for reproducible retry times.
42     */
43    public function __construct(
44        private readonly DriverInterface $driver,
45        private readonly ?LoggerInterface $logger = null,
46        private readonly ?ShutdownInterface $shutdown = null,
47        ?Randomizer $randomizer = null,
48    ) {
49        $this->randomizer = $randomizer ?? new Randomizer();
50        $this->startTime = new DateTimeImmutable();
51
52        // check if driver use ShutdownTrait
53        if ($this->shutdown instanceof ShutdownInterface && method_exists($this->driver, 'setShutdown')) {
54            $this->driver->setShutdown($this->shutdown);
55        }
56    }
57
58    /**
59     * @return $this
60     *
61     * @deprecated - use Emitter::emit method instead
62     */
63    public function emit(MessageInterface $message, int $priority = self::DEFAULT_PRIORITY): DispatcherInterface
64    {
65        $this->driver->send($message, $priority);
66
67        $this->log(
68            LogLevel::INFO,
69            "Dispatcher send message #{$message->getId()} with priority {$priority} to driver " . $this->driver::class,
70            $this->messageLoggerContext($message),
71        );
72        return $this;
73    }
74
75    /**
76     * Basic method for background job to star listening.
77     *
78     * This method hook to driver wait() method and start listening events.
79     * Method is blocking, so when you call it all processing will stop.
80     * WARNING! Don't use it on web server calls. Run it only with cli.
81     *
82     * An Exception from the driver is logged at error and rethrown, so the worker exits non-zero.
83     *
84     * @param int[] $priorities
85     *
86     * @throws Exception
87     */
88    #[\Override]
89    public function handle(array $priorities = []): void
90    {
91        // stop after the current message on SIGTERM / SIGINT; the previous handlers come back when this returns
92        $signals = SignalStop::install();
93        try {
94            $this->driver->wait(function (MessageInterface $message, int $priority = Dispatcher::DEFAULT_PRIORITY): bool {
95                $this->log(
96                    LogLevel::INFO,
97                    "Start handle message #{$message->getId()} ({$message->getType()}) priority:{$priority}",
98                    $this->messageLoggerContext($message),
99                );
100
101                $result = $this->dispatch($message);
102
103                if (SignalStop::received() !== null || ($this->shutdown instanceof ShutdownInterface && $this->shutdown->shouldShutdown($this->startTime))) {
104                    throw new ShutdownException('Shutdown');
105                }
106
107                return $result;
108            }, $priorities);
109        } catch (ShutdownException) {
110            $this->log(LogLevel::NOTICE, $this->shutdownReason());
111        } catch (Exception $exception) {
112            // a driver failure (lost connection, broken payload): log it and let the worker exit non-zero
113            $this->log(
114                LogLevel::ERROR,
115                'Exiting hermes dispatcher - driver error: ' . $exception::class . " - {$exception->getMessage()}",
116                ['exception' => $exception],
117            );
118            $this->reportToTracy($exception);
119            throw $exception;
120        }
121    }
122
123    /**
124     * Tracy is optional (suggest): write to it only when it is installed and already enabled.
125     */
126    private function reportToTracy(Throwable $exception): void
127    {
128        if (class_exists(Debugger::class) && Debugger::isEnabled()) {
129            Debugger::log($exception, Debugger::EXCEPTION);
130        }
131    }
132
133    private function shutdownReason(): string
134    {
135        $signal = SignalStop::received();
136
137        return $signal === null ? 'Exiting hermes dispatcher - shutdown' : 'Exiting hermes dispatcher - received ' . SignalStop::describe($signal);
138    }
139
140    /**
141     * Dispatch message
142     */
143    private function dispatch(MessageInterface $message): bool
144    {
145        $type = $message->getType();
146
147        if (!$this->hasHandlers($type)) {
148            return true;
149        }
150
151        $result = true;
152
153        foreach ($this->handlers[$type] as $handler) {
154            $handlerResult = $this->handleMessage($handler, $message);
155
156            if ($result && !$handlerResult) {
157                $result = false;
158            }
159        }
160
161        return $result;
162    }
163
164    /**
165     * Handle given message with given handler
166     */
167    private function handleMessage(HandlerInterface $handler, MessageInterface $message): bool
168    {
169        // check if handler implements Psr\Log\LoggerAwareInterface (you can use \Psr\Log\LoggerAwareTrait)
170        if ($this->logger instanceof LoggerInterface && method_exists($handler, 'setLogger')) {
171            $handler->setLogger($this->logger);
172        }
173
174        try {
175            $result = $handler->handle($message);
176
177            $this->log(
178                LogLevel::INFO,
179                "End handle message #{$message->getId()} ({$message->getType()})",
180                $this->messageLoggerContext($message),
181            );
182        } catch (Throwable $e) {
183            // an Error (TypeError, ...) is a failed message too, not a reason to stop the worker
184            $this->log(
185                LogLevel::ERROR,
186                'Handler ' . $handler::class . " throws exception - {$e->getMessage()}",
187                ['error' => $e, 'message' => $this->messageLoggerContext($message), 'exception' => $e],
188            );
189            $this->reportToTracy($e);
190
191            $this->retryMessage($message, $handler, $e);
192
193            $result = false;
194        }
195        return $result;
196    }
197
198    /**
199     * Helper function for sending retrying message back to driver
200     */
201    private function retryMessage(MessageInterface $message, HandlerInterface $handler, Throwable $exception): void
202    {
203        if ($handler instanceof RetryableHandlerInterface) {
204            $retry = $message->getRetries() < $handler->maxRetry() && $handler->canRetry($message, $exception);
205        } elseif ($this->usesRetryTrait($handler) && method_exists($handler, 'canRetry') && method_exists($handler, 'maxRetry')) {
206            // a handler that uses RetryTrait without the interface, as in 5.x
207            $retry = $message->getRetries() < $handler->maxRetry() && $handler->canRetry($message, $exception) === true;
208        } else {
209            return;
210        }
211        if (!$retry) {
212            return;
213        }
214
215        $this->driver->send(new Message(
216            $message->getType(),
217            $message->getPayload(),
218            $message->getId(),
219            $message->getCreated(),
220            $this->nextRetry($message),
221            $message->getRetries() + 1,
222        ));
223    }
224
225    /**
226     * True when the handler's class, a parent class or one of their traits uses RetryTrait.
227     */
228    private function usesRetryTrait(HandlerInterface $handler): bool
229    {
230        $pending = [];
231        for ($class = $handler::class; $class !== false; $class = get_parent_class($class)) {
232            $pending = array_merge($pending, class_uses($class));
233        }
234        while (($trait = array_pop($pending)) !== null) {
235            if ($trait === RetryTrait::class) {
236                return true;
237            }
238            $pending = array_merge($pending, class_uses($trait));
239        }
240
241        return false;
242    }
243
244    /**
245     * Calculate next retry
246     *
247     * Inspired by ruby sidekiq (https://github.com/mperham/sidekiq/wiki/Error-Handling#automatic-job-retry)
248     */
249    private function nextRetry(MessageInterface $message): float
250    {
251        $retries = $message->getRetries();
252        return microtime(true) + $retries ** 4 + 15 + $this->randomizer->getInt(1, 30) * ($retries + 1);
253    }
254
255    /**
256     * Check if actual dispatcher has handler for given type
257     */
258    private function hasHandlers(string $type): bool
259    {
260        return ($this->handlers[$type] ?? []) !== [];
261    }
262
263    #[\Override]
264    public function registerHandler(string $type, HandlerInterface $handler): DispatcherInterface
265    {
266        $this->handlers[$type][] = $handler;
267        return $this;
268    }
269
270    #[\Override]
271    public function registerHandlers(string $type, array $handlers): DispatcherInterface
272    {
273        foreach ($handlers as $handler) {
274            $this->registerHandler($type, $handler);
275        }
276        return $this;
277    }
278
279    #[\Override]
280    public function unregisterAllHandlers(): DispatcherInterface
281    {
282        $this->handlers = [];
283        return $this;
284    }
285
286    #[\Override]
287    public function unregisterHandler(string $type, HandlerInterface $handler): DispatcherInterface
288    {
289        if (!isset($this->handlers[$type])) {
290            return $this;
291        }
292        $this->handlers[$type] = array_filter(
293            $this->handlers[$type],
294            fn(HandlerInterface $registeredHandler): bool => $registeredHandler !== $handler,
295        );
296
297        return $this;
298    }
299
300    /**
301     * Serialize message to logger context
302     *
303     * @return array<string, mixed>
304     */
305    private function messageLoggerContext(MessageInterface $message): array
306    {
307        return [
308            'id' => $message->getId(),
309            'created' => $message->getCreated(),
310            'type' => $message->getType(),
311            'payload' => $message->getPayload(),
312            'retries' => $message->getRetries(),
313            'execute_at' => $message->getExecuteAt(),
314        ];
315    }
316
317    /**
318     * Interal log method wrapper
319     *
320     * @param array<mixed> $context
321     */
322    private function log(string $level, string $message, array $context = []): void
323    {
324        if ($this->logger instanceof LoggerInterface) {
325            $this->logger->log($level, $message, $context);
326        }
327    }
328}