Skip to content

Commit dcd4770

Browse files
boenrobotboenrobot
committed
Added the ability to process streams that use CR or CRLF as newlines instead of LF.
1 parent 7fc5855 commit dcd4770

4 files changed

Lines changed: 143 additions & 20 deletions

File tree

‎src/EventSource.php‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -181,12 +181,13 @@ private function request()
181181

182182
$buffer = '';
183183
$stream->on('data', function ($chunk) use (&$buffer, $stream) {
184-
$buffer .= $chunk;
185-
186-
while (($pos = strpos($buffer, "\n\n")) !== false) {
187-
$data = substr($buffer, 0, $pos);
188-
$buffer = substr($buffer, $pos + 2);
184+
$messageEvents = preg_split(
185+
'/(?: \r\n | \r(?!\n) | \n ){2}/x',
186+
$buffer . $chunk
187+
);
188+
$buffer = array_pop($messageEvents);
189189

190+
foreach ($messageEvents as $data) {
190191
$message = MessageEvent::parse($data);
191192
if ($message->lastEventId === null) {
192193
$message->lastEventId = $this->lastEventId;
@@ -196,6 +197,9 @@ private function request()
196197

197198
if ($message->data !== '') {
198199
$this->emit($message->type, array($message));
200+
if ($this->readyState === self::CLOSED) {
201+
break;
202+
}
199203
}
200204
}
201205
});

‎src/MessageEvent.php‎

Lines changed: 22 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -11,24 +11,31 @@ class MessageEvent
1111
*/
1212
public static function parse($data)
1313
{
14-
$message = new MessageEvent();
15-
16-
preg_match_all('/^([a-z]*)\: ?(.*)/m', $data, $matches, PREG_SET_ORDER);
17-
foreach ($matches as $match) {
18-
if ($match[1] === 'data') {
19-
$message->data .= $match[2] . "\n";
20-
} elseif ($match[1] === 'id') {
21-
$message->lastEventId .= $match[2];
22-
} elseif ($match[1] === 'event') {
23-
$message->type = $match[2];
14+
$message = new self();
15+
16+
$linesAndEOLs = preg_split(
17+
'/( \r\n | \r(?!\n) | \n )/x',
18+
$data,
19+
0,
20+
PREG_SPLIT_DELIM_CAPTURE
21+
);
22+
$prevEOL = '';
23+
for ($i = 0, $l = 1 + count($linesAndEOLs); $i < $l; $i += 2) {
24+
$name = strstr($linesAndEOLs[$i], ':', true);
25+
$value = substr(strstr($linesAndEOLs[$i], ':'), 1);
26+
if (isset($value[0]) && $value[0] === ' ') {
27+
$value = substr($value, 1);
28+
}
29+
if ($name === 'data') {
30+
$message->data .= $prevEOL . $value;
31+
$prevEOL = isset($linesAndEOLs[$i+1]) ? "\n" : '';
32+
} elseif ($name === 'id') {
33+
$message->lastEventId .= $value;
34+
} elseif ($name === 'event') {
35+
$message->type = $value;
2436
}
2537
}
2638

27-
if (substr($message->data, -1) === "\n") {
28-
$message->data = substr($message->data, 0, -1);
29-
}
30-
//$message->data = rtrim($message->data, "\r\n");
31-
3239
return $message;
3340
}
3441

‎tests/EventSourceTest.php‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -599,4 +599,88 @@ public function setExpectedException($exception, $exceptionMessage = '', $except
599599
parent::setExpectedException($exception, $exceptionMessage, $exceptionCode);
600600
}
601601
}
602+
603+
public function testSplitMessagesWithCarriageReturn()
604+
{
605+
$deferred = new Deferred();
606+
$browser = $this->getMockBuilder('React\Http\Browser')->disableOriginalConstructor()->getMock();
607+
$browser->expects($this->once())->method('withRejectErrorResponse')->willReturnSelf();
608+
$browser->expects($this->once())->method('requestStreaming')->willReturn($deferred->promise());
609+
610+
$es = new EventSource('http://example.com', $browser);
611+
612+
$stream = new ThroughStream();
613+
$response = new Response(200, array('Content-Type' => 'text/event-stream'), new ReadableBodyStream($stream));
614+
$deferred->resolve($response);
615+
616+
$messages = [];
617+
$es->on('message', function ($m) use (&$messages) {
618+
$messages[] = $m;
619+
});
620+
621+
$stream->write("data:hello\r\rdata:hi\r\r");
622+
623+
$expected = ['hello', 'hi'];
624+
$this->assertCount(count($expected), $messages);
625+
foreach ($messages as $i => $message) {
626+
$this->assertInstanceOf('Clue\React\EventSource\MessageEvent', $message);
627+
$this->assertEquals($expected[$i], $message->data);
628+
}
629+
}
630+
631+
public function testSplitMessagesWithWindowsEndOfLineSequence()
632+
{
633+
$deferred = new Deferred();
634+
$browser = $this->getMockBuilder('React\Http\Browser')->disableOriginalConstructor()->getMock();
635+
$browser->expects($this->once())->method('withRejectErrorResponse')->willReturnSelf();
636+
$browser->expects($this->once())->method('requestStreaming')->willReturn($deferred->promise());
637+
638+
$es = new EventSource('http://example.com', $browser);
639+
640+
$stream = new ThroughStream();
641+
$response = new Response(200, array('Content-Type' => 'text/event-stream'), new ReadableBodyStream($stream));
642+
$deferred->resolve($response);
643+
644+
$messages = [];
645+
$es->on('message', function ($m) use (&$messages) {
646+
$messages[] = $m;
647+
});
648+
649+
$stream->write("data:hello\r\n\r\ndata:hi\r\n\r\n");
650+
651+
$expected = ['hello', 'hi'];
652+
$this->assertCount(count($expected), $messages);
653+
foreach ($messages as $i => $message) {
654+
$this->assertInstanceOf('Clue\React\EventSource\MessageEvent', $message);
655+
$this->assertEquals($expected[$i], $message->data);
656+
}
657+
}
658+
659+
public function testSplitMessagesWithMixedEndOfLine()
660+
{
661+
$deferred = new Deferred();
662+
$browser = $this->getMockBuilder('React\Http\Browser')->disableOriginalConstructor()->getMock();
663+
$browser->expects($this->once())->method('withRejectErrorResponse')->willReturnSelf();
664+
$browser->expects($this->once())->method('requestStreaming')->willReturn($deferred->promise());
665+
666+
$es = new EventSource('http://example.com', $browser);
667+
668+
$stream = new ThroughStream();
669+
$response = new Response(200, array('Content-Type' => 'text/event-stream'), new ReadableBodyStream($stream));
670+
$deferred->resolve($response);
671+
672+
$messages = [];
673+
$es->on('message', function ($m) use (&$messages) {
674+
$messages[] = $m;
675+
});
676+
677+
$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");
678+
679+
$expected = ['LF CR', 'CRLF LF', 'CRLF CR', 'LF CRLF', 'CR CRLF'];
680+
$this->assertCount(count($expected), $messages);
681+
foreach ($messages as $i => $message) {
682+
$this->assertInstanceOf('Clue\React\EventSource\MessageEvent', $message);
683+
$this->assertEquals($expected[$i], $message->data);
684+
}
685+
}
602686
}

‎tests/MessageEventTest.php‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,38 @@ public function testParseDataOverTwoLinesWillBeCombined()
2121
$this->assertEquals("hello\nworld", $message->data);
2222
}
2323

24+
public function testParseDataOverTwoLinesWithCarrigeReturnsWillBeCombinedWithNewline()
25+
{
26+
$message = MessageEvent::parse("data: hello\rdata: world");
27+
28+
$this->assertEquals("hello\nworld", $message->data);
29+
}
30+
31+
public function testParseDataOverTwoLinesWithCarrigeReturnsAndNewlinesWillBeCombinedWithNewline()
32+
{
33+
$message = MessageEvent::parse("data: hello\r\ndata: world");
34+
35+
$this->assertEquals("hello\nworld", $message->data);
36+
}
37+
2438
public function testParseDataWithTrailingNewlineOverTwoLines()
2539
{
2640
$message = MessageEvent::parse("data: hello\ndata:");
2741

2842
$this->assertEquals("hello\n", $message->data);
2943
}
44+
45+
public function testParseDataWithCarrigeReturnOverTwoLines()
46+
{
47+
$message = MessageEvent::parse("data: hello\rdata:");
48+
49+
$this->assertEquals("hello\n", $message->data);
50+
}
51+
52+
public function testParseDataWithCarrigeReturnAndNewlineOverTwoLines()
53+
{
54+
$message = MessageEvent::parse("data: hello\r\ndata:");
55+
56+
$this->assertEquals("hello\n", $message->data);
57+
}
3058
}

0 commit comments

Comments
 (0)