Emit error when response stream closes before retrying connection

Refs https://html.spec.whatwg.org/multipage/server-sent-events.html#reestablish-the-connection

Author: clue

examples/stream.php

@@ -17,6 +17,10 @@
     var_dump($message);
 });
 
+$es->on('open', function () {
+    echo 'open' . PHP_EOL;
+});
+
 $es->on('error', function (Exception $e) use ($es) {
     if ($es->readyState === EventSource::CLOSED) {
         echo 'Permanent error: ' . $e->getMessage() . PHP_EOL;

src/EventSource.php

@@ -155,6 +155,12 @@ private function request()
                 $this->request = null;
                 if ($this->readyState === self::OPEN) {
                     $this->readyState = self::CONNECTING;
+
+                    $this->emit('error', [new \RuntimeException('Stream closed, reconnecting in ' . $this->reconnectTime . ' seconds')]);
+                    if ($this->readyState === self::CLOSED) {
+                        return;
+                    }
+
                     $this->timer = $this->loop->addTimer($this->reconnectTime, function () {
                         $this->timer = null;
                         $this->request();
@@ -170,7 +176,7 @@ private function request()
                 return;
             }
 
-            $this->emit('error', array($e));
+            $this->emit('error', [$e]);
             if ($this->readyState === self::CLOSED) {
                 return;
             }

tests/EventSourceTest.php

@@ -346,7 +346,7 @@ public function testConstructorWillReportOpenWhenGetResponseResolvesWithValidRes
         $this->assertEquals(EventSource::OPEN, $readyState);
     }
 
-    public function testCloseResponseStreamWillStartRetryTimerWithoutErrorEvent()
+    public function testCloseResponseStreamWillStartRetryTimerWithErrorEvent()
     {
         $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
         $loop->expects($this->once())->method('addTimer')->with(
@@ -373,7 +373,33 @@ public function testCloseResponseStreamWillStartRetryTimerWithoutErrorEvent()
         $stream->close();
 
         $this->assertEquals(EventSource::CONNECTING, $es->readyState);
-        $this->assertNull($error);
+        $this->assertInstanceOf('RuntimeException', $error);
+        $this->assertEquals('Stream closed, reconnecting in 3 seconds', $error->getMessage());
+    }
+
+    public function testCloseResponseStreamWillNotStartRetryTimerWhenEventSourceIsClosedFromErrorHandler()
+    {
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $loop->expects($this->never())->method('addTimer');
+
+        $deferred = new Deferred();
+        $browser = $this->getMockBuilder('React\Http\Browser')->disableOriginalConstructor()->getMock();
+        $browser->expects($this->once())->method('withRejectErrorResponse')->willReturnSelf();
+        $browser->expects($this->once())->method('requestStreaming')->willReturn($deferred->promise());
+
+        $es = new EventSource('http://example.com', $loop, $browser);
+
+        $stream = new ThroughStream();
+        $response = new Response(200, array('Content-Type' => 'text/event-stream'), new ReadableBodyStream($stream));
+        $deferred->resolve($response);
+
+        $es->on('error', function ($e) use ($es) {
+            $es->close();
+        });
+
+        $stream->close();
+
+        $this->assertEquals(EventSource::CLOSED, $es->readyState);
     }
 
     public function testCloseFromOpenEventWillCloseResponseStreamAndCloseEventSource()