ftnet-utils 0.1.1

ftnet utilities
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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

pub async fn peer_to_http(
    addr: &str,
    client_pools: ftnet_utils::HttpConnectionPools,
    send: &mut iroh::endpoint::SendStream,
    mut recv: ftnet_utils::FrameReader,
) -> eyre::Result<()> {
    use eyre::WrapErr;
    use http_body_util::BodyExt;
    use tokio_stream::StreamExt;

    tracing::info!("http request with {addr}");
    let start = std::time::Instant::now();

    let req: ftnet_utils::http::Request = match recv.next().await {
        Some(Ok(v)) => serde_json::from_str(&v)
            .wrap_err_with(|| "failed to serialize json while reading http request")?,
        Some(Err(e)) => {
            tracing::error!("failed to read request: {e}");
            return Err(eyre::anyhow!("failed to read request: {e}"));
        }
        None => {
            tracing::error!("no request");
            return Err(eyre::anyhow!("no request"));
        }
    };

    tracing::info!("got request: {req:?}");

    let mut body = recv.read_buffer().to_vec();
    let mut recv = recv.into_inner();

    let mut buf = Vec::with_capacity(1024 * 64);

    tracing::trace!("reading body");
    while let Some(v) = recv.read(&mut buf).await? {
        if v == 0 {
            tracing::trace!("finished reading");
            break;
        }
        tracing::trace!("reading body, partial: {v}");
        body.extend_from_slice(&buf);
        buf.truncate(0);
    }
    tracing::debug!("read {} bytes of body", body.len());

    let mut r = hyper::Request::builder()
        .method(req.method.as_str())
        .uri(req.uri);
    for (name, value) in req.headers {
        r = r.header(name, value);
    }

    tracing::debug!("request: {r:?}");

    let pool = get_pool(addr, client_pools).await?;
    tracing::trace!("got pool");
    let mut client = match pool.get().await {
        Ok(v) => v,
        Err(e) => {
            tracing::error!("failed to get connection: {e:?}");
            return Err(eyre::anyhow!("failed to get connection: {e:?}"));
        }
    };
    // tracing::info!("got client");

    let (resp, body) = client
        .send_request(
            r.body(
                http_body_util::Full::new(hyper::body::Bytes::from(body))
                    .map_err(|e| match e {})
                    .boxed(),
            )?,
        )
        .await
        .wrap_err_with(|| "failed to send request")?
        .into_parts();

    let r = ftnet_utils::http::Response {
        status: resp.status.as_u16(),
        headers: resp
            .headers
            .iter()
            .map(|(k, v)| (k.to_string(), v.as_bytes().to_vec()))
            .collect(),
    };

    send.write_all(
        serde_json::to_string(&r)
            .wrap_err_with(|| "failed to serialize json while writing http response")?
            .as_bytes(),
    )
    .await?;
    send.write_all(b"\n").await?;
    let bytes = body.collect().await?.to_bytes();
    tracing::debug!("got response body: {} bytes", bytes.len());
    send.write_all(&bytes).await?;

    tracing::info!("handled http request in {:?}", start.elapsed());
    Ok(())
}

async fn get_pool(
    addr: &str,
    client_pools: ftnet_utils::HttpConnectionPools,
) -> eyre::Result<bb8::Pool<ftnet_utils::HttpConnectionManager>> {
    tracing::trace!("get pool called");
    let mut pools = client_pools.lock().await;

    Ok(match pools.get(addr) {
        Some(v) => {
            tracing::debug!("found existing pool for {addr}");
            v.clone()
        }
        None => {
            tracing::debug!("creating new pool for {addr}");

            let pool = bb8::Pool::builder()
                .build(ftnet_utils::HttpConnectionManager::new(addr.to_string()))
                .await?;

            pools.insert(addr.to_string(), pool.clone());
            pool
        }
    })
}