Updated clients with docblocks, made them final, moved certain responsibilities to ApiClients

Author: WyriHaximus

src/AsyncClient.php

@@ -14,32 +14,47 @@
 use function EventLoop\getLoop;
 use function EventLoop\setLoop;
 
-class AsyncClient
+final class AsyncClient
 {
-    protected $transport;
-    protected $app;
-    protected $url;
+    /**
+     * @var Observable\RefCountObservable
+     */
     protected $client;
+
+    /**
+     * @var Observable\AnonymousObservable
+     */
     protected $messages;
+
+    /**
+     * @var array
+     */
     protected $channels = [];
 
+    /**
+     * @param LoopInterface $loop
+     * @param string $app Application ID
+     */
     public function __construct(LoopInterface $loop, string $app)
     {
+        // Set loop into global look accessor
         setLoop($loop);
-        $this->app = $app;
-        $this->url = 'wss://ws.pusherapp.com/app/' .
-            $this->app .
-            '?client=wyrihaximus-php-pusher-client&version=0.0.1&protocol=7'
-        ;
+
         //Only create one connection and share the most recent among all subscriber
-        $this->client   = (new WebsocketClient($this->url))->shareReplay(1);
+        $this->client   = (new WebsocketClient(ApiSettings::createUrl($app)))->shareReplay(1);
         $this->messages = $this->client
             ->flatMap(function (MessageSubject $ms) {
                 return $ms;
             })
             ->map('json_decode');
     }
 
+    /**
+     * Listen on a channel
+     *
+     * @param string $channel Channel to listen on
+     * @return ObservableInterface
+     */
     public function channel(string $channel): ObservableInterface
     {
         if (isset($this->channels[$channel])) {
@@ -75,6 +90,11 @@ public function channel(string $channel): ObservableInterface
         return $this->channels[$channel];
     }
 
+    /**
+     * Send a message through the client
+     *
+     * @param array $message Message to send, will be json encoded
+     */
     public function send(array $message)
     {
         $this->client

src/Client.php

@@ -3,21 +3,37 @@
 namespace ApiClients\Pusher;
 
 use React\EventLoop\Factory;
+use React\EventLoop\LoopInterface;
 use React\Promise\Deferred;
 use Rx\Observer\CallbackObserver;
 use function Clue\React\Block\await;
+use function React\Promise\all;
 
-class Client
+final class Client
 {
+    /**
+     * @var LoopInterface
+     */
     protected $loop;
+
+    /**
+     * @var AsyncClient
+     */
     protected $client;
 
+    /**
+     * @param string $app Application ID
+     */
     public function __construct(string $app)
     {
         $this->loop = Factory::create();
         $this->client = new AsyncClient($this->loop, $app);
     }
 
+    /**
+     * @param string $channel Channel to listen on
+     * @param callable $listener Listener to call on new messages
+     */
     public function channel(string $channel, callable $listener)
     {
         $this->channels(
@@ -28,10 +44,15 @@ public function channel(string $channel, callable $listener)
         );
     }
 
+    /**
+     * @param string[] $channels Channels to listen on
+     * @param callable $listener Listener to call on new messages
+     */
     public function channels(array $channels, callable $listener)
     {
-        $deferred = new Deferred();
+        $promises = [];
         foreach ($channels as $channel) {
+            $deferred = new Deferred();
             $this->client->channel($channel)->subscribe(
                 new CallbackObserver(
                     $listener,
@@ -41,9 +62,11 @@ function () use ($deferred) {
                     }
                 )
             );
+            $promises[] = $deferred->promise();
         }
+
         await(
-            $deferred->promise(),
+            all($promises),
             $this->loop
         );
     }