Initial commit

.gitignore

@@ -0,0 +1,2 @@
+/vendor/
+/composer.lock

.travis.yml

@@ -0,0 +1,21 @@
+language: php
+
+php:
+  - 5.4
+  - 5.5
+  - 5.6
+  - 7.0
+  - 7.1
+  - 7.2
+  - 7.3
+
+# lock distro so new future defaults will not break the build
+dist: trusty
+
+sudo: false
+
+install:
+  - composer install --no-interaction
+
+script:
+  - vendor/bin/phpunit --coverage-text

LICENSE

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2019 Christian Lück
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is furnished
+to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README.md

@@ -0,0 +1,54 @@
+# clue/reactphp-eventsource [![Build Status](https://travis-ci.org/clue/reactphp-eventsource.svg?branch=master)](https://travis-ci.org/clue/reactphp-eventsource)
+
+Event-driven EventSource client, receiving streaming messages from any HTML5 Server-Sent Events (SSE) server,
+built on top of [ReactPHP](https://reactphp.org/).
+
+> Note: This project is in early alpha stage! Feel free to report any issues you encounter.
+
+**Table of contents**
+
+* [Quickstart example](#quickstart-example)
+* [Install](#install)
+* [Tests](#tests)
+* [License](#license)
+
+## Quickstart example
+
+See the [examples](examples).
+
+## Install
+
+The recommended way to install this library is [through Composer](https://getcomposer.org).
+[New to Composer?](https://getcomposer.org/doc/00-intro.md)
+
+This will install the latest supported version:
+
+```bash
+$ composer require clue/reactphp-eventsource:dev-master
+```
+
+This project aims to run on any platform and thus does not require any PHP
+extensions and supports running on legacy PHP 5.4 through current PHP 7+.
+It's *highly recommended to use PHP 7+* for this project.
+
+## Tests
+
+To run the test suite, you first need to clone this repo and then install all
+dependencies [through Composer](https://getcomposer.org):
+
+```bash
+$ composer install
+```
+
+To run the test suite, go to the project root and run:
+
+```bash
+$ php vendor/bin/phpunit
+```
+
+## License
+
+This project is released under the permissive [MIT license](LICENSE).
+
+> Did you know that I offer custom development services and issuing invoices for
+  sponsorships of releases and for contributions? Contact me (@clue) for details.

composer.json

@@ -0,0 +1,24 @@
+{
+    "name": "clue/reactphp-eventsource",
+    "description": "Event-driven EventSource client, receiving streaming messages from any HTML5 Server-Sent Events (SSE) server, built on top of ReactPHP",
+    "keywords": ["EventSource", "Server-Side Events", "SSE", "event-driven", "ReactPHP", "async"],
+    "homepage": "https://github.com/clue/reactphp-eventsource",
+    "license": "MIT",
+    "authors": [
+        {
+            "name": "Christian Lück",
+            "email": "[email protected]"
+        }
+    ],
+    "autoload": {
+        "psr-4": { "Clue\\React\\EventSource\\": "src/" }
+    },
+    "require": {
+        "php": ">=5.4",
+        "clue/buzz-react": "^2.5",
+        "evenement/evenement": "^3.0 || ^2.0"
+    },
+    "require-dev": {
+        "phpunit/phpunit": "^7.0 || ^6.4 || ^5.7 || ^4.8.35"
+    }
+}

examples/stream.php

@@ -0,0 +1,28 @@
+<?php
+
+use React\EventLoop\Factory;
+use Clue\React\EventSource\EventSource;
+
+require __DIR__ . '/../vendor/autoload.php';
+
+if (!isset($argv[1]) || isset($argv[2])) {
+    exit('Usage error: stream.php <uri>' . PHP_EOL);
+}
+
+$loop = Factory::create();
+$es = new EventSource($argv[1], $loop);
+
+$es->on('message', function ($message) {
+    //$data = json_decode($message->data);
+    var_dump($message);
+});
+
+$es->on('error', function (Exception $e) use ($es) {
+    if ($es->readyState === EventSource::CLOSED) {
+        echo 'Permanent error: ' . $e->getMessage() . PHP_EOL;
+    } else {
+        echo 'Temporary error: ' . $e->getMessage() . PHP_EOL;
+    }
+});
+
+$loop->run();

phpunit.xml.dist

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<phpunit bootstrap="vendor/autoload.php"
+         colors="true"
+         convertErrorsToExceptions="true"
+         convertNoticesToExceptions="true"
+         convertWarningsToExceptions="true"
+>
+    <testsuites>
+        <testsuite name="EventSource Test Suite">
+            <directory>./tests/</directory>
+        </testsuite>
+    </testsuites>
+    <filter>
+        <whitelist>
+            <directory>./src/</directory>
+        </whitelist>
+    </filter>
+</phpunit>

src/EventSource.php

@@ -0,0 +1,167 @@
+<?php
+
+namespace Clue\React\EventSource;
+
+use React\EventLoop\LoopInterface;
+use Psr\Http\Message\ResponseInterface;
+use React\Stream\ReadableStreamInterface;
+use Clue\React\Buzz\Browser;
+use Evenement\EventEmitter;
+use React\Socket\ConnectorInterface;
+
+class EventSource extends EventEmitter
+{
+    /**
+     * @var string (read-only) last event ID received
+     */
+    public $lastEventId = '';
+
+    // ready state
+    const CONNECTING = 0;
+    const OPEN = 1;
+    const CLOSED = 2;
+
+    /**
+     * @var int (read-only)
+     * @see self::CONNECTING
+     * @see self::OPEN
+     * @see self::CLOSED
+     */
+    public $readyState = self::CLOSED;
+
+    /**
+     * @var string (read-only) URL
+     */
+    public $url;
+
+    private $loop;
+    private $browser;
+    private $request;
+    private $timer;
+    private $reconnectTime = 3.0;
+
+    public function __construct($url, LoopInterface $loop, ConnectorInterface $connector = null)
+    {
+        $parts = parse_url($url);
+        if (!isset($parts['scheme'], $parts['host']) || !in_array($parts['scheme'], array('http', 'https'))) {
+            throw new \InvalidArgumentException();
+        }
+
+        $browser = new Browser($loop, $connector);
+        $this->browser = $browser->withOptions(array('streaming' => true, 'obeySuccessCode' => false));
+        $this->loop = $loop;
+        $this->url = $url;
+
+        $this->readyState = self::CONNECTING;
+
+        $this->timer = $loop->addTimer(0, function () {
+            $this->timer = null;
+            $this->send();
+        });
+    }
+
+    private function send()
+    {
+        $headers = array(
+            'Accept' => 'text/event-stream',
+            'Cache-Control' => 'no-cache'
+        );
+        if ($this->lastEventId !== '') {
+            $headers['Last-Event-ID'] = $this->lastEventId;
+        }
+
+        $this->request = $this->browser->get(
+            $this->url,
+            $headers
+        );
+        $this->request->then(function (ResponseInterface $response) {
+            if ($response->getStatusCode() !== 200) {
+                $this->readyState = self::CLOSED;
+                $this->emit('error', array(new \UnexpectedValueException('Unexpected status code')));
+                $this->close();
+                return;
+            }
+
+            if ($response->getHeaderLine('Content-Type') !== 'text/event-stream') {
+                $this->readyState = self::CLOSED;
+                $this->emit('error', array(new \UnexpectedValueException('Unexpected Content-Type')));
+                $this->close();
+                return;
+            }
+
+            $stream = $response->getBody();
+            assert($stream instanceof ReadableStreamInterface);
+
+            $buffer = '';
+            $stream->on('data', function ($chunk) use (&$buffer, $stream) {
+                $buffer .= $chunk;
+
+                while (($pos = strpos($buffer, "\n\n")) !== false) {
+                    $data = substr($buffer, 0, $pos);
+                    $buffer = substr($buffer, $pos + 2);
+
+                    $message = MessageEvent::parse($data);
+                    if ($message->lastEventId === null) {
+                        $message->lastEventId = $this->lastEventId;
+                    } else {
+                        $this->lastEventId = $message->lastEventId;
+                    }
+
+                    if ($message->data !== '') {
+                        $this->emit($message->type, array($message));
+                    }
+                }
+            });
+
+            $stream->on('close', function () {
+                $this->request = null;
+                if ($this->readyState === self::OPEN) {
+                    $this->readyState = self::CONNECTING;
+                    $this->timer = $this->loop->addTimer($this->reconnectTime, function () {
+                        $this->timer = null;
+                        $this->send();
+                    });
+                }
+            });
+
+            $this->readyState = self::OPEN;
+            $this->emit('open');
+        })->then(null, function ($e) {
+            $this->request = null;
+            if ($this->readyState === self::CLOSED) {
+                return;
+            }
+
+            $this->emit('error', array($e));
+            if ($this->readyState === self::CLOSED) {
+                return;
+            }
+
+            $this->timer = $this->loop->addTimer($this->reconnectTime, function () {
+                $this->timer = null;
+                $this->send();
+            });
+        });
+    }
+
+    public function close()
+    {
+        $this->readyState = self::CLOSED;
+        if ($this->request !== null) {
+            $request = $this->request;
+            $this->request = null;
+
+            $request->then(function (ResponseInterface $response) {
+                $response->getBody()->close();
+            });
+            $request->cancel();
+        }
+
+        if ($this->timer !== null) {
+            $this->loop->cancelTimer($this->timer);
+            $this->timer = null;
+        }
+
+        $this->removeAllListeners();
+    }
+}

src/MessageEvent.php

@@ -0,0 +1,49 @@
+<?php
+
+namespace Clue\React\EventSource;
+
+class MessageEvent
+{
+    /**
+     * @param string $data
+     * @return self
+     * @internal
+     */
+    public static function parse($data)
+    {
+        $message = new MessageEvent();
+
+        preg_match_all('/^([a-z]*)\: ?(.*)/m', $data, $matches, PREG_SET_ORDER);
+        foreach ($matches as $match) {
+            if ($match[1] === 'data') {
+                $message->data .= $match[2] . "\n";
+            } elseif ($match[1] === 'id') {
+                $message->lastEventId .= $match[2];
+            } elseif ($match[1] === 'event') {
+                $message->type = $match[2];
+            }
+        }
+
+        if (substr($message->data, -1) === "\n") {
+            $message->data = substr($message->data, 0, -1);
+        }
+        //$message->data = rtrim($message->data, "\r\n");
+
+        return $message;
+    }
+
+    /**
+     * @var string
+     */
+    public $data = '';
+
+    /**
+     * @var ?string
+     */
+    public $lastEventId = null;
+
+    /**
+     * @var string
+     */
+    public $type = 'message';
+}