Based a refactored AsyncClient on https://gist.githubusercontent.com/davidwdan/b10e9cd434668125946c50b6fc75b323/raw/0c6ed0c2130707a021c32a3208cfa6ec464dcbf5/pusher.php

Author: WyriHaximus

examples/reddit.php

@@ -18,10 +18,10 @@
 $subReddits = $argv;
 array_shift($subReddits);
 foreach ($subReddits as $subReddit) {
-    $client->subscribe($subReddit)->subscribe(new CallbackObserver(function ($event) {
-        echo 'Channel: ', $event['channel'], PHP_EOL;
-        echo 'Event: ', $event['event'], PHP_EOL;
-        echo 'Data: ', $event['data'], PHP_EOL;
+    $client->channel($subReddit)->subscribe(new CallbackObserver(function ($event) {
+        echo 'Channel: ', $event->channel, PHP_EOL;
+        echo 'Event: ', $event->event, PHP_EOL;
+        echo 'Data: ', $event->data, PHP_EOL;
     }));
 }
 

src/AsyncClient.php

@@ -4,13 +4,16 @@
 namespace WyriHaximus\Pusher;
 
 use React\EventLoop\LoopInterface;
+use Rx\Disposable\CallbackDisposable;
 use Rx\Observable;
 use Rx\ObservableInterface;
 use Rx\Observer\CallbackObserver;
 use Rx\ObserverInterface;
 use Rx\React\Promise;
 use Rx\Scheduler\EventLoopScheduler;
+use Rx\SchedulerInterface;
 use Rx\Websocket\Client;
+use Rx\Websocket\MessageSubject;
 use WyriHaximus\ApiClient\Transport\Client as Transport;
 use WyriHaximus\ApiClient\Transport\Factory;
 use function React\Promise\resolve;
@@ -23,7 +26,7 @@ class AsyncClient
     protected $app;
     protected $url;
     protected $client;
-    protected $connection;
+    protected $message;
     protected $channels = [];
 
     public function __construct(LoopInterface $loop, string $app, Transport $transport = null)
@@ -42,40 +45,45 @@ public function __construct(LoopInterface $loop, string $app, Transport $transpo
             $this->app .
             '?client=wyrihaximus-php-pusher-client&version=0.0.1&protocol=7'
         ;
-
-        $this->client = new Client($this->url);
-        $this->connection = Observable::create(function (ObserverInterface $observer) {
-            $this->client->subscribe(new CallbackObserver(function ($ms) use ($observer) {
-                $ms->subscribe(new CallbackObserver(
-                    function ($message) use ($observer) {
-                        $observer->onNext($message);
-                    }
-                ), new EventLoopScheduler(getLoop()));
-            }));
-        }, new EventLoopScheduler($loop))->map(function ($message) {
-            return json_decode((string)$message, true);
-        });
+        $this->client   = (new Client($this->url))->shareReplay(1); //Only create one connection and share the most recent among all subscriber
+        $this->messages = $this->client
+            ->flatMap(function (MessageSubject $ms) {
+                return $ms;
+            })
+            ->map('json_decode');
     }
 
-    public function subscribe(string $channel): ObservableInterface
+    public function channel(string $channel): ObservableInterface
     {
-        $this->channels[$channel] = $channel;
-        $this->send([
-            'event' => 'pusher:subscribe',
-            'data' => [
-                'channel' => $channel,
-            ],
-        ]);
+        $channelMessages = $this->messages->filter(function ($event) use ($channel) {
+            return isset($event->channel) && $event->channel == $channel;
+        });
+
+        $events = Observable::create(function (ObserverInterface $observer, SchedulerInterface $scheduler) use ($channel, $channelMessages) {
 
-        return $this->connection->filter(function (array $event) use ($channel) {
-            return isset($event['channel']) && $event['channel'] == $channel;
-        })->filter(function (array $event) use ($channel) {
-            return $event['event'] !== 'pusher_internal:subscription_succeeded';
+            $subscription = $channelMessages
+                ->filter(function ($msg) {
+                    return $msg->event !== 'pusher_internal:subscription_succeeded';
+                })
+                ->subscribe($observer, $scheduler);
+
+            $this->send(['event' => 'pusher:subscribe', 'data' => ['channel' => $channel]]);
+
+            return new CallbackDisposable(function () use ($channel, $subscription) {
+                $this->send(['event' => 'pusher:unsubscribe', 'data' => ['channel' => $channel]]);
+                $subscription->dispose();
+            });
         });
+
+        return $events->share();
     }
 
     public function send(array $message)
     {
-        $this->client->send(json_encode($message));
+        $this->client
+            ->take(1)
+            ->subscribeCallback(function (MessageSubject $ms) use ($message) {
+                $ms->send(json_encode($message));
+            });
     }
 }