http_sse/
client.rs

1// Copyright 2019 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::{Event, EventSource};
6#[cfg(target_os = "fuchsia")]
7use fidl_fuchsia_net_http as fnet_http;
8#[cfg(target_os = "fuchsia")]
9use fuchsia_async as fasync;
10use fuchsia_hyper::HttpsClient;
11#[cfg(target_os = "fuchsia")]
12use futures::io::AsyncReadExt as _;
13use futures::stream::{Stream, StreamExt as _};
14use futures::task::{Context, Poll};
15use hyper::{Request, StatusCode};
16use std::pin::Pin;
17use thiserror::Error;
18
19/// An http SSE client.
20#[derive(derivative::Derivative)]
21#[derivative(Debug)]
22pub struct Client {
23    #[derivative(Debug = "ignore")]
24    chunks: futures::stream::BoxStream<'static, Result<hyper::body::Bytes, anyhow::Error>>,
25    source: EventSource,
26    events: std::vec::IntoIter<Event>,
27}
28
29impl Client {
30    /// Connects to an http url and, on success, returns a `Stream` of SSE events.
31    pub async fn from_hyper_client(
32        https_client: &HttpsClient,
33        url: impl AsRef<str>,
34    ) -> Result<Self, FromHyperClientError> {
35        let request = Request::get(url.as_ref())
36            .header("accept", "text/event-stream")
37            .body(http_body_util::Full::default())
38            .map_err(|e| FromHyperClientError::CreateRequest(e))?;
39        let response = https_client
40            .request(request)
41            .await
42            .map_err(|e| FromHyperClientError::MakeRequest(e))?;
43        if response.status() != StatusCode::OK {
44            return Err(FromHyperClientError::HttpStatus(response.status()));
45        }
46        Ok(Self {
47            chunks: http_body_util::BodyStream::new(response.into_body())
48                .filter_map(|frame_res| async move {
49                    match frame_res {
50                        Ok(frame) => match frame.into_data() {
51                            Ok(bytes) => Some(Ok(bytes)),
52                            Err(_) => None,
53                        },
54                        Err(e) => Some(Err(anyhow::anyhow!(e))),
55                    }
56                })
57                .boxed(),
58            source: EventSource::new(),
59            events: vec![].into_iter(),
60        })
61    }
62
63    /// Connects to an http url and, on success, returns a `Stream` of SSE events.
64    ///
65    /// `stream_buf_size` is the size of the buffer to use when reading from the http response body.
66    /// Consider making it a small multiple of the expected event size to optimize for low memory
67    /// usage or a large multiple to optimize for event processing speed.
68    #[cfg(target_os = "fuchsia")]
69    pub async fn from_http_loader(
70        loader: &fnet_http::LoaderProxy,
71        url: String,
72        stream_buf_size: usize,
73    ) -> Result<Self, FromHttpLoaderError> {
74        let resp = loader
75            .fetch(fnet_http::Request {
76                method: None,
77                url: Some(url),
78                headers: Some(vec![fnet_http::Header {
79                    name: b"accept".to_vec(),
80                    value: b"text/event-stream".to_vec(),
81                }]),
82                body: None,
83                deadline: None,
84                ..Default::default()
85            })
86            .await
87            .map_err(FromHttpLoaderError::FetchFidl)?;
88        let socket = match resp {
89            fnet_http::Response {
90                error: None, body: Some(body), status_code: Some(200), ..
91            } => body,
92            fnet_http::Response { error, status_code, status_line, .. } => {
93                return Err(FromHttpLoaderError::FetchError { error, status_code, status_line });
94            }
95        };
96        Ok(Self {
97            chunks: stream_from_socket(fasync::Socket::from_socket(socket), stream_buf_size),
98            source: EventSource::new(),
99            events: vec![].into_iter(),
100        })
101    }
102}
103
104#[cfg(target_os = "fuchsia")]
105fn stream_from_socket(
106    socket: fasync::Socket,
107    stream_buf_size: usize,
108) -> futures::stream::BoxStream<'static, Result<hyper::body::Bytes, anyhow::Error>> {
109    let initial_buf = vec![0; stream_buf_size];
110    futures::stream::try_unfold((socket, initial_buf), |(mut socket, mut buf)| async move {
111        match socket.read(&mut buf).await {
112            Ok(0) => Ok(None),
113            Ok(n) => {
114                let chunk = hyper::body::Bytes::copy_from_slice(&buf[..n]);
115                Ok(Some((chunk, (socket, buf))))
116            }
117            Err(e) => Err(anyhow::anyhow!(e).context("reading from socket")),
118        }
119    })
120    .boxed()
121}
122
123#[derive(Debug, Error)]
124pub enum FromHyperClientError {
125    #[error("error creating http request")]
126    CreateRequest(#[source] hyper::http::Error),
127
128    #[error("error making http request")]
129    MakeRequest(#[source] hyper_util::client::legacy::Error),
130
131    #[error("http server responded with status other than OK: {0}")]
132    HttpStatus(hyper::StatusCode),
133}
134
135#[cfg(target_os = "fuchsia")]
136#[derive(Debug, Error)]
137pub enum FromHttpLoaderError {
138    #[error("fuchsia.net.http/Loader.Fetch fidl error")]
139    FetchFidl(#[source] fidl::Error),
140
141    #[error(
142        "fuchsia.net.http/Loader.Fetch error: status {status_code:?} {error:?} {status_line:?}"
143    )]
144    FetchError {
145        error: Option<fnet_http::Error>,
146        status_code: Option<u32>,
147        status_line: Option<Vec<u8>>,
148    },
149}
150
151impl Stream for Client {
152    type Item = Result<Event, ClientPollError>;
153
154    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
155        loop {
156            if let Some(event) = self.events.next() {
157                return Poll::Ready(Some(Ok(event)));
158            }
159            match Pin::new(&mut self.chunks).poll_next(cx) {
160                Poll::Ready(Some(Ok(chunk))) => {
161                    self.events = self.source.parse(&chunk).into_iter();
162                }
163                Poll::Ready(Some(Err(e))) => {
164                    return Poll::Ready(Some(Err(ClientPollError::NextChunk(e))));
165                }
166                Poll::Ready(None) => {
167                    return Poll::Ready(None);
168                }
169                Poll::Pending => {
170                    return Poll::Pending;
171                }
172            }
173        }
174    }
175}
176
177#[derive(Debug, Error)]
178pub enum ClientPollError {
179    #[error("error downloading next chunk")]
180    NextChunk(#[source] anyhow::Error),
181}
182
183#[cfg(test)]
184mod tests {
185    use super::*;
186    use assert_matches::assert_matches;
187    use fuchsia_async as fasync;
188    use fuchsia_async::net::TcpListener;
189    use fuchsia_hyper::new_https_client;
190    use futures::TryStreamExt as _;
191    use futures::future::Future;
192    use hyper::Response;
193    use hyper::service::service_fn;
194    use std::net::{Ipv4Addr, SocketAddr};
195    use test_case::test_case;
196    pub enum Body {
197        Empty(http_body_util::Empty<hyper::body::Bytes>),
198        Full(http_body_util::Full<hyper::body::Bytes>),
199        Stream(
200            http_body_util::StreamBody<
201                std::pin::Pin<
202                    Box<
203                        dyn futures::stream::Stream<
204                                Item = Result<
205                                    hyper::body::Frame<hyper::body::Bytes>,
206                                    Box<dyn std::error::Error + Send + Sync + 'static>,
207                                >,
208                            > + Send,
209                    >,
210                >,
211            >,
212        ),
213    }
214
215    impl Body {
216        pub fn empty() -> Self {
217            Body::Empty(http_body_util::Empty::new())
218        }
219        pub fn wrap_stream<S, O, E>(stream: S) -> Self
220        where
221            S: futures::stream::Stream<Item = Result<O, E>> + Send + 'static,
222            O: Into<hyper::body::Bytes>,
223            E: Into<Box<dyn std::error::Error + Send + Sync + 'static>>,
224        {
225            let stream = stream.map(|res| match res {
226                Ok(data) => Ok(hyper::body::Frame::data(data.into())),
227                Err(e) => Err(e.into()),
228            });
229            Body::Stream(http_body_util::StreamBody::new(Box::pin(stream)))
230        }
231    }
232    impl From<Vec<u8>> for Body {
233        fn from(data: Vec<u8>) -> Self {
234            Body::Full(http_body_util::Full::new(data.into()))
235        }
236    }
237    impl hyper::body::Body for Body {
238        type Data = hyper::body::Bytes;
239        type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
240
241        fn poll_frame(
242            self: std::pin::Pin<&mut Self>,
243            cx: &mut std::task::Context<'_>,
244        ) -> std::task::Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
245            match self.get_mut() {
246                Body::Empty(b) => std::pin::Pin::new(b).poll_frame(cx).map_err(|e| match e {}),
247                Body::Full(b) => std::pin::Pin::new(b).poll_frame(cx).map_err(|e| match e {}),
248                Body::Stream(b) => std::pin::Pin::new(b).poll_frame(cx),
249            }
250        }
251    }
252
253    fn spawn_server<F>(handle_req: fn(Request<hyper::body::Incoming>) -> F) -> String
254    where
255        F: Future<Output = Result<Response<Body>, hyper::Error>> + Send + 'static,
256    {
257        use futures::StreamExt as _;
258        use hyper_util::rt::TokioIo;
259
260        let (listener, url) = {
261            let listener =
262                TcpListener::bind(&SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0)).unwrap();
263            let local_addr = listener.local_addr().unwrap();
264            (listener.accept_stream(), format!("http://{}", local_addr))
265        };
266        let builder = hyper_util::server::conn::auto::Builder::new(fuchsia_hyper::Executor);
267        fasync::Task::spawn(async move {
268            let mut tasks = futures::stream::FuturesUnordered::new();
269            let mut listener = listener.fuse();
270            loop {
271                futures::select! {
272                    res = listener.next() => {
273                        match res {
274                            Some(Ok((stream, _addr))) => {
275                                let stream = fuchsia_hyper::TcpStream { stream };
276                                let service = service_fn(handle_req);
277                                let builder = builder.clone();
278                                tasks.push(fasync::Task::spawn(async move {
279                                    let _ = builder.serve_connection(TokioIo::new(stream), service).await;
280                                }));
281                            }
282                            _ => break,
283                        }
284                    }
285                    _ = tasks.next() => {}
286                }
287            }
288        }).detach();
289        url
290    }
291
292    fn make_event() -> Event {
293        Event::from_type_and_data("event_type", "data_contents").unwrap()
294    }
295
296    async fn make_client(byte_source: &str, url: String) -> Client {
297        match byte_source {
298            "hyper" => Client::from_hyper_client(&new_https_client(), url).await.unwrap(),
299            #[cfg(target_os = "fuchsia")]
300            "loader" => Client::from_http_loader(
301                &fuchsia_component::client::connect_to_protocol::<fnet_http::LoaderMarker>()
302                    .unwrap(),
303                url,
304                50,
305            )
306            .await
307            .unwrap(),
308            s => panic!("unexpected byte_soure {s}"),
309        }
310    }
311
312    #[test_case("hyper")]
313    #[cfg(target_os = "fuchsia")]
314    #[test_case("loader")]
315    #[fasync::run_singlethreaded(test)]
316    async fn receive_one_event(byte_source: &str) {
317        async fn handle_req(
318            _req: Request<hyper::body::Incoming>,
319        ) -> Result<Response<Body>, hyper::Error> {
320            Ok(Response::builder()
321                .status(StatusCode::OK)
322                .header("content-type", "text/event-stream")
323                .body(make_event().to_vec().into())
324                .unwrap())
325        }
326        let url = spawn_server(handle_req);
327
328        let client = make_client(byte_source, url).await;
329        let events: Result<Vec<_>, _> = client.collect::<Vec<_>>().await.into_iter().collect();
330
331        assert_eq!(events.unwrap(), vec![make_event()]);
332    }
333
334    #[test_case("hyper")]
335    #[cfg(target_os = "fuchsia")]
336    #[test_case("loader")]
337    #[fasync::run_singlethreaded(test)]
338    async fn client_sends_correct_http_headers(byte_source: &str) {
339        async fn handle_req(
340            req: Request<hyper::body::Incoming>,
341        ) -> Result<Response<Body>, hyper::Error> {
342            assert_eq!(req.method(), &hyper::Method::GET);
343            assert_eq!(
344                req.headers().get("accept").map(|h| h.as_bytes()),
345                Some(&b"text/event-stream"[..])
346            );
347            Ok(Response::builder()
348                .status(StatusCode::OK)
349                .header("content-type", "text/event-stream")
350                .body(make_event().to_vec().into())
351                .unwrap())
352        }
353        let url = spawn_server(handle_req);
354
355        let client = make_client(byte_source, url).await;
356        client.collect::<Vec<_>>().await;
357    }
358
359    #[fasync::run_singlethreaded(test)]
360    async fn error_create_request() {
361        assert_matches!(
362            Client::from_hyper_client(&new_https_client(), "\n").await,
363            Err(FromHyperClientError::CreateRequest(_))
364        );
365    }
366
367    #[fasync::run_singlethreaded(test)]
368    async fn error_make_request() {
369        assert_matches!(
370            Client::from_hyper_client(&new_https_client(), "bad_url2").await,
371            Err(FromHyperClientError::MakeRequest(_))
372        );
373    }
374
375    #[fasync::run_singlethreaded(test)]
376    async fn hyper_error_http_status() {
377        async fn handle_req(
378            _req: Request<hyper::body::Incoming>,
379        ) -> Result<Response<Body>, hyper::Error> {
380            Ok(Response::builder().status(StatusCode::NOT_FOUND).body(Body::empty()).unwrap())
381        }
382        let url = spawn_server(handle_req);
383
384        assert_matches!(
385            Client::from_hyper_client(&new_https_client(), url).await,
386            Err(FromHyperClientError::HttpStatus(_))
387        );
388    }
389
390    #[fasync::run_singlethreaded(test)]
391    async fn loader_error_http_status() {
392        async fn handle_req(
393            _req: Request<hyper::body::Incoming>,
394        ) -> Result<Response<Body>, hyper::Error> {
395            Ok(Response::builder().status(StatusCode::NOT_FOUND).body(Body::empty()).unwrap())
396        }
397        let url = spawn_server(handle_req);
398
399        assert_matches!(
400            Client::from_http_loader(
401                &fuchsia_component::client::connect_to_protocol::<fnet_http::LoaderMarker>()
402                    .unwrap(),
403                url,
404                50,
405            )
406            .await,
407            Err(FromHttpLoaderError::FetchError { status_code: Some(404), .. })
408        );
409    }
410
411    // The fuchsia.net.http.Response contains the http response body as a zx::Socket of bytes,
412    // so it doesn't have stream reading errors unless there is an error reading from the actual
413    // socket.
414    #[fasync::run_singlethreaded(test)]
415    async fn error_downloading_chunk() {
416        // If the body of an http response is not large enough, hyper will download the body
417        // along with the header in the initial fuchsia_hyper::HttpsClient.request(). This means
418        // that even if the body is implemented with a stream that fails before the transfer is
419        // complete, the failure will occur during the initial request, before awaiting on the
420        // body chunk stream.
421        const BODY_SIZE_LARGE_ENOUGH_TO_TRIGGER_DELAYED_STREAMING: usize = 1_000_000;
422
423        async fn handle_req(
424            _req: Request<hyper::body::Incoming>,
425        ) -> Result<Response<Body>, hyper::Error> {
426            Ok(Response::builder()
427                .status(StatusCode::OK)
428                .header(
429                    "content-length",
430                    &format!("{}", BODY_SIZE_LARGE_ENOUGH_TO_TRIGGER_DELAYED_STREAMING),
431                )
432                .header("content-type", "text/event-stream")
433                .body(Body::wrap_stream(futures::stream::iter(vec![
434                    Ok(vec![b' '; BODY_SIZE_LARGE_ENOUGH_TO_TRIGGER_DELAYED_STREAMING - 1]),
435                    Err("error-text".to_string()),
436                ])))
437                .unwrap())
438        }
439        let url = spawn_server(handle_req);
440        let mut client = Client::from_hyper_client(&new_https_client(), url).await.unwrap();
441
442        assert_matches!(client.try_next().await, Err(ClientPollError::NextChunk(_)));
443    }
444
445    #[test]
446    fn test_stream_from_socket() {
447        let mut executor = fasync::TestExecutor::new_with_fake_time();
448        let (snd, rcv) = zx::Socket::create_stream();
449        let mut stream = stream_from_socket(fasync::Socket::from_socket(rcv), 10);
450
451        // Empty socket is pending.
452        let mut next = stream.next();
453        assert_matches!(executor.run_until_stalled(&mut next), Poll::Pending);
454
455        // Non-empty socket yields correctly truncated Bytes.
456        assert_eq!(snd.write(b"test msg").unwrap(), 8);
457        assert_matches!(
458            executor.run_until_stalled(&mut next),
459            Poll::Ready(Some(Ok(b))) if (&*b)[..] == *b"test msg".as_slice()
460        );
461        drop(next);
462
463        // Write larger than buf split across multiple yields.
464        assert_eq!(snd.write(b"0000000000111").unwrap(), 13);
465        assert_matches!(
466            executor.run_until_stalled(&mut stream.next()),
467            Poll::Ready(Some(Ok(b))) if (&*b)[..] == *b"0000000000".as_slice()
468        );
469        assert_matches!(
470            executor.run_until_stalled(&mut stream.next()),
471            Poll::Ready(Some(Ok(b))) if (&*b)[..] == *b"111".as_slice()
472        );
473
474        // Closed socket is finished.
475        drop(snd);
476        assert_matches!(executor.run_until_stalled(&mut stream.next()), Poll::Ready(None));
477    }
478}
http_sse/client.rs

http_sse/
client.rs
