use std::io::{self, Read, Write};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use go_lib::chan::{chan, Sender};
use go_lib::net::{TcpListener, TcpStream};
use go_lib::sync::WaitGroup;
use rustls::ServerConnection;
use url::Url;
use crate::error::HttpError;
use crate::handler::{default_serve_mux, Handler};
use crate::header::Header;
use crate::parse::request::{read_request, DEFAULT_MAX_HEADER_BYTES};
use crate::parse::transfer::Body;
use crate::status;
use crate::request::Request;
use crate::response::{ConnResponseWriter, ResponseWriter};
pub struct Server {
pub addr: String,
pub handler: Option<Arc<dyn Handler>>,
pub read_timeout: Option<Duration>,
pub write_timeout: Option<Duration>,
pub idle_timeout: Option<Duration>,
pub max_header_bytes: usize,
pub max_body_bytes: Option<u64>,
shutdown_tx: Mutex<Option<Sender<()>>>,
active_conns: Arc<WaitGroup>,
}
impl Server {
pub fn new(addr: impl Into<String>) -> Self {
Self {
addr: addr.into(),
handler: None,
read_timeout: None,
write_timeout: None,
idle_timeout: None,
max_header_bytes: DEFAULT_MAX_HEADER_BYTES,
max_body_bytes: None,
shutdown_tx: Mutex::new(None),
active_conns: Arc::new(WaitGroup::new()),
}
}
pub fn listen_and_serve(&self) -> Result<(), HttpError> {
let listener = TcpListener::bind(&self.addr as &str).map_err(HttpError::Io)?;
let handler: Arc<dyn Handler> = match &self.handler {
Some(h) => Arc::clone(h),
None => default_serve_mux(),
};
let max_header_bytes = self.max_header_bytes;
let max_body_bytes = self.max_body_bytes;
let idle_timeout = self.idle_timeout;
let active_conns = Arc::clone(&self.active_conns);
let (shutdown_tx, shutdown_rx) = chan::<()>(1);
*self.shutdown_tx.lock().unwrap() = Some(shutdown_tx);
let (conn_tx, conn_rx) = chan::<TcpStream>(8);
go_lib::go!(move || {
loop {
match listener.accept() {
Err(_) => break,
Ok(stream) => {
let _ = std::panic::catch_unwind(
std::panic::AssertUnwindSafe(|| conn_tx.send(stream))
);
}
}
}
});
loop {
go_lib::select! {
recv(shutdown_rx) -> _sig => { break }
recv(conn_rx) -> conn => {
match conn {
None => break,
Some(stream) => {
let h = Arc::clone(&handler);
let wg = Arc::clone(&active_conns);
wg.add(1);
go_lib::go!(move || {
serve_conn(stream, h, max_header_bytes, max_body_bytes, idle_timeout);
wg.done();
});
}
}
}
}
}
Ok(())
}
pub fn listen_and_serve_tls(
&self,
cert_file: &str,
key_file: &str,
) -> Result<(), HttpError> {
let tls_config = crate::tls::server_config(cert_file, key_file)?;
let listener = TcpListener::bind(&self.addr as &str).map_err(HttpError::Io)?;
let handler: Arc<dyn Handler> = match &self.handler {
Some(h) => Arc::clone(h),
None => default_serve_mux(),
};
let max_header_bytes = self.max_header_bytes;
let max_body_bytes = self.max_body_bytes;
let idle_timeout = self.idle_timeout;
let active_conns = Arc::clone(&self.active_conns);
let (shutdown_tx, shutdown_rx) = chan::<()>(1);
*self.shutdown_tx.lock().unwrap() = Some(shutdown_tx);
let (conn_tx, conn_rx) = chan::<TcpStream>(8);
go_lib::go!(move || {
loop {
match listener.accept() {
Err(_) => break,
Ok(stream) => {
let _ = std::panic::catch_unwind(
std::panic::AssertUnwindSafe(|| conn_tx.send(stream))
);
}
}
}
});
loop {
go_lib::select! {
recv(shutdown_rx) -> _sig => { break }
recv(conn_rx) -> conn => {
match conn {
None => break,
Some(stream) => {
let h = Arc::clone(&handler);
let cfg = Arc::clone(&tls_config);
let wg = Arc::clone(&active_conns);
wg.add(1);
go_lib::go!(move || {
serve_conn_tls(stream, cfg, h, max_header_bytes, max_body_bytes, idle_timeout);
wg.done();
});
}
}
}
}
}
Ok(())
}
pub fn shutdown(&self) {
if let Some(tx) = self.shutdown_tx.lock().unwrap().take() {
let _ = std::panic::catch_unwind(
std::panic::AssertUnwindSafe(|| tx.send(()))
);
}
self.active_conns.wait();
}
}
fn serve_conn(
stream: TcpStream,
handler: Arc<dyn Handler>,
max_header_bytes: usize,
max_body_bytes: Option<u64>,
idle_timeout: Option<Duration>,
) {
let remote_addr = stream.peer_addr()
.map(|a| a.to_string())
.unwrap_or_default();
let mut write_half = match stream.try_clone() {
Ok(s) => s,
Err(_) => return,
};
#[cfg(unix)]
let raw_fd = stream.as_raw_fd();
#[cfg(windows)]
let raw_socket = stream.as_raw_socket() as u64;
loop {
let (idle_cancel_tx, idle_cancel_rx) = go_lib::chan::chan::<()>(1);
if let Some(dur) = idle_timeout {
let (ctx, _cancel) = go_lib::context::with_timeout(
&go_lib::context::background(), dur,
);
go_lib::go!(move || {
go_lib::select! {
recv(ctx.done()) -> _sig => {
#[cfg(unix)] {
use std::os::unix::io::FromRawFd;
let s = unsafe { std::net::TcpStream::from_raw_fd(raw_fd) };
let _ = s.shutdown(std::net::Shutdown::Read);
std::mem::forget(s);
}
#[cfg(windows)] {
use std::os::windows::io::FromRawSocket;
let s = unsafe { std::net::TcpStream::from_raw_socket(raw_socket) };
let _ = s.shutdown(std::net::Shutdown::Read);
std::mem::forget(s);
}
}
recv(idle_cancel_rx) -> _sig => {}
}
});
}
let read_half = match stream.try_clone() {
Ok(s) => s,
Err(_) => break,
};
let parsed = match read_request(read_half, max_header_bytes) {
Ok(p) => { let _ = idle_cancel_tx.try_send(()); p }
Err(_) => break,
};
let connection_close = {
let hdr = parsed.header.get("Connection").unwrap_or("").to_ascii_lowercase();
hdr.contains("close") || parsed.proto_minor == 0
};
let expect = check_expect(&parsed.header);
if let ExpectResult::Unknown = &expect {
send_error_response(&mut write_half, status::EXPECTATION_FAILED);
break;
}
if let Some(max) = max_body_bytes
&& parsed.content_length > 0 && parsed.content_length as u64 > max
{
send_error_response(&mut write_half, status::REQUEST_ENTITY_TOO_LARGE);
break;
}
if let ExpectResult::Continue = expect {
let _ = write_half.write_all(b"HTTP/1.1 100 Continue\r\n\r\n");
let _ = write_half.flush();
}
let mut req = match build_request(parsed, remote_addr.clone()) {
Ok(r) => r,
Err(_) => break,
};
if let Some(max) = max_body_bytes
&& matches!(req.body, Some(Body::Chunked(_)) | Some(Body::Unbounded(_)))
{
let old = req.body.take().unwrap();
req.body = Some(old.capped(max));
}
let mut w = ConnResponseWriter::new(&mut write_half);
w.header().set("Server", "go-http/0.1");
if connection_close {
w.header().set("Connection", "close");
}
handler.serve_http(&mut w, &mut req);
if w.finish().is_err() {
break;
}
if !connection_close
&& let Some(ref mut body) = req.body
{
let mut drain = [0u8; 8192];
while body.read(&mut drain).map(|n| n > 0).unwrap_or(false) {}
}
if connection_close {
break;
}
}
}
struct RawTlsRead(*mut dyn Read);
unsafe impl Send for RawTlsRead {}
impl Read for RawTlsRead {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
unsafe { (*self.0).read(buf) }
}
}
fn serve_conn_tls(
stream: TcpStream,
tls_config: Arc<rustls::ServerConfig>,
handler: Arc<dyn Handler>,
max_header_bytes: usize,
max_body_bytes: Option<u64>,
idle_timeout: Option<Duration>,
) {
let remote_addr = stream.peer_addr()
.map(|a| a.to_string())
.unwrap_or_default();
#[cfg(unix)]
let raw_fd = stream.as_raw_fd();
#[cfg(windows)]
let raw_socket = stream.as_raw_socket() as u64;
let server_conn = match ServerConnection::new(tls_config) {
Ok(c) => c,
Err(_) => return,
};
let mut tls = rustls::StreamOwned::new(server_conn, stream);
loop {
let (idle_cancel_tx, idle_cancel_rx) = go_lib::chan::chan::<()>(1);
if let Some(dur) = idle_timeout {
let (ctx, _cancel) = go_lib::context::with_timeout(
&go_lib::context::background(), dur,
);
go_lib::go!(move || {
go_lib::select! {
recv(ctx.done()) -> _sig => {
#[cfg(unix)] {
use std::os::unix::io::FromRawFd;
let s = unsafe { std::net::TcpStream::from_raw_fd(raw_fd) };
let _ = s.shutdown(std::net::Shutdown::Read);
std::mem::forget(s);
}
#[cfg(windows)] {
use std::os::windows::io::FromRawSocket;
let s = unsafe { std::net::TcpStream::from_raw_socket(raw_socket) };
let _ = s.shutdown(std::net::Shutdown::Read);
std::mem::forget(s);
}
}
recv(idle_cancel_rx) -> _sig => {}
}
});
}
let read_ptr: *mut dyn Read = &mut tls as &mut dyn Read as *mut dyn Read;
let parsed = match read_request(RawTlsRead(read_ptr), max_header_bytes) {
Ok(p) => { let _ = idle_cancel_tx.try_send(()); p }
Err(_) => break,
};
let connection_close = {
let hdr = parsed.header.get("Connection").unwrap_or("").to_ascii_lowercase();
hdr.contains("close") || parsed.proto_minor == 0
};
let expect = check_expect(&parsed.header);
if let ExpectResult::Unknown = &expect {
send_error_response(&mut tls, status::EXPECTATION_FAILED);
break;
}
if let Some(max) = max_body_bytes
&& parsed.content_length > 0 && parsed.content_length as u64 > max
{
send_error_response(&mut tls, status::REQUEST_ENTITY_TOO_LARGE);
break;
}
if let ExpectResult::Continue = expect {
let _ = tls.write_all(b"HTTP/1.1 100 Continue\r\n\r\n");
let _ = tls.flush();
}
let mut req = match build_request_scheme(parsed, remote_addr.clone(), "https") {
Ok(r) => r,
Err(_) => break,
};
if let Some(max) = max_body_bytes
&& matches!(req.body, Some(Body::Chunked(_)) | Some(Body::Unbounded(_)))
{
let old = req.body.take().unwrap();
req.body = Some(old.capped(max));
}
let mut w = ConnResponseWriter::new(&mut tls);
w.header().set("Server", "go-http/0.1");
if connection_close {
w.header().set("Connection", "close");
}
handler.serve_http(&mut w, &mut req);
if w.finish().is_err() {
break;
}
if !connection_close
&& let Some(ref mut body) = req.body
{
let mut drain = [0u8; 8192];
while body.read(&mut drain).map(|n| n > 0).unwrap_or(false) {}
}
if connection_close {
break;
}
}
}
fn build_request(
parsed: crate::parse::request::ParsedRequest,
remote_addr: String,
) -> Result<Request, HttpError> {
build_request_scheme(parsed, remote_addr, "http")
}
fn build_request_scheme(
parsed: crate::parse::request::ParsedRequest,
remote_addr: String,
scheme: &str,
) -> Result<Request, HttpError> {
let host = if parsed.host.is_empty() { "localhost".to_owned() } else { parsed.host.clone() };
let url_str = if parsed.request_uri.starts_with("http://")
|| parsed.request_uri.starts_with("https://")
{
parsed.request_uri.clone()
} else {
format!("{scheme}://{host}{}", parsed.request_uri)
};
let url = Url::parse(&url_str).map_err(|e| HttpError::InvalidUrl(e.to_string()))?;
let ctx = go_lib::context::background();
let body = match parsed.body {
Body::Empty => None,
other => Some(other),
};
let mut req = Request::new_with_context(&parsed.method, url.as_str(), body, ctx)?;
req.proto = parsed.proto;
req.proto_major = parsed.proto_major;
req.proto_minor = parsed.proto_minor;
req.header = parsed.header;
req.host = parsed.host;
req.content_length = parsed.content_length;
req.transfer_encoding = parsed.transfer_encoding;
req.remote_addr = remote_addr;
Ok(req)
}
pub fn listen_and_serve(
addr: &str,
handler: Option<Arc<dyn Handler>>,
) -> Result<(), HttpError> {
let mut srv = Server::new(addr);
srv.handler = handler;
srv.listen_and_serve()
}
pub fn listen_and_serve_tls(
addr: &str,
cert_file: &str,
key_file: &str,
handler: Option<Arc<dyn Handler>>,
) -> Result<(), HttpError> {
let mut srv = Server::new(addr);
srv.handler = handler;
srv.listen_and_serve_tls(cert_file, key_file)
}
pub use crate::handler::handle;
pub use crate::handler::handle_func;
enum ExpectResult {
None,
Continue,
Unknown,
}
fn check_expect(header: &Header) -> ExpectResult {
match header.get("Expect") {
None => ExpectResult::None,
Some(v) if v.eq_ignore_ascii_case("100-continue") => ExpectResult::Continue,
Some(_) => ExpectResult::Unknown,
}
}
fn send_error_response<W: Write>(w: &mut W, code: u16) {
let text = crate::status::status_text(code);
let body = format!("{code} {text}\n");
let _ = write!(
w,
"HTTP/1.1 {code} {text}\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body
);
let _ = w.flush();
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn build_request_basic() {
use crate::parse::request::ParsedRequest;
use crate::parse::transfer::Body;
let mut hdr = crate::header::Header::new();
hdr.set("Host", "example.com");
let pr = ParsedRequest {
method: "GET".into(),
request_uri: "/hello?q=1".into(),
proto: "HTTP/1.1".into(),
proto_major: 1,
proto_minor: 1,
header: hdr,
body: Body::Empty,
content_length: -1,
transfer_encoding: vec![],
host: "example.com".into(),
};
let req = build_request(pr, "10.0.0.1:42".into()).unwrap();
assert_eq!(req.method, "GET");
assert_eq!(req.host, "example.com");
assert_eq!(req.url.path(), "/hello");
assert_eq!(req.url.query(), Some("q=1"));
assert_eq!(req.remote_addr, "10.0.0.1:42");
}
#[test]
fn response_writer_output() {
let mut buf = Vec::<u8>::new();
let mut w = ConnResponseWriter::new(&mut buf);
w.header().set("Content-Type", "text/plain");
w.write_header(200);
w.write(b"Hello!").unwrap();
w.finish().unwrap();
let s = String::from_utf8(buf).unwrap();
assert!(s.starts_with("HTTP/1.1 200 OK\r\n"), "bad status: {s:?}");
assert!(s.contains("Content-Type: text/plain\r\n"));
assert!(s.contains("Hello!"));
assert!(s.contains("0\r\n\r\n"), "missing chunked terminal: {s:?}");
}
}