Skip to content

Commit 0879c6e

Browse files
pjfanningpjfanning
authored
Create EventStreamParserCrlfSpec.scala‎ (#1038)
* Create EventStreamParserCrlfSpec.scala‎ * Update EventStreamParserCrlfSpec.scala‎
1 parent 086d094 commit 0879c6e

1 file changed

Lines changed: 239 additions & 0 deletions

File tree

‎http-tests/src/test/scala/org/apache/pekko/http/scaladsl/unmarshalling/sse/EventStreamParserCrlfSpec.scala‎‎

Lines changed: 239 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,239 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.pekko.http
19+
package scaladsl
20+
package unmarshalling
21+
package sse
22+
23+
import org.apache.pekko
24+
import pekko.http.scaladsl.model.sse.ServerSentEvent
25+
import pekko.stream.scaladsl.{ Sink, Source }
26+
import pekko.util.ByteString
27+
28+
import org.scalatest.matchers.should.Matchers
29+
import org.scalatest.wordspec.AsyncWordSpec
30+
31+
/**
32+
* Tests for CRLF line-ending handling via the public [[EventStreamParser]] API.
33+
* Covers issue https://github.com/apache/pekko-http/issues/797.
34+
*/
35+
final class EventStreamParserCrlfSpec extends AsyncWordSpec with Matchers with BaseUnmarshallingSpec {
36+
37+
private val maxLineSize = 1048576
38+
private val maxEventSize = 1048576
39+
40+
"EventStreamParser" when {
41+
42+
"receiving a stream with CRLF line endings" should {
43+
44+
"parse a single event with CRLF-terminated data line" in {
45+
val input = ByteString("data: hello\r\n\r\n")
46+
Source.single(input)
47+
.via(EventStreamParser(maxLineSize, maxEventSize))
48+
.runWith(Sink.seq)
49+
.map(_ shouldBe Vector(ServerSentEvent("hello")))
50+
}
51+
52+
"parse multiple events all using CRLF line endings" in {
53+
val input = ByteString("data: event1\r\n\r\ndata: event2\r\n\r\ndata: event3\r\n\r\n")
54+
Source.single(input)
55+
.via(EventStreamParser(maxLineSize, maxEventSize))
56+
.runWith(Sink.seq)
57+
.map(_ shouldBe Vector(
58+
ServerSentEvent("event1"),
59+
ServerSentEvent("event2"),
60+
ServerSentEvent("event3")))
61+
}
62+
63+
"parse all SSE field types with CRLF line endings" in {
64+
val input = ByteString(
65+
"data: the data\r\n" +
66+
"event: my-event\r\n" +
67+
"id: 99\r\n" +
68+
"retry: 3000\r\n" +
69+
"\r\n")
70+
Source.single(input)
71+
.via(EventStreamParser(maxLineSize, maxEventSize))
72+
.runWith(Sink.seq)
73+
.map(_ shouldBe Vector(
74+
ServerSentEvent("the data", Some("my-event"), Some("99"), Some(3000))))
75+
}
76+
77+
"parse multi-line data fields with CRLF line endings" in {
78+
val input = ByteString(
79+
"data: line1\r\n" +
80+
"data: line2\r\n" +
81+
"data: line3\r\n" +
82+
"\r\n")
83+
Source.single(input)
84+
.via(EventStreamParser(maxLineSize, maxEventSize))
85+
.runWith(Sink.seq)
86+
.map(_ shouldBe Vector(ServerSentEvent("line1\nline2\nline3")))
87+
}
88+
89+
"ignore comment lines with CRLF endings" in {
90+
val input = ByteString(
91+
"data: event1\r\n" +
92+
":this is a comment\r\n" +
93+
"\r\n")
94+
Source.single(input)
95+
.via(EventStreamParser(maxLineSize, maxEventSize))
96+
.runWith(Sink.seq)
97+
.map(_ shouldBe Vector(ServerSentEvent("event1")))
98+
}
99+
100+
"not emit events with no data field when emitEmptyEvents is false" in {
101+
val input = ByteString(
102+
"data: real\r\n" +
103+
"\r\n" +
104+
"\r\n" +
105+
"data: also real\r\n" +
106+
"\r\n")
107+
Source.single(input)
108+
.via(EventStreamParser(maxLineSize, maxEventSize, emitEmptyEvents = false))
109+
.runWith(Sink.seq)
110+
.map(_ shouldBe Vector(
111+
ServerSentEvent("real"),
112+
ServerSentEvent("also real")))
113+
}
114+
115+
"emit empty events (heartbeats) when emitEmptyEvents is true with CRLF" in {
116+
// A heartbeat is a data field with an empty value (i.e. "data: " or "data:"),
117+
// not merely a blank separator line.
118+
val input = ByteString(
119+
"data: before\r\n" +
120+
"\r\n" +
121+
"data: \r\n" +
122+
"\r\n" +
123+
"data: after\r\n" +
124+
"\r\n")
125+
Source.single(input)
126+
.via(EventStreamParser(maxLineSize, maxEventSize, emitEmptyEvents = true))
127+
.runWith(Sink.seq)
128+
.map(_ shouldBe Vector(
129+
ServerSentEvent("before"),
130+
ServerSentEvent.heartbeat,
131+
ServerSentEvent("after")))
132+
}
133+
}
134+
135+
"receiving a stream with CR-only (\\r) line endings" should {
136+
137+
"parse a single event with CR-only line endings" in {
138+
val input = ByteString("data: hello\r\r")
139+
Source.single(input)
140+
.via(EventStreamParser(maxLineSize, maxEventSize))
141+
.runWith(Sink.seq)
142+
.map(_ shouldBe Vector(ServerSentEvent("hello")))
143+
}
144+
145+
"parse multiple events with CR-only line endings" in {
146+
val input = ByteString("data: event1\r\rdata: event2\r\r")
147+
Source.single(input)
148+
.via(EventStreamParser(maxLineSize, maxEventSize))
149+
.runWith(Sink.seq)
150+
.map(_ shouldBe Vector(
151+
ServerSentEvent("event1"),
152+
ServerSentEvent("event2")))
153+
}
154+
155+
"parse all SSE field types with CR-only line endings" in {
156+
val input = ByteString("data: the data\revent: my-event\rid: 42\rretry: 1000\r\r")
157+
Source.single(input)
158+
.via(EventStreamParser(maxLineSize, maxEventSize))
159+
.runWith(Sink.seq)
160+
.map(_ shouldBe Vector(
161+
ServerSentEvent("the data", Some("my-event"), Some("42"), Some(1000))))
162+
}
163+
}
164+
165+
"receiving a stream with mixed line endings" should {
166+
167+
"parse events correctly when line endings vary within the stream" in {
168+
// Mix of LF-only, CR-only, and CRLF
169+
val input = ByteString(
170+
"data: lf-event\n" +
171+
"\n" +
172+
"data: cr-event\r" +
173+
"\r" +
174+
"data: crlf-event\r\n" +
175+
"\r\n")
176+
Source.single(input)
177+
.via(EventStreamParser(maxLineSize, maxEventSize))
178+
.runWith(Sink.seq)
179+
.map(_ shouldBe Vector(
180+
ServerSentEvent("lf-event"),
181+
ServerSentEvent("cr-event"),
182+
ServerSentEvent("crlf-event")))
183+
}
184+
185+
"parse a single event whose fields use different line endings" in {
186+
// Each field line uses a different terminator; the event is terminated by a lone LF
187+
val input = ByteString(
188+
"data: the data\r\n" +
189+
"event: my-event\n" +
190+
"id: 7\r\n" +
191+
"\n")
192+
Source.single(input)
193+
.via(EventStreamParser(maxLineSize, maxEventSize))
194+
.runWith(Sink.seq)
195+
.map(_ shouldBe Vector(
196+
ServerSentEvent("the data", Some("my-event"), Some("7"))))
197+
}
198+
}
199+
200+
"receiving a CRLF stream delivered in multiple small chunks" should {
201+
202+
"parse events correctly when CRLF is split across chunk boundaries" in {
203+
// The \r and \n of the CRLF pair for event1 arrive in separate ByteString chunks
204+
val chunks = Vector(
205+
ByteString("data: event1\r"), // ends with \r
206+
ByteString("\n\r\n"), // \n completes the CRLF; \r\n is the event separator
207+
ByteString("data: event2\r\n\r\n"))
208+
Source(chunks)
209+
.via(EventStreamParser(maxLineSize, maxEventSize))
210+
.runWith(Sink.seq)
211+
.map(_ shouldBe Vector(
212+
ServerSentEvent("event1"),
213+
ServerSentEvent("event2")))
214+
}
215+
216+
"parse events correctly when data arrives byte by byte with CRLF" in {
217+
val bytes = ByteString("data: hello\r\n\r\n")
218+
Source(bytes.map(ByteString(_)))
219+
.via(EventStreamParser(maxLineSize, maxEventSize))
220+
.runWith(Sink.seq)
221+
.map(_ shouldBe Vector(ServerSentEvent("hello")))
222+
}
223+
224+
"reassemble a multi-event CRLF stream delivered in arbitrary chunks" in {
225+
val fullStream =
226+
"data: first\r\nevent: alpha\r\nid: 1\r\n\r\n" +
227+
"data: second\r\nid: 2\r\n\r\n"
228+
// split into 5-byte chunks
229+
val chunks = ByteString(fullStream).grouped(5).map(ByteString(_)).toVector
230+
Source(chunks)
231+
.via(EventStreamParser(maxLineSize, maxEventSize))
232+
.runWith(Sink.seq)
233+
.map(_ shouldBe Vector(
234+
ServerSentEvent("first", Some("alpha"), Some("1")),
235+
ServerSentEvent("second", None, Some("2"))))
236+
}
237+
}
238+
}
239+
}

0 commit comments

Comments
 (0)