- predis -> 1.1.10 with ErrorResponse handling (#147)

Author: vtl3000

composer.json

@@ -30,8 +30,8 @@
  "ext-pcre": "*",
  "ext-posix": "*",
  "ext-sockets": "*",
- "monolog/monolog": "^2.0",
- "predis/predis": "1.1.*",
+ "monolog/monolog": "^1.23 || ^2.0",
+ "predis/predis": "^1.1.10",
  "psr/log": "^1",
  "ramsey/uuid": "^3.7 || ^4",
  "seld/signal-handler": "1.1.*"

src/Subscribers/WatchdogSubscriber.php

@@ -105,58 +105,82 @@ public function watchdog(string $jid, string $worker, ?int $pid = null): void
 
  call_user_func_array($callable, $this->channels);
 
- /** @var \stdClass $message */
- foreach ($pubsub as $message) {
- if ($message->kind !== 'message' || empty($message->payload)) {
- continue;
- }
-
- $payload = json_decode($message->payload, true);
- if (empty($payload)) {
- continue;
- }
-
- if (empty($payload['event']) || !is_array($payload)) {
- continue;
- }
-
- if (!in_array($payload['event'], self::WATCHDOG_EVENTS, true) || empty($payload['jid'])) {
- continue;
- }
-
- if ($payload['jid'] !== $jid) {
- continue;
- }
-
- $who = 'watchdog:' . $worker;
-
- switch ($payload['event']) {
- case self::LOCK_LOST:
- if (!empty($payload['worker']) && $payload['worker'] === $worker) {
- $this->logger->info(
- "{type}: sending SIGKILL to child {$pid}; job {jid} handed out to another worker",
- ['type' => $who, 'jid' => $jid]
- );
-
- $this->system->posixKill($pid, SIGKILL);
- $pubsub->stop();
- }
- break;
- case self::CANCELED:
- if (!empty($payload['worker']) && $payload['worker'] === $worker) {
- $this->logger->info(
- "{type}: sending SIGKILL to child {$pid}; job {jid} canceled",
- ['type' => $who, 'jid' => $jid]
- );
- $this->system->posixKill($pid, SIGKILL);
+ try {
+ /** @var \stdClass $message */
+ foreach ($pubsub as $message) {
+ if ($message->kind !== 'message' || empty($message->payload)) {
+ continue;
+ }
+
+ $payload = json_decode($message->payload, true);
+ if (empty($payload)) {
+ continue;
+ }
+
+ if (empty($payload['event']) || !is_array($payload)) {
+ continue;
+ }
+
+ if (!in_array($payload['event'], self::WATCHDOG_EVENTS, true) || empty($payload['jid'])) {
+ continue;
+ }
+
+ if ($payload['jid'] !== $jid) {
+ continue;
+ }
+
+ switch ($payload['event']) {
+ case self::LOCK_LOST:
+ if (!empty($payload['worker']) && $payload['worker'] === $worker) {
+ $this->logger->info(
+ "{type}: sending SIGKILL to child {$pid}; job {jid} handed out to another worker",
+ [
+ 'type' => 'watchdog:' . $worker,
+ 'jid' => $jid,
+ ]
+ );
+
+ $this->system->posixKill($pid, SIGKILL);
+ $pubsub->stop();
+ }
+ break;
+ case self::CANCELED:
+ if (!empty($payload['worker']) && $payload['worker'] === $worker) {
+ $this->logger->info(
+ "{type}: sending SIGKILL to child {$pid}; job {jid} canceled",
+ [
+ 'type' => 'watchdog:' . $worker,
+ 'jid' => $jid,
+ ]
+ );
+ $this->system->posixKill($pid, SIGKILL);
+ $pubsub->stop();
+ }
+ break;
+ case self::COMPLETED:
+ case self::FAILED:
  $pubsub->stop();
- }
- break;
- case self::COMPLETED:
- case self::FAILED:
- $pubsub->stop();
- break;
+ break;
+ }
  }
+ } catch (\Throwable $exception) {
+ $this->logger->critical(
+ "Critical error",
+ [
+ 'type' => 'watchdog:' . $worker,
+ 'jid' => $jid,
+ 'exception' => $exception
+ ]
+ );
+ $this->logger->info(
+ "{type}: sending SIGKILL to child {$pid}; job {jid} canceled due to exception",
+ [
+ 'type' => 'watchdog:' . $worker,
+ 'jid' => $jid,
+ ]
+ );
+ $this->system->posixKill($pid, SIGKILL);
+ $pubsub->stop();
  }
 
  // Always unset the pubsub consumer instance when you are done! The