feat: add a timeout for redis operations

README.md

@@ -101,6 +101,10 @@ To launch the hub with the Redis transport, change the `TRANSPORT_DSN` environme
 TRANSPORT_DSN="redis://127.0.0.1:6379" ./bin/freddie
 ```
 
+Optional parameters you can pass in the DSN's query string:
+- `pingInterval` - regularly ping Redis connection, which will help detect outages (default `2.0`)
+- `readTimeout` - max duration in seconds of a ping or publish request (default `null`)
+
 _Alternatively, you can set this variable into `.env.local`._
 
 ## Advantages and limitations

composer.json

@@ -21,6 +21,7 @@
         "nyholm/dsn": "^2.0",
         "phpdocumentor/reflection-docblock": "^5.3",
         "react/async": "^4.0.0",
+        "react/promise-timer": "^1.10",
         "rize/uri-template": "^0.3.4",
         "symfony/console": "^5.4.0|^6.0.0",
         "symfony/dotenv": "^5.4.0|^6.0.0",

phpstan.neon

@@ -5,3 +5,5 @@ parameters:
 
     ignoreErrors:
         - '#React\\Promise\\PromiseInterface is not generic#'
+        - '#Unable to resolve the template type T in call to function React\\Promise\\Timer\\timeout#'
+        - '#Unable to resolve the template type T in call to function Freddie\\maybeTimeout#'

src/Hub/Transport/Redis/RedisTransport.php

@@ -13,10 +13,9 @@
 use Generator;
 use React\EventLoop\Loop;
 use React\Promise\PromiseInterface;
-use RuntimeException;
 use Symfony\Component\OptionsResolver\OptionsResolver;
-use Throwable;
 
+use function Freddie\maybeTimeout;
 use function React\Async\await;
 use function React\Promise\resolve;
 
@@ -45,6 +44,7 @@ public function __construct(
             'channel' => 'mercure',
             'key' => 'mercureUpdates',
             'pingInterval' => 2.0,
+            'readTimeout' => null,
         ]);
         $this->options = $resolver->resolve($options);
         if ($this->options['pingInterval']) {
@@ -59,6 +59,7 @@ private function ping(): void
     {
         /** @var PromiseInterface $ping */
         $ping = $this->redis->ping(); // @phpstan-ignore-line
+        $ping = maybeTimeout($ping, $this->options['readTimeout']);
         $ping->then(
             onRejected: Hub::die(...),
         );
@@ -80,7 +81,10 @@ public function publish(Update $update): PromiseInterface
         $this->init();
         $payload = $this->serializer->serialize($update);
 
-        return $this->redis->publish($this->options['channel'], $payload) // @phpstan-ignore-line
+        /** @var PromiseInterface $promise */
+        $promise = $this->redis->publish($this->options['channel'], $payload); // @phpstan-ignore-line
+
+        return maybeTimeout($promise, $this->options['readTimeout'])
             ->then(fn () => $this->store($update))
             ->then(fn () => $update);
     }

src/functions.php

@@ -7,10 +7,12 @@
 use Freddie\Helper\FlatQueryParser;
 use Freddie\Helper\TopicHelper;
 use Psr\Http\Message\ServerRequestInterface;
+use React\Promise\PromiseInterface;
 
 use function BenTools\QueryString\query_string;
 use function in_array;
 use function is_string;
+use function React\Promise\Timer\timeout;
 use function settype;
 use function strtolower;
 use function trim;
@@ -53,3 +55,13 @@ function extract_last_event_id(ServerRequestInterface $request): ?string
         ?? $qs->getParam('LAST-EVENT-ID')
         ?? null;
 }
+
+/**
+ * @template T
+ * @param PromiseInterface<T> $promise
+ * @return PromiseInterface<T>
+ */
+function maybeTimeout(PromiseInterface $promise, ?float $time = null): PromiseInterface
+{
+    return null === $time ? $promise : timeout($promise, $time);
+}