diff --git a/eventsource-client/src/client.rs b/eventsource-client/src/client.rs index 2db6b31..804e88d 100644 --- a/eventsource-client/src/client.rs +++ b/eventsource-client/src/client.rs @@ -199,10 +199,20 @@ impl Client for ClientImpl { /// Connect to the server and begin consuming the stream. Produces a /// [`Stream`] of [`Event`](crate::Event)s wrapped in [`Result`]. /// - /// Do not use the stream after it returned an error! + /// Errors yielded by the stream are not terminal: keep polling. + /// When [`ReconnectOptions::reconnect`] is enabled (the default), + /// the stream schedules a reconnect on retryable errors and the + /// next poll resumes from a fresh connection. /// - /// After the first successful connection, the stream will - /// reconnect for retryable errors. + /// The stream is exhausted only when [`Stream::poll_next`] returns + /// [`Poll::Ready(None)`]. That happens when the underlying state + /// machine reaches `StreamClosed` (e.g. a redirect-limit overrun, + /// a malformed `Location` header, or an error during initial + /// connection while [`ReconnectOptions::retry_initial`] is + /// disabled), or after any error when reconnect is disabled. + /// + /// [`Poll::Ready(None)`]: std::task::Poll::Ready + /// [`Stream::poll_next`]: futures::Stream::poll_next fn stream(&self) -> BoxStream> { Box::pin(ReconnectingRequest::new( Arc::clone(&self.transport), @@ -495,7 +505,21 @@ impl Stream for ReconnectingRequest { }, StateProj::Connected(mut body) => match ready!(body.as_mut().poll_next(cx)) { Some(Ok(result)) => { - this.event_parser.process_bytes(result)?; + if let Err(e) = this.event_parser.process_bytes(result) { + // A parse error means the current response body is + // unusable; schedule a reconnect to abandon it. + if self.props.reconnect_opts.reconnect { + let duration = self + .as_mut() + .project() + .retry_strategy + .next_delay(Instant::now()); + self.as_mut().project().state.set(State::WaitingToReconnect( + delay(duration, "reconnecting"), + )); + } + return Poll::Ready(Some(Err(e))); + } continue; } Some(Err(e)) => { @@ -712,4 +736,67 @@ mod tests { initial_connection(&mock_server.url(), retry_initial, expected).await; } + + // When a parse error happens during streaming and reconnect is + // enabled, the next stream item should be a fresh `Connected` from + // the reconnect, not another error from continuing to drain the + // broken response body. + #[cfg(feature = "hyper")] + #[tokio::test(flavor = "multi_thread")] + async fn parser_error_schedules_reconnect_immediately() { + use crate::{Client, ClientBuilder, Error, ReconnectOptionsBuilder, SSE}; + use futures::StreamExt; + use launchdarkly_sdk_transport::HyperTransport; + + let mut server = mockito::Server::new_async().await; + let _mock = server + .mock("GET", "/") + .with_status(200) + .with_body(b"\xff\xfe:bad\n\n".as_ref()) + .create_async() + .await; + + let transport = HyperTransport::new().expect("failed to build transport"); + let client = ClientBuilder::for_url(&server.url()) + .unwrap() + .reconnect( + ReconnectOptionsBuilder::new(true) + .delay(Duration::from_millis(10)) + .delay_max(Duration::from_millis(10)) + .retry_initial(true) + .build(), + ) + .build_with_transport(transport); + + let mut stream = client.stream(); + + // Expected order: Connected, parse error, Connected (reconnect). + let mut items = Vec::new(); + tokio::time::timeout(Duration::from_secs(2), async { + while items.len() < 3 { + match stream.next().await { + Some(item) => items.push(item), + None => break, + } + } + }) + .await + .expect("timed out waiting for parse error and reconnect"); + + assert!( + matches!(items.first(), Some(Ok(SSE::Connected(_)))), + "expected initial Connected, got {:?}", + items.first() + ); + assert!( + matches!(items.get(1), Some(Err(Error::InvalidLine(_)))), + "expected InvalidLine error after first connection, got {:?}", + items.get(1) + ); + assert!( + matches!(items.get(2), Some(Ok(SSE::Connected(_)))), + "expected reconnect (Connected) immediately after parse error, got {:?}", + items.get(2) + ); + } }