surf_sse/lib.rs
1//! An implementation of the [EventSource][] API using [Surf][].
2//!
3//! # Surf Cargo Features
4//! `surf-sse` wraps the [Surf][] library. By default, [`surf-sse`][surf-sse] uses the default
5//! [Surf][] features. If you are using a non-default Surf client implementation or feature set,
6//! you can disable [`surf-sse`][surf-sse]'s default feature:
7//! ```toml
8//! surf = { version = "*", features = ["hyper-client"] }
9//! surf-sse = { version = "*", default-features = false }
10//! ```
11//!
12//! This way, [`surf-sse`][surf-sse] will also use the "hyper-client" feature.
13//!
14//! # Logging
15//!
16//! [`surf-sse`][surf-sse] uses the [`log`][log] crate for some rudimentary connection logging. If you need to debug
17//! an EventSource connection, enable trace logging for the `surf-sse` target. For example, with
18//! [`env_logger`][env_logger]:
19//! ```bash
20//! RUST_LOG=surf-sse=trace \
21//! cargo run
22//! ```
23//!
24//! # Examples
25//! ```rust,no_run
26//! # async_std::task::block_on(async move {
27//! #
28//! use futures_util::stream::TryStreamExt; // for try_next()
29//! use surf_sse::EventSource;
30//!
31//! let mut events = EventSource::new("https://announce.u-wave.net/events".parse().unwrap());
32//!
33//! while let Some(message) = events.try_next().await.unwrap() {
34//! dbg!(message);
35//! }
36//! #
37//! # });
38//! ```
39//!
40//! [EventSource]: https://developer.mozilla.org/en-US/docs/Web/API/EventSource
41//! [Surf]: https://github.com/http-rs/surf
42//! [surf-sse]: https://github.com/goto-bus-stop/surf-sse
43//! [log]: https://docs.rs/log
44//! [env_logger]: https://docs.rs/env_logger
45
46#![deny(future_incompatible)]
47#![deny(nonstandard_style)]
48#![deny(rust_2018_idioms)]
49#![deny(unsafe_code)]
50#![warn(missing_docs)]
51#![warn(unused)]
52
53use futures_core::stream::Stream;
54use futures_timer::Delay;
55use log::trace;
56use std::fmt;
57use std::future::Future;
58use std::pin::Pin;
59use std::task::{Context, Poll};
60use std::time::Duration;
61pub use surf::Url;
62
63/// An event.
64#[derive(Debug, Clone, PartialEq, Eq)]
65pub struct Event {
66 /// The event name, defaults to "message".
67 pub event: String,
68 /// The event data as a UTF-8 String.
69 pub data: String,
70}
71
72/// The state of the connection.
73///
74/// Unlike browser implementations of the EventSource API, this does not have a `Closed` value, because
75/// the stream is closed by dropping the struct. At that point, there's no way to inspect the state
76/// anyway.
77///
78/// A `Closed` value may be added in the future when there is more robust error handling in place.
79#[derive(Debug, Clone, Copy, PartialEq, Eq)]
80pub enum ReadyState {
81 /// The client is connecting. It may have sent a request already, or be waiting for a retry
82 /// timer.
83 Connecting = 0,
84 /// The connection is open and ready to read messages.
85 Open = 1,
86}
87
88/// Represents an EventSource "error" event.
89///
90/// Note that you should not always stop reading messages when an error occurs. Many errors are
91/// benign. EventSources retry a lot, emitting errors on every failure.
92#[derive(Debug)]
93pub enum Error {
94 /// The connection was closed by the server. EventSource will reopen the connection.
95 Retry,
96 /// An error occurred while connecting to the endpoint. EventSource will retry the connection.
97 ConnectionError(surf::Error),
98}
99
100impl fmt::Display for Error {
101 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
102 match self {
103 Self::Retry => write!(f, "the connection was closed by the server, retrying."),
104 Self::ConnectionError(inner) => inner.fmt(f),
105 }
106 }
107}
108
109impl std::error::Error for Error {}
110
111/// Wrapper for a dynamic Future type that adds an opaque Debug implementation.
112struct DynDebugFuture<T>(Pin<Box<dyn Future<Output = T>>>);
113impl<T> Future for DynDebugFuture<T> {
114 type Output = T;
115
116 fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
117 Pin::new(&mut self.0).poll(cx)
118 }
119}
120
121impl<T> fmt::Debug for DynDebugFuture<T> {
122 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
123 write!(f, "[opaque Future type]")
124 }
125}
126
127type EventStream = sse_codec::DecodeStream<surf::Response>;
128type ConnectionFuture = DynDebugFuture<Result<surf::Response, surf::Error>>;
129
130/// Represents the internal state machine.
131#[derive(Debug)]
132#[allow(clippy::large_enum_variant)] // `EventStream` is large but it's the active one most of the time
133enum ConnectState {
134 /// We're receiving messages.
135 Streaming(EventStream),
136 /// We're connecting to the endpoint.
137 Connecting(ConnectionFuture),
138 /// We're waiting to retry.
139 WaitingToRetry(Delay),
140 /// We're not doing anything. Currently only used as a default value. This can be used for the Closed state later.
141 Idle,
142}
143
144/// A Server-Sent Events/Event Sourcing client, similar to [`EventSource`][EventSource] in the browser.
145///
146/// [EventSource]: https://developer.mozilla.org/en-US/docs/Web/API/EventSource
147#[derive(Debug)]
148pub struct EventSource {
149 client: surf::Client,
150 url: Url,
151 retry_time: Duration,
152 last_event_id: Option<String>,
153 state: ConnectState,
154}
155
156impl EventSource {
157 /// Create a new connection.
158 ///
159 /// This constructor creates a new [`surf::Client`][] for the event sourcing connection. If you
160 /// already have a [`surf::Client`][], consider using [`EventSource::with_client`][].
161 ///
162 /// [`surf::Client`]: https://docs.rs/surf/2.x/surf/struct.Client.html
163 /// [`EventSource::with_client`]: #method.with_client
164 pub fn new(url: Url) -> Self {
165 Self::with_client(surf::client(), url)
166 }
167
168 /// Create a new connection.
169 pub fn with_client(client: surf::Client, url: Url) -> Self {
170 let mut event_source = Self {
171 client,
172 url,
173 retry_time: Duration::from_secs(3),
174 last_event_id: None,
175 state: ConnectState::Idle,
176 };
177 event_source.start_connect();
178 event_source
179 }
180
181 /// Get the URL that this client connects to.
182 pub fn url(&self) -> &Url {
183 &self.url
184 }
185
186 /// Get the state of the connection. See the documentation for `ReadyState`.
187 pub fn ready_state(&self) -> ReadyState {
188 match self.state {
189 ConnectState::Streaming(_) => ReadyState::Open,
190 ConnectState::Connecting(_) | ConnectState::WaitingToRetry(_) => ReadyState::Connecting,
191 ConnectState::Idle => unreachable!("ReadyState::Closed"),
192 }
193 }
194
195 /// Get the current retry timeout. If the connection dies, it is reopened after this time.
196 pub fn retry_time(&self) -> Duration {
197 self.retry_time
198 }
199
200 /// Override the retry timeout. If the connection dies, it is reopened after this time. Note
201 /// that the timeout will be reset again if the server sends a new `retry:` message.
202 pub fn set_retry_time(&mut self, duration: Duration) {
203 self.retry_time = duration;
204 }
205
206 fn start_connect(&mut self) {
207 trace!(target: "surf-sse", "connecting to {}", self.url);
208 let mut request = surf::get(self.url.clone()).header("Accept", "text/event-stream");
209 // If the EventSource object's last event ID string is not the empty string, set `Last-Event-ID`/last event ID string, encoded as UTF-8, in request's header list.
210 if let Some(id) = &self.last_event_id {
211 request = request.header("Last-Event-ID", id.as_str());
212 }
213
214 let client = self.client.clone();
215 let request_future = Box::pin(async move { client.send(request).await });
216
217 self.state = ConnectState::Connecting(DynDebugFuture(request_future));
218 }
219
220 fn start_retry(&mut self) {
221 trace!(target: "surf-sse", "connection to {}, retrying in {:?}", self.url, self.retry_time);
222 self.state = ConnectState::WaitingToRetry(Delay::new(self.retry_time));
223 }
224
225 fn start_receiving(&mut self, response: surf::Response) {
226 trace!(target: "surf-sse", "connected to {}, now waiting for events", self.url);
227 self.state = ConnectState::Streaming(sse_codec::decode_stream(response));
228 }
229}
230
231impl Stream for EventSource {
232 type Item = Result<Event, Error>;
233
234 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
235 match &mut self.state {
236 ConnectState::Streaming(event_stream) => {
237 match Pin::new(event_stream).poll_next(cx) {
238 Poll::Pending => Poll::Pending,
239 Poll::Ready(Some(Ok(event))) => match event {
240 sse_codec::Event::Message { id, event, data } => {
241 self.last_event_id = id;
242 Poll::Ready(Some(Ok(Event { event, data })))
243 }
244 sse_codec::Event::Retry { retry } => {
245 self.retry_time = Duration::from_millis(retry);
246 Poll::Pending
247 }
248 },
249 Poll::Ready(Some(Err(_))) => {
250 // we care even less about "incorrect" messages than sse_codec does!
251 Poll::Pending
252 }
253 // Clients will reconnect if the connection is closed.
254 Poll::Ready(None) => {
255 self.start_retry();
256 let error = Error::Retry;
257 Poll::Ready(Some(Err(error)))
258 }
259 }
260 }
261
262 ConnectState::WaitingToRetry(timer) => {
263 match Pin::new(timer).poll(cx) {
264 Poll::Pending => (),
265 Poll::Ready(()) => self.start_connect(),
266 }
267 Poll::Pending
268 }
269
270 ConnectState::Connecting(connecting) => {
271 match Pin::new(connecting).poll(cx) {
272 Poll::Pending => Poll::Pending,
273 // A client can be told to stop reconnecting using the HTTP 204 No Content response code.
274 Poll::Ready(Ok(response)) if response.status() == 204 => Poll::Ready(None),
275 Poll::Ready(Ok(response)) => {
276 self.start_receiving(response);
277 self.poll_next(cx)
278 }
279 Poll::Ready(Err(error)) => {
280 self.start_retry();
281 let error = Error::ConnectionError(error);
282 Poll::Ready(Some(Err(error)))
283 }
284 }
285 }
286
287 ConnectState::Idle => unreachable!(),
288 }
289 }
290}
291
292/// Extension trait with event sourcing methods for Surf clients.
293///
294/// ```rust,no_run
295/// use surf_sse::ClientExt;
296/// use futures_util::stream::StreamExt; // for `.next`
297///
298/// let client = surf::client();
299/// let mut events = client.connect_event_source("https://announce.u-wave.net".parse().unwrap());
300/// async_std::task::block_on(async move {
301/// while let Some(event) = events.next().await {
302/// dbg!(event.unwrap());
303/// }
304/// });
305/// ```
306pub trait ClientExt {
307 /// Connect to an event sourcing / server-sent events endpoint.
308 fn connect_event_source(&self, url: Url) -> EventSource;
309}
310
311impl ClientExt for surf::Client {
312 fn connect_event_source(&self, url: Url) -> EventSource {
313 EventSource::with_client(self.clone(), url)
314 }
315}