Implemented event steam

README.md

@@ -51,6 +51,7 @@ $latestBlockHash = $latestBlock->getHash();
 ## More examples
 - [Key management](docs/Example/KeyManagement.md)
 - [Sending a Transfer](docs/Example/SendingTransfer.md)
+- [Event Stream](docs/Example/EventStream.md)
 
 ## API
 ### RpcClient

docs/Example/EventStream.md

@@ -0,0 +1,42 @@
+# Event stream
+
+Create `EventSource` instance by passing node url to constructor
+
+```php
+$nodeUrl = 'http://3.208.91.63:9999';
+$es = new Casper\EventStream\EventSource($nodeUrl);
+```
+
+Use `connect()` method for connecting to the node. The method has two parameters `$eventsType` and `$startFrom`. The first one is mandatory and second is optional. 
+
+Available events type:
+- `EventSource::EVENT_TYPE_MAIN` - listen system events
+- `EventSource::EVENT_TYPE_DEPLOYS` - listen deploy events
+- `EventSource::EVENT_TYPE_SIGS` - listen finality signatures events
+
+```php
+$es->connect(\Casper\EventStream\EventSource::EVENT_TYPE_MAIN);
+```
+
+Set event handler callback function to `EventSource` instance with `onMessage()` method.
+For event stream aborting use `abort()` method.
+
+## Example
+
+```php
+$nodeUrl = 'http://localhost:9999';
+$es = new Casper\EventStream\EventSource($nodeUrl);
+
+$es->onMessage(
+    function (\Casper\EventStream\Event $event) use ($es) {
+        if ($event->getId() === null) {
+            $es->abort();
+            return;
+        }
+
+        echo json_encode($event->getData()) . "\n";
+    }
+);
+
+$es->connect(\Casper\EventStream\EventSource::EVENT_TYPE_MAIN);
+```

src/EventStream/Event.php

@@ -0,0 +1,48 @@
+<?php
+
+namespace Casper\EventStream;
+
+class Event
+{
+    private ?string $id = null;
+
+    private ?string $name = null;
+
+    private ?array $data = null;
+
+    /**
+     * @param string $data
+     * @throws \Exception
+     */
+    public function __construct(string $data)
+    {
+        if (preg_match_all("/data:(?<data>.*)/", $data, $match)) {
+            $this->data = json_decode($match['data'][0], true);
+        }
+
+        if (preg_match_all("/(?<key>id|name)\:(?<value>.*)/", $data, $match)) {
+            foreach ($match['key'] as $i => $key) {
+                $this->{$key} = trim($match['value'][$i]);
+            }
+        }
+
+        if (!$this->data) {
+            throw new \Exception('Invalid event');
+        }
+    }
+
+    public function getId(): ?string
+    {
+        return $this->id;
+    }
+
+    public function getName(): ?string
+    {
+        return $this->name;
+    }
+
+    public function getData(): ?array
+    {
+        return $this->data;
+    }
+}

src/EventStream/EventSource.php

@@ -0,0 +1,76 @@
+<?php
+
+namespace Casper\EventStream;
+
+class EventSource
+{
+    public const EVENT_TYPE_MAIN = 'main';
+    public const EVENT_TYPE_DEPLOYS = 'deploys';
+    public const EVENT_TYPE_SIGS = 'sigs';
+
+    private string $url;
+
+    private $onMessage;
+
+    private bool $aborted = false;
+
+    public function __construct(string $url)
+    {
+        $this->url = $url;
+    }
+
+    public function onMessage(callable $callback): void
+    {
+        $this->onMessage = $callback;
+    }
+
+    public function abort()
+    {
+        $this->aborted = true;
+    }
+
+    /**
+     * @throws \Exception
+     */
+    public function connect(string $eventsType, int $startFrom = 0): void
+    {
+        $this->assertEventTypeIsValid($eventsType);
+
+        $curl = curl_init($this->url . '/events/' . $eventsType);
+        curl_setopt_array($curl, array(
+            CURLOPT_WRITEFUNCTION => function ($_, $data) {
+                try {
+                    $event = new Event(trim($data));
+
+                    if (is_callable($this->onMessage)) {
+                        ($this->onMessage)($event);
+                    }
+                } catch (\Exception $_) {
+                }
+                return strlen($data);
+            },
+            CURLOPT_NOPROGRESS => false,
+            CURLOPT_PROGRESSFUNCTION => function () {
+                return $this->aborted;
+            }
+        ));
+        curl_exec($curl);
+        $error = curl_error($curl);
+
+        if (!$this->aborted && $error) {
+            throw new \Exception($error);
+        }
+
+        curl_close($curl);
+    }
+
+    /**
+     * @throws \Exception
+     */
+    private function assertEventTypeIsValid(string $eventType): void
+    {
+        if (!in_array($eventType, [self::EVENT_TYPE_MAIN, self::EVENT_TYPE_DEPLOYS, self::EVENT_TYPE_SIGS])) {
+            throw new \Exception('Invalid event type');
+        }
+    }
+}