Removed unnecessary promises. Observables are only converted to promises at the end of the stream.

Author: davidwdan

examples/reddit-single.php

@@ -9,7 +9,7 @@
  * The App ID isn't a secret and comes from a Pusher blog post:
  * @link https://blog.pusher.com/pusher-realtime-reddit-api/
  */
-$client = Client::create(require 'reddit.key.php');
+$client = new Client(require 'reddit.key.php');
 
 $client->channel((string) $argv[1], function (Event $event) {
     echo 'Channel: ', $event->getChannel(), PHP_EOL;

examples/reddit.php

@@ -9,7 +9,7 @@
  * The App ID isn't a secret and comes from a Pusher blog post:
  * @link https://blog.pusher.com/pusher-realtime-reddit-api/
  */
-$client = Client::create(require 'reddit.key.php');
+$client = new Client(require 'reddit.key.php');
 
 $subReddits = $argv;
 array_shift($subReddits);

src/AsyncClient.php

@@ -13,8 +13,6 @@
 use Rx\Websocket\MessageSubject;
 use Rx\Websocket\WebsocketErrorException;
 use Throwable;
-use function React\Promise\reject;
-use function React\Promise\resolve;
 
 final class AsyncClient
 {
@@ -102,17 +100,17 @@ public function __construct(Observable $client)
                 $event = Event::createFromMessage($message);
 
                 if ($event->getEvent() === 'pusher:error') {
-                    return Observable::fromPromise(reject(
-                        new PusherErrorException($event->getData()['message'], $event->getData()['code'])
-                    ));
+                    $throwable = new PusherErrorException($event->getData()['message'], $event->getData()['code']);
+
+                    return Observable::error($throwable);
                 }
 
                 // If this event represents the connection_established event set the timeout
                 if ($event->getEvent() === 'pusher:connection_established') {
                     $this->setActivityTimeout($event);
                 }
 
-                return Observable::fromPromise(resolve($event));
+                return Observable::of($event);
             })
 
             // Handle connection level and Pusher procotol errors
@@ -240,15 +238,15 @@ private function handleLowLevelError(Throwable $throwable)
             !($throwable instanceof RuntimeException) &&
             !($throwable instanceof PusherErrorException)
         ) {
-            return Observable::fromPromise(reject($throwable));
+            return Observable::error($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));
+            return Observable::error($throwable);
         }
 
         // Errors 4100-4199 reconnect after 1 or more seconds, we do it after 1.001 second

src/Client.php

@@ -4,10 +4,8 @@
 
 use React\EventLoop\Factory;
 use React\EventLoop\LoopInterface;
-use React\Promise\Deferred;
-use Throwable;
+use Rx\Observable;
 use function Clue\React\Block\await;
-use function React\Promise\all;
 
 final class Client
 {
@@ -36,12 +34,7 @@ public function __construct(string $app)
      */
     public function channel(string $channel, callable $listener)
     {
-        $this->channels(
-            [
-                $channel,
-            ],
-            $listener
-        );
+        $this->channels([$channel], $listener);
     }
 
     /**
@@ -50,24 +43,12 @@ public function channel(string $channel, callable $listener)
      */
     public function channels(array $channels, callable $listener)
     {
-        $promises = [];
-        foreach ($channels as $channel) {
-            $deferred = new Deferred();
-            $this->client->channel($channel)->subscribe(
-                $listener,
-                function (Throwable $throwable) {
-                    throw $throwable;
-                },
-                function () use ($deferred) {
-                    $deferred->resolve();
-                }
-            );
-            $promises[] = $deferred->promise();
-        }
+        $promise = Observable::fromArray($channels)
+            ->flatMap([$this->client, 'channel'])
+            ->do($listener)
+            ->count()
+            ->toPromise();
 
-        await(
-            all($promises),
-            $this->loop
-        );
+        await($promise, $this->loop);
     }
 }