tako-rs 2.0.0

Multi-transport Rust framework for modern network services.
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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170

//! End-to-end tests for the unified `Server::builder()` entry point.
//!
//! These spin up a real TCP listener, hit it with a tokio `TcpStream`, and
//! verify the response. The compio path is excluded — these tests only run on
//! the default tokio transport.

#![cfg(not(feature = "compio"))]

use std::time::Duration;

use http::Method;
use http::StatusCode;
use tako::Server;
use tako::ServerConfig;
use tako::body::TakoBody;
use tako::router::Router;
use tako::types::Request;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tokio::net::TcpStream;

async fn hello(_req: Request) -> &'static str {
  "ok"
}

async fn fetch_status_line(addr: &std::net::SocketAddr) -> String {
  let mut stream = TcpStream::connect(addr).await.unwrap();
  let _ = stream
    .write_all(b"GET /ping HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
    .await;
  let mut buf = Vec::new();
  let _ = stream.read_to_end(&mut buf).await;
  let txt = String::from_utf8_lossy(&buf).to_string();
  txt.lines().next().unwrap_or("").to_string()
}

#[tokio::test]
async fn server_builder_handles_http_request() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();

  let mut router = Router::new();
  router.get("/ping", hello);

  let server = Server::builder().config(ServerConfig::default()).build();
  let handle = server.spawn_http(listener, router);

  // Tiny delay to let the spawn task start the accept loop.
  tokio::time::sleep(Duration::from_millis(50)).await;

  let status_line = fetch_status_line(&addr).await;
  assert!(
    status_line.starts_with("HTTP/1.1 200"),
    "status line was {status_line:?}"
  );

  // Trigger graceful shutdown.
  handle.shutdown(Duration::from_secs(2)).await;
}

#[tokio::test]
async fn server_builder_default_config_is_default() {
  let server = Server::builder().build();
  assert_eq!(server.config().drain_timeout, Duration::from_secs(30));
  assert_eq!(server.config().h2_max_concurrent_streams, 100);
}

// Smoke-test the raw TCP path on the builder.
#[tokio::test]
async fn server_builder_handles_raw_tcp_echo() {
  use tokio::io::AsyncReadExt as _;
  use tokio::io::AsyncWriteExt as _;

  // Bind once just to discover a free port, then drop the listener so the
  // builder can rebind via serve_tcp_with_shutdown.
  let probe = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = probe.local_addr().unwrap();
  drop(probe);

  let server = Server::builder().build();
  let handle = server.spawn_tcp_raw(addr.to_string(), |mut stream, _peer| {
    Box::pin(async move {
      let mut buf = [0u8; 16];
      let n = stream.read(&mut buf).await?;
      stream.write_all(&buf[..n]).await?;
      Ok(())
    })
  });

  tokio::time::sleep(Duration::from_millis(80)).await;

  let mut s = TcpStream::connect(&addr).await.unwrap();
  s.write_all(b"hi").await.unwrap();
  let mut out = [0u8; 16];
  let n = s.read(&mut out).await.unwrap();
  assert_eq!(&out[..n], b"hi");

  handle.shutdown(Duration::from_secs(2)).await;
}

// Smoke-test the raw UDP path on the builder.
#[tokio::test]
async fn server_builder_handles_raw_udp_echo() {
  let probe = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
  let addr = probe.local_addr().unwrap();
  drop(probe);

  let server = Server::builder().build();
  let handle = server.spawn_udp_raw(addr.to_string(), |data, peer, sock| {
    Box::pin(async move {
      let _ = sock.send_to(&data, peer).await;
    })
  });

  tokio::time::sleep(Duration::from_millis(80)).await;

  let client = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
  client.send_to(b"udp-hi", &addr).await.unwrap();
  let mut buf = [0u8; 32];
  let recv = tokio::time::timeout(Duration::from_secs(2), client.recv_from(&mut buf))
    .await
    .unwrap()
    .unwrap();
  assert_eq!(&buf[..recv.0], b"udp-hi");

  handle.trigger();
}

// Smoke-test that the 405 + Allow path keeps working through the builder
// without any caller-side wiring.
#[tokio::test]
async fn server_builder_propagates_405_with_allow() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();

  let mut router = Router::new();
  router.route(Method::GET, "/only-get", hello);

  let server = Server::builder().build();
  let handle = server.spawn_http(listener, router);

  tokio::time::sleep(Duration::from_millis(50)).await;

  let mut stream = TcpStream::connect(&addr).await.unwrap();
  let _ = stream
    .write_all(
      b"POST /only-get HTTP/1.1\r\nHost: localhost\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
    )
    .await;
  let mut buf = Vec::new();
  let _ = stream.read_to_end(&mut buf).await;
  let txt = String::from_utf8_lossy(&buf);
  assert!(
    txt.contains(&format!(
      "HTTP/1.1 {}",
      StatusCode::METHOD_NOT_ALLOWED.as_u16()
    )),
    "wanted 405, got: {txt:?}"
  );
  assert!(
    txt.to_ascii_lowercase().contains("allow:"),
    "wanted Allow header, got: {txt:?}"
  );

  // Drop the in-flight body so we can use TakoBody import without a warning.
  let _ = TakoBody::empty();

  handle.shutdown(Duration::from_secs(2)).await;
}