eventsource_stream2/
event_stream.rs

1#[cfg(not(feature = "std"))]
2use alloc::string::{FromUtf8Error, String, ToString};
3
4#[cfg(feature = "std")]
5use std::string::FromUtf8Error;
6
7use crate::event::Event;
8use crate::parser::{is_bom, is_lf, line, RawEventLine};
9use crate::utf8_stream::{Utf8Stream, Utf8StreamError};
10use core::fmt;
11use core::pin::Pin;
12use core::time::Duration;
13use futures_core::stream::Stream;
14use futures_core::task::{Context, Poll};
15use pin_project_lite::pin_project;
16
17#[derive(Default, Debug)]
18struct EventBuilder {
19    event: Event,
20    is_complete: bool,
21}
22
23impl EventBuilder {
24    /// From the HTML spec
25    ///
26    /// -> If the field name is "event"
27    ///    Set the event type buffer to field value.
28    ///
29    /// -> If the field name is "data"
30    ///    Append the field value to the data buffer, then append a single U+000A LINE FEED (LF)
31    ///    character to the data buffer.
32    ///
33    /// -> If the field name is "id"
34    ///    If the field value does not contain U+0000 NULL, then set the last event ID buffer
35    ///    to the field value. Otherwise, ignore the field.
36    ///
37    /// -> If the field name is "retry"
38    ///    If the field value consists of only ASCII digits, then interpret the field value as
39    ///    an integer in base ten, and set the event stream's reconnection time to that integer.
40    ///    Otherwise, ignore the field.
41    ///
42    /// -> Otherwise
43    ///    The field is ignored.
44    fn add(&mut self, line: RawEventLine) {
45        match line {
46            RawEventLine::Field(field, val) => {
47                let val = val.unwrap_or("");
48                match field {
49                    "event" => {
50                        self.event.event = val.to_string();
51                    }
52                    "data" => {
53                        self.event.data.push_str(val);
54                        self.event.data.push('\u{000A}');
55                    }
56                    "id" => {
57                        if !val.contains('\u{0000}') {
58                            self.event.id = val.to_string()
59                        }
60                    }
61                    "retry" => {
62                        if let Ok(val) = val.parse::<u64>() {
63                            self.event.retry = Some(Duration::from_millis(val))
64                        }
65                    }
66                    _ => {}
67                }
68            }
69            RawEventLine::Comment => {}
70            RawEventLine::Empty => self.is_complete = true,
71        }
72    }
73
74    /// From the HTML spec
75    ///
76    /// 1. Set the last event ID string of the event source to the value of the last event ID buffer. The buffer does not get reset, so the last event ID string of the event source remains set to this value until the next time it is set by the server.
77    /// 2. If the data buffer is an empty string, set the data buffer and the event type buffer to the empty string and return.
78    /// 3. If the data buffer's last character is a U+000A LINE FEED (LF) character, then remove the last character from the data buffer.
79    /// 4. Let event be the result of creating an event using MessageEvent, in the relevant Realm of the EventSource object.
80    /// 5. Initialize event's type attribute to message, its data attribute to data, its origin attribute to the serialization of the origin of the event stream's final URL (i.e., the URL after redirects), and its lastEventId attribute to the last event ID string of the event source.
81    /// 6. If the event type buffer has a value other than the empty string, change the type of the newly created event to equal the value of the event type buffer.
82    /// 7. Set the data buffer and the event type buffer to the empty string.
83    /// 8. Queue a task which, if the readyState attribute is set to a value other than CLOSED, dispatches the newly created event at the EventSource object.
84    fn dispatch(&mut self) -> Option<Event> {
85        let builder = core::mem::take(self);
86        let mut event = builder.event;
87        self.event.id = event.id.clone();
88
89        if event.data.is_empty() {
90            return None;
91        }
92
93        if is_lf(event.data.chars().next_back().unwrap()) {
94            event.data.pop();
95        }
96
97        if event.event.is_empty() {
98            event.event = "message".to_string();
99        }
100
101        Some(event)
102    }
103}
104
105#[derive(Debug, Clone, Copy)]
106pub enum EventStreamState {
107    NotStarted,
108    Started,
109    Terminated,
110}
111
112impl EventStreamState {
113    fn is_terminated(self) -> bool {
114        matches!(self, Self::Terminated)
115    }
116    fn is_started(self) -> bool {
117        matches!(self, Self::Started)
118    }
119}
120
121pin_project! {
122    /// A Stream of events
123    pub struct EventStream<S> {
124        #[pin]
125        stream: Utf8Stream<S>,
126        buffer: String,
127        builder: EventBuilder,
128        state: EventStreamState,
129        last_event_id: String,
130        already_scanned: usize,
131    }
132}
133
134impl<S> EventStream<S> {
135    /// Initialize the EventStream with a Stream
136    pub fn new(stream: S) -> Self {
137        Self {
138            stream: Utf8Stream::new(stream),
139            buffer: String::new(),
140            builder: EventBuilder::default(),
141            state: EventStreamState::NotStarted,
142            last_event_id: String::new(),
143            already_scanned: 0,
144        }
145    }
146
147    /// Set the last event ID of the stream. Useful for initializing the stream with a previous
148    /// last event ID
149    pub fn set_last_event_id(&mut self, id: impl Into<String>) {
150        self.last_event_id = id.into();
151    }
152
153    /// Get the last event ID of the stream
154    pub fn last_event_id(&self) -> &str {
155        &self.last_event_id
156    }
157}
158
159/// Error thrown while parsing an event line
160#[derive(Debug, PartialEq)]
161pub enum EventStreamError<E> {
162    /// Source stream is not valid UTF8
163    Utf8(FromUtf8Error),
164    /// Underlying source stream error
165    Transport(E),
166}
167
168impl<E> From<Utf8StreamError<E>> for EventStreamError<E> {
169    fn from(err: Utf8StreamError<E>) -> Self {
170        match err {
171            Utf8StreamError::Utf8(err) => Self::Utf8(err),
172            Utf8StreamError::Transport(err) => Self::Transport(err),
173        }
174    }
175}
176
177impl<E> fmt::Display for EventStreamError<E>
178where
179    E: fmt::Display,
180{
181    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
182        match self {
183            Self::Utf8(err) => f.write_fmt(format_args!("UTF8 error: {}", err)),
184            Self::Transport(err) => f.write_fmt(format_args!("Transport error: {}", err)),
185        }
186    }
187}
188
189#[cfg(feature = "std")]
190impl<E> std::error::Error for EventStreamError<E> where E: fmt::Display + fmt::Debug + Send + Sync {}
191
192fn parse_event(
193    buffer: &mut String,
194    builder: &mut EventBuilder,
195    already_scanned: &mut usize,
196) -> Option<Event> {
197    if buffer.is_empty() {
198        return None;
199    }
200    loop {
201        let (rem, next_line) = line(buffer.as_ref(), *already_scanned)
202            .inspect_err(|resume_from| *already_scanned = *resume_from)
203            .ok()?;
204        builder.add(next_line);
205        let consumed = buffer.len() - rem.len();
206        let rem = buffer.split_off(consumed);
207
208        *already_scanned = 0;
209        *buffer = rem;
210        if builder.is_complete {
211            if let Some(event) = builder.dispatch() {
212                return Some(event);
213            }
214        }
215    }
216}
217
218impl<S, B, E> Stream for EventStream<S>
219where
220    S: Stream<Item = Result<B, E>>,
221    B: AsRef<[u8]>,
222{
223    type Item = Result<Event, EventStreamError<E>>;
224
225    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
226        let mut this = self.project();
227
228        if let Some(event) = parse_event(this.buffer, this.builder, this.already_scanned) {
229            *this.last_event_id = event.id.clone();
230            return Poll::Ready(Some(Ok(event)));
231        }
232
233        if this.state.is_terminated() {
234            return Poll::Ready(None);
235        }
236
237        loop {
238            match this.stream.as_mut().poll_next(cx) {
239                Poll::Ready(Some(Ok(string))) => {
240                    if string.is_empty() {
241                        continue;
242                    }
243
244                    let slice = if this.state.is_started() {
245                        &string
246                    } else {
247                        *this.state = EventStreamState::Started;
248                        if is_bom(string.chars().next().unwrap()) {
249                            &string[3..]
250                        } else {
251                            &string
252                        }
253                    };
254                    this.buffer.push_str(slice);
255
256                    if let Some(event) =
257                        parse_event(this.buffer, this.builder, this.already_scanned)
258                    {
259                        *this.last_event_id = event.id.clone();
260                        return Poll::Ready(Some(Ok(event)));
261                    }
262                }
263                Poll::Ready(Some(Err(err))) => return Poll::Ready(Some(Err(err.into()))),
264                Poll::Ready(None) => {
265                    *this.state = EventStreamState::Terminated;
266                    return Poll::Ready(None);
267                }
268                Poll::Pending => return Poll::Pending,
269            }
270        }
271    }
272}
273
274#[cfg(test)]
275mod tests {
276    use super::*;
277    use futures::prelude::*;
278
279    #[tokio::test]
280    async fn valid_data_fields() {
281        assert_eq!(
282            EventStream::new(futures::stream::iter(vec![Ok::<_, ()>(
283                "data: Hello, world!\n\n"
284            )]))
285            .try_collect::<Vec<_>>()
286            .await
287            .unwrap(),
288            vec![Event {
289                event: "message".to_string(),
290                data: "Hello, world!".to_string(),
291                ..Default::default()
292            }]
293        );
294        assert_eq!(
295            EventStream::new(futures::stream::iter(vec![
296                Ok::<_, ()>("data: Hello,"),
297                Ok::<_, ()>(" world!\n\n")
298            ]))
299            .try_collect::<Vec<_>>()
300            .await
301            .unwrap(),
302            vec![Event {
303                event: "message".to_string(),
304                data: "Hello, world!".to_string(),
305                ..Default::default()
306            }]
307        );
308        assert_eq!(
309            EventStream::new(futures::stream::iter(vec![
310                Ok::<_, ()>("data: Hello,"),
311                Ok::<_, ()>(""),
312                Ok::<_, ()>(" world!\n\n")
313            ]))
314            .try_collect::<Vec<_>>()
315            .await
316            .unwrap(),
317            vec![Event {
318                event: "message".to_string(),
319                data: "Hello, world!".to_string(),
320                ..Default::default()
321            }]
322        );
323        assert_eq!(
324            EventStream::new(futures::stream::iter(vec![Ok::<_, ()>(
325                "data: Hello, world!\n"
326            )]))
327            .try_collect::<Vec<_>>()
328            .await
329            .unwrap(),
330            vec![]
331        );
332        assert_eq!(
333            EventStream::new(futures::stream::iter(vec![Ok::<_, ()>(
334                "data: Hello,\ndata: world!\n\n"
335            )]))
336            .try_collect::<Vec<_>>()
337            .await
338            .unwrap(),
339            vec![Event {
340                event: "message".to_string(),
341                data: "Hello,\nworld!".to_string(),
342                ..Default::default()
343            }]
344        );
345        assert_eq!(
346            EventStream::new(futures::stream::iter(vec![Ok::<_, ()>(
347                "data: Hello,\n\ndata: world!\n\n"
348            )]))
349            .try_collect::<Vec<_>>()
350            .await
351            .unwrap(),
352            vec![
353                Event {
354                    event: "message".to_string(),
355                    data: "Hello,".to_string(),
356                    ..Default::default()
357                },
358                Event {
359                    event: "message".to_string(),
360                    data: "world!".to_string(),
361                    ..Default::default()
362                }
363            ]
364        );
365    }
366
367    #[tokio::test]
368    async fn spec_examples() {
369        assert_eq!(
370            EventStream::new(futures::stream::iter(vec![Ok::<_, ()>(
371                "data: This is the first message.
372
373data: This is the second message, it
374data: has two lines.
375
376data: This is the third message.
377
378"
379            )]))
380            .try_collect::<Vec<_>>()
381            .await
382            .unwrap(),
383            vec![
384                Event {
385                    event: "message".to_string(),
386                    data: "This is the first message.".to_string(),
387                    ..Default::default()
388                },
389                Event {
390                    event: "message".to_string(),
391                    data: "This is the second message, it\nhas two lines.".to_string(),
392                    ..Default::default()
393                },
394                Event {
395                    event: "message".to_string(),
396                    data: "This is the third message.".to_string(),
397                    ..Default::default()
398                }
399            ]
400        );
401        assert_eq!(
402            EventStream::new(futures::stream::iter(vec![Ok::<_, ()>(
403                "event: add
404data: 73857293
405
406event: remove
407data: 2153
408
409event: add
410data: 113411
411
412"
413            )]))
414            .try_collect::<Vec<_>>()
415            .await
416            .unwrap(),
417            vec![
418                Event {
419                    event: "add".to_string(),
420                    data: "73857293".to_string(),
421                    ..Default::default()
422                },
423                Event {
424                    event: "remove".to_string(),
425                    data: "2153".to_string(),
426                    ..Default::default()
427                },
428                Event {
429                    event: "add".to_string(),
430                    data: "113411".to_string(),
431                    ..Default::default()
432                }
433            ]
434        );
435        assert_eq!(
436            EventStream::new(futures::stream::iter(vec![Ok::<_, ()>(
437                "data: YHOO
438data: +2
439data: 10
440
441"
442            )]))
443            .try_collect::<Vec<_>>()
444            .await
445            .unwrap(),
446            vec![Event {
447                event: "message".to_string(),
448                data: "YHOO\n+2\n10".to_string(),
449                ..Default::default()
450            },]
451        );
452        assert_eq!(
453            EventStream::new(futures::stream::iter(vec![Ok::<_, ()>(
454                ": test stream
455
456data: first event
457id: 1
458
459data:second event
460id
461
462data:  third event
463
464"
465            )]))
466            .try_collect::<Vec<_>>()
467            .await
468            .unwrap(),
469            vec![
470                Event {
471                    event: "message".to_string(),
472                    id: "1".to_string(),
473                    data: "first event".to_string(),
474                    ..Default::default()
475                },
476                Event {
477                    event: "message".to_string(),
478                    data: "second event".to_string(),
479                    ..Default::default()
480                },
481                Event {
482                    event: "message".to_string(),
483                    data: " third event".to_string(),
484                    ..Default::default()
485                }
486            ]
487        );
488        assert_eq!(
489            EventStream::new(futures::stream::iter(vec![Ok::<_, ()>(
490                "data
491
492data
493data
494
495data:
496"
497            )]))
498            .try_collect::<Vec<_>>()
499            .await
500            .unwrap(),
501            vec![
502                Event {
503                    event: "message".to_string(),
504                    data: "".to_string(),
505                    ..Default::default()
506                },
507                Event {
508                    event: "message".to_string(),
509                    data: "\n".to_string(),
510                    ..Default::default()
511                },
512            ]
513        );
514        assert_eq!(
515            EventStream::new(futures::stream::iter(vec![Ok::<_, ()>(
516                "data:test
517
518data: test
519
520"
521            )]))
522            .try_collect::<Vec<_>>()
523            .await
524            .unwrap(),
525            vec![
526                Event {
527                    event: "message".to_string(),
528                    data: "test".to_string(),
529                    ..Default::default()
530                },
531                Event {
532                    event: "message".to_string(),
533                    data: "test".to_string(),
534                    ..Default::default()
535                },
536            ]
537        );
538    }
539}
eventsource_stream2/event_stream.rs

eventsource_stream2/
event_stream.rs
