fuchsia_repo/
body.rs

1// Copyright 2026 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
5//! Various types representing what can be sent from a repository.
6
7use futures::prelude::*;
8use std::pin::Pin;
9use std::task::{Context, Poll};
10
11/// Helper for sending [hyper::body::Bytes] over a channel.
12pub struct BodySender {
13    tx: futures::channel::mpsc::Sender<Result<hyper::body::Bytes, std::io::Error>>,
14}
15
16impl BodySender {
17    /// Try to send [hyper::body::Bytes] in the channel.
18    #[allow(clippy::result_unit_err)]
19    pub fn try_send_data(&mut self, data: hyper::body::Bytes) -> Result<(), ()> {
20        self.tx.try_send(Ok(data)).map_err(|_| ())
21    }
22}
23
24/// [Body] represents the type of content that be served from a repository.
25#[allow(clippy::type_complexity)]
26pub enum Body {
27    Empty(http_body_util::Empty<hyper::body::Bytes>),
28    Full(http_body_util::Full<hyper::body::Bytes>),
29    Stream(
30        http_body_util::StreamBody<
31            Pin<
32                Box<
33                    dyn Stream<
34                            Item = Result<hyper::body::Frame<hyper::body::Bytes>, std::io::Error>,
35                        > + Send
36                        + Sync,
37                >,
38            >,
39        >,
40    ),
41    #[cfg(not(target_os = "fuchsia"))]
42    Sse(http_sse::Body),
43}
44
45impl std::fmt::Debug for Body {
46    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47        match self {
48            Body::Empty(_) => f.debug_tuple("Empty").finish(),
49            Body::Full(_) => f.debug_tuple("Full").finish(),
50            Body::Stream(_) => f.debug_tuple("Stream").finish(),
51            #[cfg(not(target_os = "fuchsia"))]
52            Body::Sse(_) => f.debug_tuple("Sse").finish(),
53        }
54    }
55}
56
57impl Body {
58    /// An empty body.
59    pub fn empty() -> Self {
60        Body::Empty(http_body_util::Empty::new())
61    }
62
63    /// Stream the bytes as the body.
64    pub fn wrap_stream<S, O, E>(stream: S) -> Self
65    where
66        S: Stream<Item = Result<O, E>> + Send + Sync + 'static,
67        O: Into<hyper::body::Bytes>,
68        E: Into<Box<dyn std::error::Error + Send + Sync + 'static>>,
69    {
70        let stream = stream.map(|res| match res {
71            Ok(data) => Ok(hyper::body::Frame::data(data.into())),
72            Err(e) => Err(std::io::Error::other(e)),
73        });
74        Body::Stream(http_body_util::StreamBody::new(Box::pin(stream)))
75    }
76
77    /// Stream teh body over a channel.
78    pub fn channel() -> (BodySender, Self) {
79        let (tx, rx) = futures::channel::mpsc::channel(1);
80        (BodySender { tx }, Self::wrap_stream(rx))
81    }
82}
83
84impl From<Vec<u8>> for Body {
85    fn from(data: Vec<u8>) -> Self {
86        Body::Full(http_body_util::Full::new(data.into()))
87    }
88}
89
90impl From<String> for Body {
91    fn from(data: String) -> Self {
92        Body::Full(http_body_util::Full::new(data.into()))
93    }
94}
95
96impl From<&'static str> for Body {
97    fn from(data: &'static str) -> Self {
98        Body::Full(http_body_util::Full::new(data.into()))
99    }
100}
101
102#[cfg(not(target_os = "fuchsia"))]
103impl From<http_sse::Body> for Body {
104    fn from(body: http_sse::Body) -> Self {
105        Body::Sse(body)
106    }
107}
108
109impl hyper::body::Body for Body {
110    type Data = hyper::body::Bytes;
111    type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
112
113    fn poll_frame(
114        self: Pin<&mut Self>,
115        cx: &mut Context<'_>,
116    ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
117        match self.get_mut() {
118            Body::Empty(b) => Pin::new(b).poll_frame(cx).map_err(|e| match e {}),
119            Body::Full(b) => Pin::new(b).poll_frame(cx).map_err(|e| match e {}),
120            Body::Stream(b) => Pin::new(b).poll_frame(cx).map_err(|e| Box::new(e) as _),
121            #[cfg(not(target_os = "fuchsia"))]
122            Body::Sse(b) => {
123                Pin::new(b).poll_frame(cx).map_err(|e| Box::new(std::io::Error::other(e)) as _)
124            }
125        }
126    }
127}
fuchsia_repo/body.rs

fuchsia_repo/
body.rs
