Added the ability to process streams that use CR or CRLF as newlines

src/EventSource.php

@@ -181,12 +181,13 @@ private function request()
 
             $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);
+                $messageEvents = preg_split(
+                    '/(?:\r\n|\r(?!\n)|\n){2}/S',
+                    $buffer . $chunk
+                );
+                $buffer = array_pop($messageEvents);
 
+                foreach ($messageEvents as $data) {
                     $message = MessageEvent::parse($data);
                     if ($message->lastEventId === null) {
                         $message->lastEventId = $this->lastEventId;
@@ -196,6 +197,9 @@ private function request()
 
                     if ($message->data !== '') {
                         $this->emit($message->type, array($message));
+                        if ($this->readyState === self::CLOSED) {
+                            break;
+                        }
                     }
                 }
             });

src/MessageEvent.php

@@ -11,23 +11,30 @@ class MessageEvent
      */
     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];
+        $message = new self();
+
+        $lines = preg_split(
+            '/\r\n|\r(?!\n)|\n/S',
+            $data
+        );
+        foreach ($lines as $line) {
+            $name = strstr($line, ':', true);
+            $value = substr(strstr($line, ':'), 1);
+            if (isset($value[0]) && $value[0] === ' ') {
+                $value = substr($value, 1);
+            }
+            if ($name === 'data') {
+                $message->data .= $value . "\n";
+            } elseif ($name === 'id') {
+                $message->lastEventId .= $value;
+            } elseif ($name === 'event') {
+                $message->type = $value;
             }
         }
 
         if (substr($message->data, -1) === "\n") {
             $message->data = substr($message->data, 0, -1);
         }
-        //$message->data = rtrim($message->data, "\r\n");
 
         return $message;
     }

tests/EventSourceTest.php

@@ -599,4 +599,117 @@ public function setExpectedException($exception, $exceptionMessage = '', $except
             parent::setExpectedException($exception, $exceptionMessage, $exceptionCode);
         }
     }
+
+    public function testSplitMessagesWithCarriageReturn()
+    {
+        $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', $browser);
+
+        $stream = new ThroughStream();
+        $response = new Response(200, array('Content-Type' => 'text/event-stream'), new ReadableBodyStream($stream));
+        $deferred->resolve($response);
+
+        $messages = [];
+        $es->on('message', function ($m) use (&$messages) {
+            $messages[] = $m;
+        });
+
+        $stream->write("data:hello\r\rdata:hi\r\r");
+
+        $expected = ['hello', 'hi'];
+        $this->assertCount(count($expected), $messages);
+        foreach ($messages as $i => $message) {
+            $this->assertInstanceOf('Clue\React\EventSource\MessageEvent', $message);
+            $this->assertEquals($expected[$i], $message->data);
+        }
+    }
+
+    public function testSplitMessagesWithWindowsEndOfLineSequence()
+    {
+        $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', $browser);
+
+        $stream = new ThroughStream();
+        $response = new Response(200, array('Content-Type' => 'text/event-stream'), new ReadableBodyStream($stream));
+        $deferred->resolve($response);
+
+        $messages = [];
+        $es->on('message', function ($m) use (&$messages) {
+            $messages[] = $m;
+        });
+
+        $stream->write("data:hello\r\n\r\ndata:hi\r\n\r\n");
+
+        $expected = ['hello', 'hi'];
+        $this->assertCount(count($expected), $messages);
+        foreach ($messages as $i => $message) {
+            $this->assertInstanceOf('Clue\React\EventSource\MessageEvent', $message);
+            $this->assertEquals($expected[$i], $message->data);
+        }
+    }
+
+    public function testSplitMessagesWithBufferedWindowsEndOfLineSequence()
+    {
+        $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', $browser);
+
+        $stream = new ThroughStream();
+        $response = new Response(200, array('Content-Type' => 'text/event-stream'), new ReadableBodyStream($stream));
+        $deferred->resolve($response);
+
+        $messages = [];
+        $es->on('message', function ($m) use (&$messages) {
+            $messages[] = $m;
+        });
+
+        $stream->write("data:hello\r\n\r");
+        $stream->write("\ndata:hi\r\n\r\n");
+
+        $expected = ['hello', 'hi'];
+        $this->assertCount(count($expected), $messages);
+        foreach ($messages as $i => $message) {
+            $this->assertInstanceOf('Clue\React\EventSource\MessageEvent', $message);
+            $this->assertEquals($expected[$i], $message->data);
+        }
+    }
+
+    public function testSplitMessagesWithMixedEndOfLine()
+    {
+        $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', $browser);
+
+        $stream = new ThroughStream();
+        $response = new Response(200, array('Content-Type' => 'text/event-stream'), new ReadableBodyStream($stream));
+        $deferred->resolve($response);
+
+        $messages = [];
+        $es->on('message', function ($m) use (&$messages) {
+            $messages[] = $m;
+        });
+
+        $stream->write("data: LF CR\n\rdata: CRLF LF\r\n\ndata: CRLF CR\r\n\rdata: LF CRLF\n\r\ndata: CR CRLF\r\r\n");
+
+        $expected = ['LF CR', 'CRLF LF', 'CRLF CR', 'LF CRLF', 'CR CRLF'];
+        $this->assertCount(count($expected), $messages);
+        foreach ($messages as $i => $message) {
+            $this->assertInstanceOf('Clue\React\EventSource\MessageEvent', $message);
+            $this->assertEquals($expected[$i], $message->data);
+        }
+    }
 }

tests/MessageEventTest.php

@@ -21,10 +21,38 @@ public function testParseDataOverTwoLinesWillBeCombined()
         $this->assertEquals("hello\nworld", $message->data);
     }
 
+    public function testParseDataOverTwoLinesWithCarrigeReturnsWillBeCombinedWithNewline()
+    {
+        $message = MessageEvent::parse("data: hello\rdata: world");
+
+        $this->assertEquals("hello\nworld", $message->data);
+    }
+
+    public function testParseDataOverTwoLinesWithCarrigeReturnsAndNewlinesWillBeCombinedWithNewline()
+    {
+        $message = MessageEvent::parse("data: hello\r\ndata: world");
+
+        $this->assertEquals("hello\nworld", $message->data);
+    }
+
     public function testParseDataWithTrailingNewlineOverTwoLines()
     {
         $message = MessageEvent::parse("data: hello\ndata:");
 
         $this->assertEquals("hello\n", $message->data);
     }
+
+    public function testParseDataWithCarrigeReturnOverTwoLines()
+    {
+        $message = MessageEvent::parse("data: hello\rdata:");
+
+        $this->assertEquals("hello\n", $message->data);
+    }
+
+    public function testParseDataWithCarrigeReturnAndNewlineOverTwoLines()
+    {
+        $message = MessageEvent::parse("data: hello\r\ndata:");
+
+        $this->assertEquals("hello\n", $message->data);
+    }
 }