All streamed messages are now events

Author: WyriHaximus

composer.json

@@ -2,6 +2,8 @@
   "name": "api-clients/pusher",
   "homepage": "https://php-api-clients.org/clients/pusher",
   "license": "MIT",
+  "minimum-stability": "dev",
+  "prefer-stable": true,
   "authors": [
     {
       "name": "Cees-Jan Kiewiet",
@@ -11,6 +13,7 @@
   "require": {
     "php": "^7.0",
     "api-clients/command-bus": "^2.0",
+    "api-clients/rx-operators": "dev-master",
     "api-clients/service": "dev-master",
     "clue/block-react": "^1.1",
     "ocramius/package-versions": "^1.1",

composer.lock

@@ -1,5 +1,6 @@
 <?php
 
+use ApiClients\Client\Pusher\Event;
 use React\EventLoop\Factory;
 use Rx\Observer\CallbackObserver;
 use ApiClients\Client\Pusher\AsyncClient;
@@ -17,10 +18,10 @@
 $subReddits = $argv;
 array_shift($subReddits);
 foreach ($subReddits as $subReddit) {
-    $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;
+    $client->channel($subReddit)->subscribe(new CallbackObserver(function (Event $event) {
+        echo 'Channel: ', $event->getChannel(), PHP_EOL;
+        echo 'Event: ', $event->getEvent(), PHP_EOL;
+        echo 'Data: ', json_encode($event->getData()), PHP_EOL;
     }));
 }
 

examples/reddit-async.php

@@ -1,6 +1,7 @@
 <?php declare(strict_types=1);
 
 use ApiClients\Client\Pusher\Client;
+use ApiClients\Client\Pusher\Event;
 
 require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
 
@@ -10,8 +11,8 @@
  */
 $client = new Client(require 'reddit.key.php');
 
-$client->channel((string) $argv[1], function ($event) {
-    echo 'Channel: ', $event->channel, PHP_EOL;
-    echo 'Event: ', $event->event, PHP_EOL;
-    echo 'Data: ', $event->data, PHP_EOL;
+$client->channel((string) $argv[1], function (Event $event) {
+    echo 'Channel: ', $event->getChannel(), PHP_EOL;
+    echo 'Event: ', $event->getEvent(), PHP_EOL;
+    echo 'Data: ', json_encode($event->getData()), PHP_EOL;
 });

examples/reddit-single.php

@@ -1,6 +1,7 @@
 <?php declare(strict_types=1);
 
 use ApiClients\Client\Pusher\Client;
+use ApiClients\Client\Pusher\Event;
 
 require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
 
@@ -12,8 +13,8 @@
 
 $subReddits = $argv;
 array_shift($subReddits);
-$client->channels($subReddits, function ($event) {
-    echo 'Channel: ', $event->channel, PHP_EOL;
-    echo 'Event: ', $event->event, PHP_EOL;
-    echo 'Data: ', $event->data, PHP_EOL;
+$client->channels($subReddits, function (Event $event) {
+    echo 'Channel: ', $event->getChannel(), PHP_EOL;
+    echo 'Event: ', $event->getEvent(), PHP_EOL;
+    echo 'Data: ', json_encode($event->getData()), PHP_EOL;
 });

examples/reddit.php

@@ -46,7 +46,10 @@ public function __construct(LoopInterface $loop, string $app)
             ->flatMap(function (MessageSubject $ms) {
                 return $ms;
             })
-            ->map('json_decode');
+            ->_ApiClients_jsonDecode()
+            ->map(function (array $message) {
+                return Event::createFromMessage($message);
+            });
     }
 
     /**
@@ -61,8 +64,8 @@ public function channel(string $channel): Observable
             return $this->channels[$channel];
         }
 
-        $channelMessages = $this->messages->filter(function ($event) use ($channel) {
-            return isset($event->channel) && $event->channel == $channel;
+        $channelMessages = $this->messages->filter(function (Event $event) use ($channel) {
+            return $event->getChannel() !== '' && $event->getChannel() === $channel;
         });
 
         $events = Observable::create(function (
@@ -73,8 +76,8 @@ public function channel(string $channel): Observable
             $channelMessages
         ) {
             $subscription = $channelMessages
-                ->filter(function ($msg) {
-                    return $msg->event !== 'pusher_internal:subscription_succeeded';
+                ->filter(function (Event $event) {
+                    return $event->getEvent() !== 'pusher_internal:subscription_succeeded';
                 })
                 ->subscribe($observer, $scheduler);
 

src/AsyncClient.php

@@ -0,0 +1,66 @@
+<?php declare(strict_types=1);
+
+namespace ApiClients\Client\Pusher;
+
+final class Event
+{
+    /**
+     * @var string
+     */
+    private $event;
+
+    /**
+     * @var string
+     */
+    private $channel;
+
+    /**
+     * @var array
+     */
+    private $data;
+
+    public static function createFromMessage(array $message): self
+    {
+        return new self(
+            $message['event'],
+            json_decode($message['data'], true),
+            isset($message['channel']) ? $message['channel'] : ''
+        );
+    }
+
+    /**
+     * @param string $event
+     * @param array $data
+     * @param string $channel
+     */
+    public function __construct(string $event, array $data, string $channel = '')
+    {
+        $this->event = $event;
+        $this->data = $data;
+        $this->channel = $channel;
+    }
+
+    /**
+     * @return string
+     */
+    public function getEvent()
+    {
+        return $this->event;
+    }
+
+    /**
+     * @return string
+     */
+    public function getChannel()
+    {
+        return $this->channel;
+    }
+
+    /**
+     * @return array
+     */
+    public function getData()
+    {
+        return $this->data;
+    }
+}