actix_web_lab/
sse.rs

1//! Semantic server-sent events (SSE) responder.
2//!
3//! Message data and comments can contain multiple lines. CRLF and CR line endings are converted
4//! to LF. An ID with NUL, CR, or LF, or an event name with CR or LF, causes the response body to
5//! return an encoding error.
6//!
7//! # Examples
8//! ```no_run
9//! use std::{convert::Infallible, time::Duration};
10//!
11//! use actix_web::{Responder, get};
12//! use actix_web_lab::sse;
13//!
14//! #[get("/from-channel")]
15//! async fn from_channel() -> impl Responder {
16//!     let (tx, rx) = tokio::sync::mpsc::channel(10);
17//!
18//!     // note: sender will typically be spawned or handed off somewhere else
19//!     let _ = tx.send(sse::Event::Comment("my comment".into())).await;
20//!     let _ = tx
21//!         .send(sse::Data::new("my data").event("chat_msg").into())
22//!         .await;
23//!
24//!     sse::Sse::from_infallible_receiver(rx).with_retry_duration(Duration::from_secs(10))
25//! }
26//!
27//! #[get("/from-stream")]
28//! async fn from_stream() -> impl Responder {
29//!     let event_stream = futures_util::stream::iter([Ok::<_, Infallible>(sse::Event::Data(
30//!         sse::Data::new("foo"),
31//!     ))]);
32//!
33//!     sse::Sse::from_stream(event_stream).with_keep_alive(Duration::from_secs(5))
34//! }
35//! ```
36//!
37//! Complete usage examples can be found in the examples directory of the source code repo.
38#![doc(
39    alias = "server sent",
40    alias = "server-sent",
41    alias = "server sent events",
42    alias = "server-sent events",
43    alias = "event-stream"
44)]
45
46use std::{
47    pin::Pin,
48    task::{Context, Poll},
49    time::Duration,
50};
51
52use actix_web::{
53    HttpRequest, HttpResponse, Responder,
54    body::{BodySize, BoxBody, MessageBody},
55    http::header::ContentEncoding,
56};
57use bytes::Bytes;
58use bytestring::ByteString;
59use futures_core::Stream;
60use pin_project_lite::pin_project;
61use serde::Serialize;
62use tokio::{
63    sync::mpsc,
64    time::{Interval, interval},
65};
66use tokio_stream::wrappers::ReceiverStream;
67
68use crate::{
69    BoxError,
70    header::{CacheControl, CacheDirective},
71    util::InfallibleStream,
72};
73
74/// Server-sent events data message containing a `data` field and optional `id` and `event` fields.
75///
76/// # Examples
77/// ```
78/// # #[actix_web::main] async fn test() {
79/// use std::convert::Infallible;
80///
81/// use actix_web::body;
82/// use actix_web_lab::sse;
83/// use futures_util::stream;
84/// use serde::Serialize;
85///
86/// #[derive(serde::Serialize)]
87/// struct Foo {
88///     bar: u32,
89/// }
90///
91/// let sse = sse::Sse::from_stream(stream::iter([
92///     Ok::<_, Infallible>(sse::Event::Data(sse::Data::new("foo"))),
93///     Ok::<_, Infallible>(sse::Event::Data(
94///         sse::Data::new_json(Foo { bar: 42 }).unwrap(),
95///     )),
96/// ]));
97///
98/// assert_eq!(
99///     body::to_bytes(sse).await.unwrap(),
100///     "data: foo\n\ndata: {\"bar\":42}\n\n",
101/// );
102/// # }; test();
103/// ```
104#[must_use]
105#[derive(Debug, Clone)]
106pub struct Data {
107    id: Option<ByteString>,
108    event: Option<ByteString>,
109    data: ByteString,
110}
111
112impl Data {
113    /// Constructs a new SSE data message with just the `data` field.
114    ///
115    /// # Examples
116    /// ```
117    /// use actix_web_lab::sse;
118    /// let event = sse::Event::Data(sse::Data::new("foo"));
119    /// ```
120    pub fn new(data: impl Into<ByteString>) -> Self {
121        Self {
122            id: None,
123            event: None,
124            data: data.into(),
125        }
126    }
127
128    /// Constructs a new SSE data message the `data` field set to `data` serialized as JSON.
129    ///
130    /// # Examples
131    /// ```
132    /// use actix_web_lab::sse;
133    ///
134    /// #[derive(serde::Serialize)]
135    /// struct Foo {
136    ///     bar: u32,
137    /// }
138    ///
139    /// let event = sse::Event::Data(sse::Data::new_json(Foo { bar: 42 }).unwrap());
140    /// ```
141    pub fn new_json(data: impl Serialize) -> Result<Self, serde_json::Error> {
142        Ok(Self {
143            id: None,
144            event: None,
145            data: serde_json::to_string(&data)?.into(),
146        })
147    }
148
149    /// Sets `data` field.
150    pub fn set_data(&mut self, data: impl Into<ByteString>) {
151        self.data = data.into();
152    }
153
154    /// Sets `id` field, returning a new data message.
155    pub fn id(mut self, id: impl Into<ByteString>) -> Self {
156        self.id = Some(id.into());
157        self
158    }
159
160    /// Sets `id` field.
161    pub fn set_id(&mut self, id: impl Into<ByteString>) {
162        self.id = Some(id.into());
163    }
164
165    /// Sets `event` name field, returning a new data message.
166    pub fn event(mut self, event: impl Into<ByteString>) -> Self {
167        self.event = Some(event.into());
168        self
169    }
170
171    /// Sets `event` name field.
172    pub fn set_event(&mut self, event: impl Into<ByteString>) {
173        self.event = Some(event.into());
174    }
175}
176
177impl From<Data> for Event {
178    fn from(data: Data) -> Self {
179        Self::Data(data)
180    }
181}
182
183/// Server-sent events message containing one or more fields.
184#[must_use]
185#[derive(Debug, Clone)]
186pub enum Event {
187    /// A `data` message with optional ID and event name.
188    ///
189    /// Data messages looks like this in the response stream.
190    /// ```plain
191    /// event: foo
192    /// id: 42
193    /// data: my data
194    ///
195    /// data: {
196    /// data:   "multiline": "data"
197    /// data: }
198    /// ```
199    Data(Data),
200
201    /// A comment message.
202    ///
203    /// Comments look like this in the response stream.
204    /// ```plain
205    /// : my comment
206    ///
207    /// : another comment
208    /// ```
209    Comment(ByteString),
210}
211
212impl Event {
213    /// Encodes the event in event-stream format.
214    fn into_bytes(self) -> Result<Bytes, BoxError> {
215        let event = match self {
216            Self::Data(Data { id, event, data }) => russe::Event::Message(russe::Message {
217                data,
218                event,
219                id,
220                retry: None,
221            }),
222
223            Self::Comment(comment) => russe::Event::Comment(comment),
224        };
225
226        event
227            .into_bytestring()
228            .map(ByteString::into_bytes)
229            .map_err(Into::into)
230    }
231}
232
233pin_project! {
234    /// Server-sent events (`text/event-stream`) responder.
235    ///
236    /// Constructed using a [Tokio channel](Self::from_receiver) or using your [own
237    /// stream](Self::from_stream).
238    #[must_use]
239    #[derive(Debug)]
240    pub struct Sse<S> {
241        #[pin]
242        stream: S,
243        keep_alive: Option<Interval>,
244        retry_interval: Option<Duration>,
245    }
246}
247
248impl<S, E> Sse<S>
249where
250    S: Stream<Item = Result<Event, E>> + 'static,
251    E: Into<BoxError>,
252{
253    /// Create an SSE response from a stream that yields SSE [Event]s.
254    pub fn from_stream(stream: S) -> Self {
255        Self {
256            stream,
257            keep_alive: None,
258            retry_interval: None,
259        }
260    }
261}
262
263impl<S> Sse<InfallibleStream<S>>
264where
265    S: Stream<Item = Event> + 'static,
266{
267    /// Create an SSE response from an infallible stream that yields SSE [Event]s.
268    pub fn from_infallible_stream(stream: S) -> Self {
269        Sse::from_stream(InfallibleStream::new(stream))
270    }
271}
272
273impl<E> Sse<ReceiverStream<Result<Event, E>>>
274where
275    E: Into<BoxError> + 'static,
276{
277    /// Create an SSE response from a receiver that yields SSE [Event]s.
278    pub fn from_receiver(receiver: mpsc::Receiver<Result<Event, E>>) -> Self {
279        Self::from_stream(ReceiverStream::new(receiver))
280    }
281}
282
283impl Sse<InfallibleStream<ReceiverStream<Event>>> {
284    /// Create an SSE response from a receiver that yields SSE [Event]s.
285    pub fn from_infallible_receiver(receiver: mpsc::Receiver<Event>) -> Self {
286        Self::from_stream(InfallibleStream::new(ReceiverStream::new(receiver)))
287    }
288}
289
290impl<S> Sse<S> {
291    /// Enables "keep-alive" messages to be sent in the event stream after a period of inactivity.
292    ///
293    /// By default, no keep-alive is set up.
294    pub fn with_keep_alive(mut self, keep_alive_period: Duration) -> Self {
295        let mut int = interval(keep_alive_period);
296        int.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
297
298        self.keep_alive = Some(int);
299        self
300    }
301
302    /// Queues the first event to inform the client of a custom retry period.
303    ///
304    /// Browsers default to retry every 3 seconds or so.
305    pub fn with_retry_duration(mut self, retry: Duration) -> Self {
306        self.retry_interval = Some(retry);
307        self
308    }
309}
310
311impl<S, E> Responder for Sse<S>
312where
313    S: Stream<Item = Result<Event, E>> + 'static,
314    E: Into<BoxError>,
315{
316    type Body = BoxBody;
317
318    fn respond_to(self, _req: &HttpRequest) -> HttpResponse<Self::Body> {
319        HttpResponse::Ok()
320            .content_type(russe::MEDIA_TYPE_STR)
321            .insert_header(ContentEncoding::Identity)
322            .insert_header(CacheControl(vec![CacheDirective::NoCache]))
323            .body(self)
324    }
325}
326
327impl<S, E> MessageBody for Sse<S>
328where
329    S: Stream<Item = Result<Event, E>>,
330    E: Into<BoxError>,
331{
332    type Error = BoxError;
333
334    fn size(&self) -> BodySize {
335        BodySize::Stream
336    }
337
338    fn poll_next(
339        self: Pin<&mut Self>,
340        cx: &mut Context<'_>,
341    ) -> Poll<Option<Result<Bytes, Self::Error>>> {
342        let this = self.project();
343
344        if let Some(retry) = this.retry_interval.take() {
345            cx.waker().wake_by_ref();
346            return Poll::Ready(Some(
347                russe::Event::Retry(retry)
348                    .into_bytestring()
349                    .map(ByteString::into_bytes)
350                    .map_err(Into::into),
351            ));
352        }
353
354        if let Poll::Ready(msg) = this.stream.poll_next(cx) {
355            return match msg {
356                Some(Ok(msg)) => Poll::Ready(Some(msg.into_bytes())),
357                Some(Err(err)) => Poll::Ready(Some(Err(err.into()))),
358                None => Poll::Ready(None),
359            };
360        }
361
362        if let Some(keep_alive) = this.keep_alive
363            && keep_alive.poll_tick(cx).is_ready()
364        {
365            return Poll::Ready(Some(
366                russe::Event::Comment("keep-alive".into())
367                    .into_bytestring()
368                    .map(ByteString::into_bytes)
369                    .map_err(Into::into),
370            ));
371        }
372
373        Poll::Pending
374    }
375}
376
377#[cfg(test)]
378mod tests {
379    use std::convert::Infallible;
380
381    use actix_web::{body, test::TestRequest};
382    use futures_util::{FutureExt as _, StreamExt as _, future::poll_fn, stream, task::noop_waker};
383    use tokio::time::sleep;
384
385    use super::*;
386    use crate::assert_response_matches;
387
388    #[actix_web::test]
389    async fn existing_data_api_is_preserved() {
390        let mut data = Data::new("old data").id("old ID").event("old event");
391
392        data.set_data("payload");
393        data.set_id("42");
394        data.set_event("update");
395
396        let json = Data::new_json(serde_json::json!({ "bar": 42 })).unwrap();
397        let events = stream::iter([Event::Data(data), json.into()]);
398        let sse = Sse::from_infallible_stream(events);
399
400        assert_eq!(
401            body::to_bytes(sse).await.unwrap(),
402            "id: 42\nevent: update\ndata: payload\n\ndata: {\"bar\":42}\n\n",
403        );
404    }
405
406    #[actix_web::test]
407    async fn message_fields_are_encoded() {
408        let events = stream::iter([
409            Event::Comment("foo".into()),
410            Data::new("\n").into(),
411            Data::new("foo").id("42").event("bar").into(),
412        ]);
413        let sse = Sse::from_infallible_stream(events);
414
415        assert_eq!(
416            body::to_bytes(sse).await.unwrap(),
417            ": foo\n\ndata: \ndata: \n\nid: 42\nevent: bar\ndata: foo\n\n",
418        );
419    }
420
421    #[actix_web::test]
422    async fn sse_from_external_streams() {
423        let st = stream::empty::<Result<_, Infallible>>();
424        let sse = Sse::from_stream(st);
425        assert_eq!(body::to_bytes(sse).await.unwrap(), "");
426
427        let st = stream::once(async { Ok::<_, Infallible>(Event::Data(Data::new("foo"))) });
428        let sse = Sse::from_stream(st);
429        assert_eq!(body::to_bytes(sse).await.unwrap(), "data: foo\n\n");
430
431        let st = stream::repeat(Ok::<_, Infallible>(Event::Data(Data::new("foo")))).take(2);
432        let sse = Sse::from_stream(st);
433        assert_eq!(
434            body::to_bytes(sse).await.unwrap(),
435            "data: foo\n\ndata: foo\n\n",
436        );
437    }
438
439    #[actix_web::test]
440    async fn normalizes_line_endings() {
441        let events = stream::iter([
442            Event::Data(Data::new("first\r\nsecond\rthird\n")),
443            Event::Comment("first\r\nsecond\rthird\n".into()),
444        ]);
445        let sse = Sse::from_infallible_stream(events);
446
447        assert_eq!(
448            body::to_bytes(sse).await.unwrap(),
449            "data: first\ndata: second\ndata: third\ndata: \n\n: first\n: second\n: third\n: \n\n",
450        );
451    }
452
453    #[actix_web::test]
454    async fn rejects_invalid_message_fields() {
455        for (id, event) in [
456            ("bad\nid", "update"),
457            ("bad\rid", "update"),
458            ("bad\0id", "update"),
459            ("42", "bad\nevent"),
460            ("42", "bad\revent"),
461        ] {
462            let events = stream::iter([Event::Data(Data::new("payload").id(id).event(event))]);
463            let sse = Sse::from_infallible_stream(events);
464
465            let err = body::to_bytes(sse).await.unwrap_err();
466
467            assert!(matches!(
468                err.downcast_ref::<russe::Error>(),
469                Some(russe::Error::Invalid),
470            ));
471        }
472    }
473
474    #[test]
475    fn retry_is_first_msg() {
476        let waker = noop_waker();
477        let mut cx = Context::from_waker(&waker);
478
479        let mut sse = Sse::from_stream(InfallibleStream::new(tokio_stream::empty()))
480            .with_retry_duration(Duration::from_millis(42));
481        match Pin::new(&mut sse).poll_next(&mut cx) {
482            Poll::Ready(Some(Ok(bytes))) => assert_eq!(bytes, "retry: 42\n\n"),
483            res => panic!("poll should return retry message, got {res:?}"),
484        }
485    }
486
487    #[actix_web::test]
488    async fn retry_precedes_stream_events() {
489        let events = stream::iter([Event::Comment("first".into())]);
490        let sse =
491            Sse::from_infallible_stream(events).with_retry_duration(Duration::from_millis(42));
492
493        assert_eq!(
494            body::to_bytes(sse).await.unwrap(),
495            "retry: 42\n\n: first\n\n",
496        );
497    }
498
499    #[actix_web::test]
500    async fn receiver_errors_are_preserved() {
501        let (sender, receiver) = mpsc::channel(1);
502        let sse = Sse::from_receiver(receiver);
503
504        sender
505            .send(Err(std::io::Error::other("stream failed")))
506            .await
507            .unwrap();
508
509        let err = body::to_bytes(sse).await.unwrap_err();
510
511        assert_eq!(
512            err.downcast_ref::<std::io::Error>().unwrap().to_string(),
513            "stream failed",
514        );
515    }
516
517    #[actix_web::test]
518    async fn closed_receiver_ends_with_keep_alive() {
519        let (sender, receiver) = mpsc::channel(1);
520        let sse = Sse::from_infallible_receiver(receiver).with_keep_alive(Duration::from_millis(4));
521
522        drop(sender);
523
524        assert_eq!(body::to_bytes(sse).await.unwrap(), "");
525    }
526
527    #[actix_web::test]
528    async fn appropriate_headers_are_set_on_responder() {
529        let st = stream::empty::<Result<_, Infallible>>();
530        let sse = Sse::from_stream(st);
531
532        let res = sse.respond_to(&TestRequest::default().to_http_request());
533
534        assert_response_matches!(res, OK;
535            "content-type" => "text/event-stream"
536            "content-encoding" => "identity"
537            "cache-control" => "no-cache"
538        );
539    }
540
541    #[actix_web::test]
542    async fn messages_are_received_from_sender() {
543        let (sender, receiver) = tokio::sync::mpsc::channel(2);
544        let mut sse = Sse::from_infallible_receiver(receiver);
545
546        assert!(
547            poll_fn(|cx| Pin::new(&mut sse).poll_next(cx))
548                .now_or_never()
549                .is_none()
550        );
551
552        sender
553            .send(Event::Data(Data::new("bar").event("foo")))
554            .await
555            .unwrap();
556
557        match poll_fn(|cx| Pin::new(&mut sse).poll_next(cx)).now_or_never() {
558            Some(Some(Ok(bytes))) => assert_eq!(bytes, "event: foo\ndata: bar\n\n"),
559            res => panic!("poll should return data message, got {res:?}"),
560        }
561    }
562
563    #[actix_web::test]
564    async fn keep_alive_is_sent() {
565        let waker = noop_waker();
566        let mut cx = Context::from_waker(&waker);
567
568        let (sender, receiver) = tokio::sync::mpsc::channel(2);
569        let mut sse =
570            Sse::from_infallible_receiver(receiver).with_keep_alive(Duration::from_millis(4));
571
572        assert!(Pin::new(&mut sse).poll_next(&mut cx).is_pending());
573
574        sleep(Duration::from_millis(20)).await;
575
576        match Pin::new(&mut sse).poll_next(&mut cx) {
577            Poll::Ready(Some(Ok(bytes))) => assert_eq!(bytes, ": keep-alive\n\n"),
578            res => panic!("poll should return keep-alive message, got {res:?}"),
579        }
580
581        assert!(Pin::new(&mut sse).poll_next(&mut cx).is_pending());
582
583        sender.send(Event::Data(Data::new("foo"))).await.unwrap();
584
585        match Pin::new(&mut sse).poll_next(&mut cx) {
586            Poll::Ready(Some(Ok(bytes))) => assert_eq!(bytes, "data: foo\n\n"),
587            res => panic!("poll should return data message, got {res:?}"),
588        }
589    }
590}
actix_web_lab/sse.rs

actix_web_lab/
sse.rs
