leyline-http 0.1.0

HTTP client that mimics browsers on the wire
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

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use tokio::sync::{mpsc, oneshot};

use crate::h2::error::{ErrorCode, H2Error};

#[cfg(feature = "websocket")]
use super::driver::PeerSettingsSnapshot;
use super::driver::{
    DriverCommand, DriverRequestBody, Head, ResponseSink, STREAM_REQ_BODY_CAPACITY,
    pump_request_body,
};
use super::types::{H2ResponseEx, RequestBody, ResponseBody};

#[derive(Clone)]
pub struct H2Client {
    pub(super) tx: mpsc::Sender<DriverCommand>,
    pub(super) closed: Arc<AtomicBool>,
    #[cfg(feature = "websocket")]
    pub(super) peer_settings: Arc<PeerSettingsSnapshot>,
}

impl H2Client {
    pub async fn send_shared(
        &self,
        head: Arc<Head>,
        body: RequestBody,
        stream_response: bool,
    ) -> Result<H2ResponseEx, H2Error> {
        if self.closed.load(Ordering::Acquire) {
            return Err(H2Error::Stream {
                stream_id: 0,
                code: ErrorCode::RefusedStream,
            });
        }

        let body_in = match body {
            RequestBody::None => DriverRequestBody::None,
            RequestBody::Buffered(b) => DriverRequestBody::Buffered(b),
            RequestBody::Streaming { stream, .. } => {
                let (body_tx, body_rx) = mpsc::channel(STREAM_REQ_BODY_CAPACITY);
                tokio::spawn(pump_request_body(stream, body_tx));
                DriverRequestBody::Streaming(body_rx)
            }
        };

        let (response_tx, response_rx) = oneshot::channel::<Result<H2ResponseEx, H2Error>>();
        let (sink, stream_body_rx) = if stream_response {
            let (sink, receiver) = ResponseSink::streaming(response_tx);
            (sink, Some(receiver))
        } else {
            (ResponseSink::Buffered(response_tx), None)
        };

        let cmd = DriverCommand::SendRequest {
            head,
            body: body_in,
            sink,
        };
        self.tx.send(cmd).await.map_err(|_| H2Error::Stream {
            stream_id: 0,
            code: ErrorCode::RefusedStream,
        })?;

        match response_rx.await {
            Ok(Ok(mut resp)) => {
                if let Some(rx) = stream_body_rx {
                    resp.body = ResponseBody::Streaming(rx);
                }
                Ok(resp)
            }
            Ok(Err(e)) => Err(e),
            Err(_) => Err(H2Error::Connection {
                code: ErrorCode::NoError,
                reason: "driver dropped response sender".into(),
            }),
        }
    }

    pub async fn ping(&self) -> Result<(), H2Error> {
        if self.closed.load(Ordering::Acquire) {
            return Err(H2Error::Connection {
                code: ErrorCode::NoError,
                reason: "connection closed".into(),
            });
        }
        let (ack_tx, ack_rx) = oneshot::channel();
        self.tx
            .send(DriverCommand::Ping { ack_tx })
            .await
            .map_err(|_| H2Error::Connection {
                code: ErrorCode::NoError,
                reason: "driver task has exited".into(),
            })?;
        ack_rx.await.map_err(|_| H2Error::Connection {
            code: ErrorCode::NoError,
            reason: "connection closed before the ping was acknowledged".into(),
        })
    }

    pub fn is_closed(&self) -> bool {
        self.closed.load(Ordering::Acquire)
    }
}