Skip to content

Commit c8f07a6

Browse files
clueclue
committed
Update EventSource to use SseDecoder internally
1 parent c782fda commit c8f07a6

2 files changed

Lines changed: 148 additions & 21 deletions

File tree

‎src/EventSource.php‎

Lines changed: 11 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -217,29 +217,19 @@ private function request()
217217
$stream = $response->getBody();
218218
assert($stream instanceof ReadableStreamInterface);
219219

220-
$buffer = '';
221-
$stream->on('data', function ($chunk) use (&$buffer, $stream) {
222-
$messageEvents = preg_split(
223-
'/(?:\r\n|\r(?!\n)|\n){2}/S',
224-
$buffer . $chunk
225-
);
226-
$buffer = array_pop($messageEvents);
227-
228-
foreach ($messageEvents as $data) {
229-
$message = MessageEvent::parse($data, $this->lastEventId, $this->reconnectTime);
230-
$this->lastEventId = $message->lastEventId;
231-
232-
if ($message->data !== '') {
233-
$this->emit($message->type, array($message));
234-
if ($this->readyState === self::CLOSED) {
235-
break;
236-
}
237-
}
238-
}
220+
// resume from the last event ID received, it persists across reconnects
221+
$sse = new SseDecoder($stream, $this->lastEventId);
222+
$sse->on('data', function (MessageEvent $message) {
223+
$this->emit($message->type, array($message));
239224
});
240225

241-
$stream->on('close', function () use (&$buffer) {
242-
$buffer = '';
226+
$sse->on('close', function () use ($sse) {
227+
// `id` and `retry` may arrive on events that are never dispatched
228+
$this->lastEventId = $sse->lastEventId;
229+
if ($sse->lastRetryTime !== null) {
230+
$this->reconnectTime = $sse->lastRetryTime;
231+
}
232+
243233
$this->request = null;
244234
if ($this->readyState === self::OPEN) {
245235
$this->readyState = self::CONNECTING;

‎tests/EventSourceTest.php‎

Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -657,6 +657,143 @@ public function testReconnectAfterStreamClosesUsesLastEventIdFromParsedEventStre
657657
$timerReconnect();
658658
}
659659

660+
public function testReconnectTwiceAfterSecondStreamWithoutIdStillUsesLastEventIdFromFirstStream()
661+
{
662+
$loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
663+
$timerReconnect = null;
664+
$loop->expects($this->exactly(2))->method('addTimer')->with(
665+
3.0,
666+
$this->callback(function ($cb) use (&$timerReconnect) {
667+
$timerReconnect = $cb;
668+
return true;
669+
})
670+
);
671+
672+
$first = new Deferred();
673+
$second = new Deferred();
674+
$browser = $this->getMockBuilder('React\Http\Browser')->disableOriginalConstructor()->getMock();
675+
$browser->expects($this->once())->method('withRejectErrorResponse')->willReturnSelf();
676+
$browser->expects($this->exactly(3))->method('requestStreaming')->withConsecutive(
677+
['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache']],
678+
['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache', 'Last-Event-ID' => '123']],
679+
['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache', 'Last-Event-ID' => '123']]
680+
)->willReturnOnConsecutiveCalls(
681+
$first->promise(),
682+
$second->promise(),
683+
new Promise(function () { })
684+
);
685+
686+
$es = new EventSource('http://example.com', $browser, $loop);
687+
688+
$stream = new ThroughStream();
689+
$first->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
690+
691+
$stream->write("id:123\n\n");
692+
$stream->end();
693+
694+
$this->assertNotNull($timerReconnect);
695+
$timerReconnect();
696+
697+
$stream = new ThroughStream();
698+
$second->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
699+
700+
$stream->write("data:hello\n\n");
701+
$stream->end();
702+
703+
$this->assertNotNull($timerReconnect);
704+
$timerReconnect();
705+
}
706+
707+
public function testReconnectTwiceAfterSecondStreamWithEmptyIdClearsLastEventIdForNextRequest()
708+
{
709+
$loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
710+
$timerReconnect = null;
711+
$loop->expects($this->exactly(2))->method('addTimer')->with(
712+
3.0,
713+
$this->callback(function ($cb) use (&$timerReconnect) {
714+
$timerReconnect = $cb;
715+
return true;
716+
})
717+
);
718+
719+
$first = new Deferred();
720+
$second = new Deferred();
721+
$browser = $this->getMockBuilder('React\Http\Browser')->disableOriginalConstructor()->getMock();
722+
$browser->expects($this->once())->method('withRejectErrorResponse')->willReturnSelf();
723+
$browser->expects($this->exactly(3))->method('requestStreaming')->withConsecutive(
724+
['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache']],
725+
['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache', 'Last-Event-ID' => '123']],
726+
['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache']]
727+
)->willReturnOnConsecutiveCalls(
728+
$first->promise(),
729+
$second->promise(),
730+
new Promise(function () { })
731+
);
732+
733+
$es = new EventSource('http://example.com', $browser, $loop);
734+
735+
$stream = new ThroughStream();
736+
$first->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
737+
738+
$stream->write("id:123\n\n");
739+
$stream->end();
740+
741+
$this->assertNotNull($timerReconnect);
742+
$timerReconnect();
743+
744+
$stream = new ThroughStream();
745+
$second->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
746+
747+
$stream->write("id:\ndata:hello\n\n");
748+
$stream->end();
749+
750+
$this->assertNotNull($timerReconnect);
751+
$timerReconnect();
752+
}
753+
754+
public function testReconnectTwiceAfterSecondStreamWithoutRetryStillUsesRetryTimeFromFirstStream()
755+
{
756+
$loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
757+
$timerReconnect = null;
758+
$loop->expects($this->exactly(2))->method('addTimer')->with(
759+
2.543,
760+
$this->callback(function ($cb) use (&$timerReconnect) {
761+
$timerReconnect = $cb;
762+
return true;
763+
})
764+
);
765+
766+
$first = new Deferred();
767+
$second = new Deferred();
768+
$browser = $this->getMockBuilder('React\Http\Browser')->disableOriginalConstructor()->getMock();
769+
$browser->expects($this->once())->method('withRejectErrorResponse')->willReturnSelf();
770+
$browser->expects($this->exactly(3))->method('requestStreaming')->willReturnOnConsecutiveCalls(
771+
$first->promise(),
772+
$second->promise(),
773+
new Promise(function () { })
774+
);
775+
776+
$es = new EventSource('http://example.com', $browser, $loop);
777+
778+
$stream = new ThroughStream();
779+
$first->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
780+
781+
$stream->write("retry:2543\n\n");
782+
$stream->end();
783+
784+
$this->assertNotNull($timerReconnect);
785+
$timerReconnect();
786+
787+
$stream = new ThroughStream();
788+
$second->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
789+
790+
$stream->write("data:hello\n\n");
791+
$stream->end();
792+
793+
$this->assertNotNull($timerReconnect);
794+
$timerReconnect();
795+
}
796+
660797
public function testReconnectAfterStreamClosesUsesSpecifiedRetryTime()
661798
{
662799
$loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();

0 commit comments

Comments
 (0)