Handle errors coming from Pusher

Author: WyriHaximus

src/AsyncClient.php

@@ -4,13 +4,17 @@
 
 use React\Dns\Resolver\Resolver;
 use React\EventLoop\LoopInterface;
+use RuntimeException;
 use Rx\Disposable\CallbackDisposable;
 use Rx\Observable;
 use Rx\ObserverInterface;
 use Rx\Scheduler;
 use Rx\Websocket\Client as WebsocketClient;
 use Rx\Websocket\MessageSubject;
+use Rx\Websocket\WebsocketErrorException;
 use Throwable;
+use function React\Promise\reject;
+use function React\Promise\resolve;
 
 final class AsyncClient
 {
@@ -88,29 +92,35 @@ public function __construct(Observable $client)
                     ->startWith($x);
             })
 
-            // Handle connection level errors
-            ->retryWhen(function (Observable $errors) {
-                return $errors->flatMap(function (Throwable $throwable) {
-                    return $this->handleLowLevelError($throwable);
-                });
-            })
-
             // Decode JSON
             ->_ApiClients_jsonDecode()
 
             // Deal with connection established messages
-            ->map(function (array $message) {
+            ->flatMap(function (array $message) {
                 $this->delay = self::DEFAULT_DELAY;
 
                 $event = Event::createFromMessage($message);
 
+                if ($event->getEvent() === 'pusher:error') {
+                    return Observable::fromPromise(reject(new PusherErrorException($event->getData()['message'], $event->getData()['code'])));
+                }
+
                 if ($event->getEvent() === 'pusher:connection_established') {
                     $this->setActivityTimeout($event);
                 }
 
-                return $event;
+                return Observable::fromPromise(resolve($event));
+            })
+
+            // Handle connection level errors
+            ->retryWhen(function (Observable $errors) {
+                return $errors->flatMap(function (Throwable $throwable) {
+                    return $this->handleLowLevelError($throwable);
+                });
             })
-            ->share();
+
+        // Share client
+        ->share();
     }
 
     /**
@@ -206,8 +216,33 @@ public function send(array $message): bool
         return true;
     }
 
+    /**
+     *  Handle errors as described at https://pusher.com/docs/pusher_protocol#error-codes.
+     */
     private function handleLowLevelError(Throwable $throwable)
     {
+        if (!($throwable instanceof WebsocketErrorException) && !($throwable instanceof RuntimeException) && !($throwable instanceof PusherErrorException)) {
+            return Observable::fromPromise(reject($throwable));
+        }
+
+        $code = $throwable->getCode();
+        $pusherError = ($throwable instanceof WebsocketErrorException || $throwable instanceof PusherErrorException);
+
+        // Errors 4000-4099, don't retry connecting
+        if ($pusherError && $code >= 4000 && $code <= 4099) {
+            return Observable::fromPromise(reject($throwable));
+        }
+
+        // Errors 4100-4199 reconnect after 1 or more seconds, we do it after 1.001 second
+        if ($pusherError && $code >= 4100 && $code <= 4199) {
+            return Observable::timer(1001);
+        }
+
+        // Errors 4200-4299 connection closed by Pusher, reconnect immediately, we wait 0.001 second
+        if ($pusherError && $code >= 4200 && $code <= 4299) {
+            return Observable::timer(1);
+        }
+
         $this->delay *= 2;
 
         return Observable::timer($this->delay);

tests/AsyncClientTest.php

@@ -1,11 +1,16 @@
-<?php
-declare(strict_types=1);
+<?php declare(strict_types=1);
 
 namespace ApiClients\Tests\Client\Pusher;
 
+use React\Dns\Resolver\Resolver;
+use function React\Promise\reject;
+use RuntimeException;
 use ApiClients\Client\Pusher\AsyncClient;
 use ApiClients\Tools\TestUtilities\TestCase;
 use React\EventLoop\Factory;
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+use Rx\Websocket\Client;
 
 final class AsyncClientTest extends TestCase
 {
@@ -15,4 +20,31 @@ public function testCreateFactory()
         $appId = uniqid('app-id-', true);
         self::assertInstanceOf(AsyncClient::class, AsyncClient::create($loop, $appId));
     }
+
+    public function testConnectionError()
+    {
+        $capturedException = null;
+        $error = new RuntimeException();
+        $observable = Observable::error($error, new ImmediateScheduler());
+        $client = new AsyncClient($observable);
+        $client->channel('test')->subscribe(
+            function () {},
+            function ($e) use (&$capturedException) {
+                $capturedException = $e;
+            }
+        );
+        self::assertNull($capturedException);
+    }
+
+    public function testConnectionRetry()
+    {
+        $loop = Factory::create();
+        $error = new RuntimeException('', 4199);
+        $resolver = $this->prophesize(Resolver::class);
+        $resolver->resolve('ws.pusherapp.com')->shouldBeCalled()->willReturn(reject($error));
+        $client = AsyncClient::create($loop, 'abc', $resolver->reveal());
+        $client->channel('test')->subscribe();
+        $loop->addTimer(1, function () {});
+        $loop->run();
+    }
 }