eventsource_stream2/
event_stream.rs

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

eventsource_stream2/
event_stream.rs
