dwh 0.7.1

A library to use the digitronic protocol dwh
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

use std::{
    sync::{Arc, Mutex},
    time::Duration,
};

use axum::{
    response::Response,
    routing::{get, post},
    Extension, Router,
};
use bytes::Bytes;
use tokio::{net::TcpListener, task::AbortHandle};

async fn fail(Extension(state): Extension<Arc<Mutex<State>>>) -> Response {
    let mut headers = http::HeaderMap::new();
    state.lock().unwrap().fail += 1;
    headers.insert("Content-Length", "1000".parse().unwrap());
    let stream = async_stream::stream! {

            yield Ok::<_, hyper::Error>(Bytes::try_from("hello").unwrap());
            // Wait for a second before sending the next chunk
    };
    let response = Response::builder()
        .header("Content-Length", "1000")
        .body(axum::body::Body::from_stream(stream))
        .unwrap();
    response
}

async fn chunk_timeout(Extension(state): Extension<Arc<Mutex<State>>>) -> Response {
    state.lock().unwrap().chunk_timeout += 1;
    let stream = async_stream::stream! {
            for _ in 0..10{
                yield Ok::<_, hyper::Error>(Bytes::try_from("hello").unwrap());
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
    };
    let response = Response::builder()
        .body(axum::body::Body::from_stream(stream))
        .unwrap();
    response
}

async fn time_to_first_byte() -> String {
    tokio::time::sleep(Duration::from_millis(500)).await;
    "hello".to_owned()
}

pub struct DroppingAbortHandler(AbortHandle);

impl Drop for DroppingAbortHandler {
    fn drop(&mut self) {
        self.0.abort();
    }
}

#[derive(Debug, Clone, Default)]
pub struct State {
    pub fail: usize,
    pub chunk_timeout: usize,
}

pub async fn create_server() -> anyhow::Result<(url::Url, DroppingAbortHandler, Arc<Mutex<State>>)>
{
    let listener = TcpListener::bind("[::]:0").await?;
    let url = format!("http://127.0.0.1:{}/", listener.local_addr()?.port()).parse()?;
    let state = Arc::new(Mutex::new(State::default()));
    let app = Router::new()
        .route("/fail", get(fail))
        .route("/chunk_timeout", get(chunk_timeout))
        .route("/web.dwh", post(chunk_timeout))
        .route("/time_to_first_byte", get(time_to_first_byte))
        .layer(Extension(state.clone()));
    let abort_handler = tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    })
    .abort_handle();
    Ok((url, DroppingAbortHandler(abort_handler), state))
}