camber 0.1.6

Opinionated async Rust for IO-bound services on top of Tokio
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

use super::response::HeaderPair;
use crate::RuntimeError;
use bytes::Bytes;
use std::borrow::Cow;
use tokio::sync::mpsc;

/// Default channel buffer capacity for streaming responses.
const DEFAULT_STREAM_BUFFER: usize = 32;

impl std::fmt::Debug for StreamSender {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StreamSender").finish_non_exhaustive()
    }
}

/// Sender half of a streaming HTTP response.
///
/// Push chunks to the connected client. Returns `RuntimeError::ChannelClosed`
/// when the client disconnects or the receiver is dropped.
pub struct StreamSender {
    tx: mpsc::Sender<Bytes>,
}

impl StreamSender {
    /// Send one response chunk to the client.
    pub async fn send(&self, data: impl Into<Bytes>) -> Result<(), RuntimeError> {
        self.tx
            .send(data.into())
            .await
            .map_err(|_| RuntimeError::ChannelClosed)
    }
}

impl std::fmt::Debug for StreamResponse {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StreamResponse")
            .field("status", &self.status)
            .field("header_count", &self.headers.len())
            .finish()
    }
}

/// A streaming HTTP response returned from async handlers.
///
/// Created via `StreamResponse::new()`, which returns both the response
/// (to return from the handler) and a `StreamSender` (to push chunks).
pub struct StreamResponse {
    status: u16,
    headers: Vec<HeaderPair>,
    rx: mpsc::Receiver<Bytes>,
}

impl StreamResponse {
    /// Create a streaming response with the given status code.
    ///
    /// Returns the response to return from the handler and a sender
    /// for pushing body chunks. Uses the default buffer capacity.
    pub fn new(status: u16) -> (Self, StreamSender) {
        let (tx, rx) = mpsc::channel(DEFAULT_STREAM_BUFFER);
        let resp = Self {
            status,
            headers: Vec::new(),
            rx,
        };
        (resp, StreamSender { tx })
    }

    /// Create a streaming response with explicit buffer capacity.
    ///
    /// `cap` controls the channel depth for backpressure. Must be greater
    /// than zero.
    pub fn with_buffer(status: u16, cap: usize) -> Result<(Self, StreamSender), RuntimeError> {
        match cap {
            0 => Err(RuntimeError::InvalidArgument(
                "stream buffer capacity must be greater than zero".into(),
            )),
            _ => {
                let (tx, rx) = mpsc::channel(cap);
                let resp = Self {
                    status,
                    headers: Vec::new(),
                    rx,
                };
                Ok((resp, StreamSender { tx }))
            }
        }
    }

    /// Add a custom header to the streaming response.
    pub fn with_header(mut self, name: &str, value: &str) -> Self {
        self.headers
            .push((Cow::Owned(name.to_owned()), Cow::Owned(value.to_owned())));
        self
    }

    pub(crate) fn into_parts(self) -> StreamParts {
        StreamParts {
            status: self.status,
            headers: self.headers.into_boxed_slice(),
            rx: self.rx,
        }
    }
}

pub(crate) struct StreamParts {
    pub(crate) status: u16,
    pub(crate) headers: Box<[HeaderPair]>,
    pub(crate) rx: mpsc::Receiver<Bytes>,
}