trillium 1.1.0

a modular toolkit for building async web apps
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

use futures_lite::{AsyncRead, AsyncReadExt, AsyncWriteExt, future::poll_once};
use std::{
    future::{Future, pending},
    io,
    pin::Pin,
    task::{Context, Poll},
    time::Duration,
};
use test_harness::test;
use trillium::Conn;
use trillium_testing::{ArcedConnector, Connector, TestResult, client_config, config, harness};

#[test(harness)]
async fn infinitely_pending_task() -> TestResult {
    let connector = ArcedConnector::new(client_config());

    let handle = config()
        .with_host("localhost")
        .with_port(0)
        .spawn(|mut conn: Conn| async move {
            conn.cancel_on_disconnect(pending::<()>()).await;
            conn
        });

    let info = handle.info().await;
    let url = info.url().unwrap();
    let mut client = connector.connect(url).await?;

    client
        .write_all(b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
        .await?;

    let mut byte = [0u8];
    assert!(poll_once(client.read(&mut byte)).await.is_none()); // nothing to read; the handler has
    // not responded

    client.close().await?; // closing the client before we receive a response

    handle.shut_down().await; // wait for a graceful shutdown the fact that we terminate here indicates
    // that the handler is not still running even though it polls an infinitely
    // pending future

    Ok(())
}

#[test(harness)]
async fn is_disconnected() -> TestResult {
    let connector = ArcedConnector::new(client_config());
    let (delay_sender, delay_receiver) = async_channel::unbounded();
    let (disconnected_sender, disconnected_receiver) = async_channel::unbounded();
    let handle = config()
        .with_host("localhost")
        .with_port(0)
        .spawn(move |mut conn: Conn| {
            let disconnected_sender = disconnected_sender.clone();
            let delay_receiver = delay_receiver.clone();
            async move {
                delay_receiver.recv().await.unwrap();
                disconnected_sender
                    .send(conn.is_disconnected().await)
                    .await
                    .unwrap();
                conn.ok("ok")
            }
        });

    let info = handle.info().await;
    let runtime = handle.runtime();

    let url = info.url().unwrap();
    let mut client = connector.connect(url).await?;

    client
        .write_all(b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
        .await?;

    delay_sender.send(()).await?;

    assert!(!disconnected_receiver.recv().await?);

    let s = String::from_utf8(ReadAvailable(&mut client).await?)?;
    assert!(s.starts_with("HTTP/1.1 200 OK\r\n"));
    client.close().await?;

    let mut client = connector.connect(url).await?;
    client
        .write_all(b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
        .await?;
    drop(client);
    runtime.delay(Duration::from_millis(10)).await;
    delay_sender.send(()).await?;
    assert!(disconnected_receiver.recv().await?);

    handle.shut_down().await;

    Ok(())
}

struct ReadAvailable<T>(T);
impl<T: AsyncRead + Unpin> Future for ReadAvailable<T> {
    type Output = io::Result<Vec<u8>>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut buf = vec![];
        let mut bytes_read = 0;
        loop {
            if buf.len() == bytes_read {
                buf.reserve(32);
                buf.resize(buf.capacity(), 0);
            }
            match Pin::new(&mut self.0).poll_read(cx, &mut buf[bytes_read..]) {
                Poll::Ready(Ok(0)) => break,
                Poll::Ready(Ok(new_bytes)) => {
                    bytes_read += new_bytes;
                }
                Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                Poll::Pending if bytes_read == 0 => return Poll::Pending,
                Poll::Pending => break,
            }
        }

        buf.truncate(bytes_read);
        Poll::Ready(Ok(buf))
    }
}