Initial implementation of Redis as a pub/sub backend, WIP

TODO:
- Presence channels need the user lists stored in Redis (tricky, requires a lot of changes and async code in HTTP controllers)
- Channels in Redis should be scoped by the app ID

Author: francislavoie

PubSub/PubSubInterface.php

@@ -25,9 +25,8 @@
         "php": "^7.1",
         "ext-json": "*",
         "cboden/ratchet": "^0.4.1",
-        "clue/block-react": "^1.3",
         "clue/buzz-react": "^2.5",
-        "clue/redis-react": "^2.2",
+        "clue/redis-react": "^2.3",
         "facade/ignition-contracts": "^1.0",
         "guzzlehttp/psr7": "^1.5",
         "illuminate/broadcasting": "5.7.* || 5.8.* || ^6.0",

PubSub/Redis/RedisClient.php

@@ -9,8 +9,7 @@
 use React\Dns\Resolver\ResolverInterface;
 use React\EventLoop\Factory as LoopFactory;
 use React\Dns\Resolver\Factory as DnsFactory;
-use BeyondCode\LaravelWebSockets\Statistics\DnsResolver;
-use BeyondCode\LaravelWebSockets\PubSub\PubSubInterface;
+use BeyondCode\LaravelWebSockets\PubSub\ReplicationInterface;
 use BeyondCode\LaravelWebSockets\Statistics\DnsResolver;
 use BeyondCode\LaravelWebSockets\Facades\StatisticsLogger;
 use BeyondCode\LaravelWebSockets\Facades\WebSocketsRouter;
@@ -146,13 +145,11 @@ protected function configurePubSubReplication()
         }
 
         if (config('websockets.replication.driver') === 'redis') {
-            $connection = (new RedisClient())->subscribe($this->loop);
+            app()->singleton(ReplicationInterface::class, function () {
+                return (new RedisClient())->boot($this->loop);
+            });
         }
 
-        app()->singleton(PubSubInterface::class, function () use ($connection) {
-            return $connection;
-        });
-
         return $this;
     }
 

composer.json

@@ -46,7 +46,7 @@ public function onOpen(ConnectionInterface $connection, RequestInterface $reques
 
         $this->requestBuffer = (string) $request->getBody();
 
-        if (! $this->checkContentLength()) {
+        if (! $this->verifyContentLength()) {
             return;
         }
 
@@ -64,16 +64,16 @@ public function onMessage(ConnectionInterface $from, $msg)
     {
         $this->requestBuffer .= $msg;
 
-        if (! $this->checkContentLength()) {
+        if (! $this->verifyContentLength()) {
             return;
         }
 
         $this->handleRequest($from);
     }
 
-    protected function checkContentLength()
+    protected function verifyContentLength()
     {
-        return strlen($this->requestBuffer) !== $this->contentLength;
+        return strlen($this->requestBuffer) === $this->contentLength;
     }
 
     protected function handleRequest(ConnectionInterface $connection)