camber 0.2.1

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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

use crate::runtime_support as common;

use camber::http::{Request, Router};
use camber::runtime;
use std::io::{BufReader, Write};
use std::net::TcpStream;
use std::sync::mpsc;
use std::time::Duration;

#[test]
fn sse_streams_multiple_events() {
    common::test_runtime()
        .shutdown_timeout(Duration::from_secs(2))
        .run(|| {
            let mut router = Router::new();
            router.get_sse(
                "/events",
                |_req: &Request, writer: &mut camber::http::SseWriter| {
                    for i in 0..3 {
                        writer.event("message", &format!("data-{i}"))?;
                    }
                    Ok(())
                },
            );

            let addr = common::spawn_server(router);

            let response =
                crate::http::request(addr, "GET", "/events", &[], &[], Duration::from_secs(5))
                    .expect("read complete SSE response");
            assert_eq!(response.status, 200);
            assert_eq!(response.header("content-type"), Some("text/event-stream"));
            assert_eq!(response.header("cache-control"), Some("no-cache"));
            assert_eq!(response.header("transfer-encoding"), Some("chunked"));
            assert_eq!(
                response.body.as_ref(),
                b"event: message\ndata: data-0\n\nevent: message\ndata: data-1\n\nevent: message\ndata: data-2\n\n"
            );

            runtime::request_shutdown();
        })
        .unwrap();
}

#[test]
fn sse_route_ignores_request_body_limit() {
    common::test_runtime()
        .shutdown_timeout(Duration::from_secs(2))
        .run(|| {
            let mut router = Router::new().max_request_body(10);
            router.get_sse(
                "/events",
                |_req: &Request, writer: &mut camber::http::SseWriter| {
                    writer.event("ping", "hello")?;
                    Ok(())
                },
            );

            let addr = common::spawn_server(router);

            let body = "x".repeat(1024);
            let response = crate::http::request(
                addr,
                "GET",
                "/events",
                &[],
                body.as_bytes(),
                Duration::from_secs(5),
            )
            .expect("read complete SSE response for oversized request body");
            assert_eq!(response.status, 200);
            assert_eq!(response.header("content-type"), Some("text/event-stream"));
            assert_eq!(response.header("cache-control"), Some("no-cache"));
            assert_eq!(response.header("transfer-encoding"), Some("chunked"));
            assert_eq!(response.body.as_ref(), b"event: ping\ndata: hello\n\n");

            runtime::request_shutdown();
        })
        .unwrap();
}

#[test]
fn sse_client_disconnect_stops_handler() {
    common::test_runtime()
        .shutdown_timeout(Duration::from_secs(2))
        .run(|| {
            let (stopped_tx, stopped_rx) = mpsc::sync_channel(1);

            let mut router = Router::new();
            router.get_sse(
                "/stream",
                move |_req: &Request, writer: &mut camber::http::SseWriter| {
                    loop {
                        match writer.event("tick", "ping") {
                            Ok(()) => {}
                            Err(_) => {
                                stopped_tx.send(()).unwrap();
                                return Ok(());
                            }
                        }
                    }
                },
            );

            let addr = common::spawn_server(router);

            // Connect and read 2 events, then drop
            {
                let mut stream = TcpStream::connect(addr).unwrap();
                stream
                    .set_read_timeout(Some(Duration::from_secs(5)))
                    .unwrap();
                write!(
                    stream,
                    "GET /stream HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"
                )
                .unwrap();
                stream.flush().unwrap();

                let mut reader = BufReader::new(stream);
                let (status, _) = crate::wire::read_response_head(&mut reader);
                assert_eq!(status, 200);
                for _ in 0..2 {
                    let event = crate::wire::read_chunk(&mut reader, 1024)
                        .expect("decode bounded SSE chunk")
                        .expect("SSE stream remained open for two events");
                    assert_eq!(event.as_ref(), b"event: tick\ndata: ping\n\n");
                }
            }
            stopped_rx
                .recv_timeout(Duration::from_secs(2))
                .expect("SSE handler observed client disconnect");

            runtime::request_shutdown();
        })
        .unwrap();
}