rust_web_server/server/
mod.rs

1#[cfg(test)]
2pub mod tests;
3#[cfg(test)]
4mod example;
5
6use std::io::prelude::*;
7use std::borrow::Borrow;
8use std::net::{IpAddr, SocketAddr, TcpListener};
9use std::str::FromStr;
10use std::time::Duration;
11
12use crate::request::{METHOD, Request};
13use crate::response::{Response, STATUS_CODE_REASON_PHRASE};
14use crate::app::App;
15use crate::application::Application;
16use crate::entry_point::{bootstrap, get_ip_port_thread_count, get_request_allocation_size, set_default_values};
17use crate::header::Header;
18use crate::log::Log;
19use crate::mime_type::MimeType;
20use crate::range::{ContentRange, Range};
21use crate::symbol::SYMBOL;
22use crate::thread_pool::ThreadPool;
23
24pub struct Server {}
25impl Server {
26    pub fn process_request(mut stream: impl Read + Write + Unpin, peer_addr: SocketAddr) -> Vec<u8> {
27        let request_allocation_size = get_request_allocation_size();
28        let mut buffer = vec![0; request_allocation_size as usize];
29        let boxed_read = stream.read(&mut buffer);
30        if boxed_read.is_err() {
31            let message = boxed_read.err().unwrap().to_string();
32            eprintln!("unable to read TCP stream {}", &message);
33
34            let raw_response = Server::bad_request_response(message);
35            let boxed_stream = stream.write(raw_response.borrow());
36            if boxed_stream.is_ok() {
37                stream.flush().unwrap();
38            };
39            return raw_response;
40        }
41
42        boxed_read.unwrap();
43        let request : &[u8] = &buffer;
44
45        // let raw_request = String::from_utf8(Vec::from(request)).unwrap();
46        // println!("\n\n______{}______\n\n", raw_request);
47
48
49        let boxed_request = Request::parse_request(request);
50        if boxed_request.is_err() {
51            let message = boxed_request.err().unwrap();
52            eprintln!("unable to parse request: {}", &message);
53
54            let raw_response = Server::bad_request_response(message);
55            let boxed_stream = stream.write(raw_response.borrow());
56            if boxed_stream.is_ok() {
57                stream.flush().unwrap();
58            };
59            return raw_response;
60        }
61
62
63        let request: Request = boxed_request.unwrap();
64        let (response, request) = App::handle_request(request);
65
66
67        let log_request_response = Log::combined(&request, &response, &peer_addr);
68        println!("{}", log_request_response);
69        let raw_response = Response::generate_response(response, request);
70
71        let boxed_stream = stream.write(raw_response.borrow());
72        if boxed_stream.is_ok() {
73            stream.flush().unwrap();
74        };
75
76        raw_response
77    }
78
79    pub fn bad_request_response(message: String) -> Vec<u8> {
80        let error_request = Request {
81            method: METHOD.get.to_string(),
82            request_uri: "".to_string(),
83            http_version: "".to_string(),
84            headers: vec![],
85            body: vec![],
86        };
87
88        let size = message.chars().count() as u64;
89        let content_range = ContentRange {
90            unit: Range::BYTES.to_string(),
91            range: Range { start: 0, end: size },
92            size: size.to_string(),
93            body: Vec::from(message.as_bytes()),
94            content_type: MimeType::TEXT_PLAIN.to_string(),
95        };
96
97        let header_list = Header::get_header_list(&error_request);
98        let error_response: Response = Response::get_response(
99            STATUS_CODE_REASON_PHRASE.n400_bad_request,
100            Some(header_list),
101            Some(vec![content_range])
102        );
103
104        let response = Response::generate_response(error_response, error_request);
105        return response;
106    }
107
108    pub fn process(mut stream: impl Read + Write + Unpin,
109                   connection: ConnectionInfo,
110                   app: impl Application) -> Result<(), String> {
111        use crate::http::VERSION;
112
113        let request_allocation_size = connection.request_size;
114        let client = connection.client.clone();
115        let client_addr = SocketAddr::new(IpAddr::from_str(client.ip.as_str()).unwrap(), client.port as u16);
116
117        loop {
118            let mut buffer = vec![0; request_allocation_size as usize];
119            let boxed_read = stream.read(&mut buffer);
120            if boxed_read.is_err() {
121                // timeout or client closed — normal end of keep-alive session
122                break;
123            }
124            if boxed_read.unwrap() == 0 {
125                break;
126            }
127
128            let request = match Request::parse(&buffer) {
129                Ok(r) => r,
130                Err(message) => {
131                    let raw_response = Server::bad_request_response(message.clone());
132                    let boxed_stream = stream.write(raw_response.borrow());
133                    if boxed_stream.is_ok() { stream.flush().unwrap(); }
134                    return Err(message);
135                }
136            };
137
138            let keep_alive = {
139                let conn_hdr = request.get_header(Header::_CONNECTION.to_string());
140                match conn_hdr {
141                    Some(h) => h.value.to_lowercase() != "close",
142                    None => request.http_version == VERSION.http_1_1,
143                }
144            };
145
146            let mut response = match app.execute(&request, &connection) {
147                Ok(r) => r,
148                Err(message) => {
149                    let raw_response = Server::bad_request_response(message.clone());
150                    let boxed_stream = stream.write(raw_response.borrow());
151                    if boxed_stream.is_ok() { stream.flush().unwrap(); }
152                    return Err(message);
153                }
154            };
155
156            crate::metrics::record_request();
157            crate::compression::apply_gzip(&request, &mut response);
158
159            response.headers.push(Header {
160                name: Header::_CONNECTION.to_string(),
161                value: if keep_alive { "keep-alive".to_string() } else { "close".to_string() },
162            });
163
164            Log::log_access(&request, &response, &client_addr);
165
166            if let Some(ref filepath) = response.stream_file.clone() {
167                if let Err(e) = Server::write_chunked_file(&mut stream, response, request, filepath) {
168                    return Err(e);
169                }
170            } else {
171                let raw_response = Response::generate_response(response, request);
172                if let Err(e) = stream.write(raw_response.borrow()) {
173                    return Err(e.to_string());
174                }
175                stream.flush().unwrap();
176            }
177
178            if !keep_alive { break; }
179        }
180
181        Ok(())
182    }
183
184    /// Streams a file to `stream` using HTTP/1.1 chunked transfer encoding.
185    /// The response headers are written first, then the file is read and written in 64 KB chunks.
186    pub(crate) fn write_chunked_file(
187        stream: &mut impl Write,
188        mut response: Response,
189        request: Request,
190        filepath: &str,
191    ) -> Result<(), String> {
192        use std::fs::File;
193        use std::io::Read as _;
194
195        response.headers.push(Header {
196            name: Header::_TRANSFER_ENCODING.to_string(),
197            value: "chunked".to_string(),
198        });
199
200        // build status line + headers (no body)
201        let status = [
202            response.http_version.clone(),
203            response.status_code.to_string(),
204            response.reason_phrase.clone(),
205        ].join(SYMBOL.whitespace);
206
207        let mut headers_str = SYMBOL.new_line_carriage_return.to_string();
208        for header in &response.headers {
209            headers_str.push_str(&header.name);
210            headers_str.push_str(Header::NAME_VALUE_SEPARATOR);
211            headers_str.push_str(&header.value);
212            headers_str.push_str(SYMBOL.new_line_carriage_return);
213        }
214        let head = format!("{}{}{}", status, headers_str, SYMBOL.new_line_carriage_return);
215
216        stream.write_all(head.as_bytes()).map_err(|e| e.to_string())?;
217
218        if request.method != METHOD.head && request.method != METHOD.options {
219            let mut file = File::open(filepath).map_err(|e| e.to_string())?;
220            let mut buf = vec![0u8; 65536];
221            loop {
222                let n = file.read(&mut buf).map_err(|e| e.to_string())?;
223                if n == 0 { break; }
224                // chunk header: hex size + CRLF
225                stream.write_all(format!("{:x}\r\n", n).as_bytes()).map_err(|e| e.to_string())?;
226                stream.write_all(&buf[..n]).map_err(|e| e.to_string())?;
227                stream.write_all(b"\r\n").map_err(|e| e.to_string())?;
228            }
229            // terminal chunk
230            stream.write_all(b"0\r\n\r\n").map_err(|e| e.to_string())?;
231        }
232
233        stream.flush().map_err(|e| e.to_string())
234    }
235
236    /// Reads configuration (IP, port, thread count, TLS paths) from the layered config system
237    /// and returns a bound `TcpListener` and a sized `ThreadPool`. Call once at startup.
238    pub fn setup() -> Result<(TcpListener, ThreadPool), String> {
239        let info = Log::info("Rust Web Server");
240        println!("{}", info);
241
242        let usage_info = Log::usage_information();
243        println!("{}", usage_info);
244
245
246        println!("RWS Configuration Start: \n");
247
248        set_default_values();
249        bootstrap();
250
251        println!("\nRWS Configuration End\n\n");
252
253
254        let (ip, port, thread_count) = get_ip_port_thread_count();
255
256
257        let mut ip_readable = ip.to_string();
258
259        if ip.contains(":") {
260            ip_readable = [SYMBOL.opening_square_bracket, &ip, SYMBOL.closing_square_bracket].join("");
261        }
262
263        let bind_addr = [ip_readable, SYMBOL.colon.to_string(), port.to_string()].join(SYMBOL.empty_string);
264
265        #[cfg(feature = "http2")]
266        let protocol = {
267            let cert = std::env::var(crate::entry_point::Config::RWS_CONFIG_TLS_CERT_FILE).unwrap_or_default();
268            if cert.is_empty() { "http" } else { "https" }
269        };
270        #[cfg(not(feature = "http2"))]
271        let protocol = "http";
272
273        println!("Setting up {}://{}...", protocol, &bind_addr);
274
275        let boxed_listener = TcpListener::bind(&bind_addr);
276        if boxed_listener.is_err() {
277            let message = format!("unable to set up TCP listener: {}", boxed_listener.err().unwrap());
278            return Err(message);
279        }
280
281        let listener = boxed_listener.unwrap();
282        let pool = ThreadPool::new(thread_count as usize);
283
284
285        let server_url_thread_count = Log::server_url_thread_count(protocol, &bind_addr, thread_count);
286        println!("{}", server_url_thread_count);
287
288        Ok((listener, pool))
289    }
290
291    /// Accepts TCP connections in a loop and dispatches each to the thread pool.
292    ///
293    /// When built with the `http1` feature, Ctrl+C and SIGTERM stop the accept
294    /// loop gracefully: `SERVER_READY` is cleared and the pool drains all
295    /// in-flight connections before returning.
296    ///
297    /// For TLS/HTTP2/HTTP3 use [`Server::run_tls`].
298    pub fn run(listener: TcpListener,
299               pool: ThreadPool,
300               app: impl Application + Send + 'static + Clone) {
301        #[cfg(feature = "http1")]
302        {
303            use std::sync::Arc;
304            use std::sync::atomic::{AtomicBool, Ordering};
305
306            let shutdown = Arc::new(AtomicBool::new(false));
307            let s = shutdown.clone();
308            if let Err(e) = ctrlc::set_handler(move || {
309                s.store(true, Ordering::SeqCst);
310            }) {
311                eprintln!("unable to install signal handler: {}", e);
312            }
313            crate::config_reload::install_sighup_handler();
314            if let Err(e) = listener.set_nonblocking(true) {
315                eprintln!("unable to set non-blocking listener: {}", e);
316            }
317
318            loop {
319                if shutdown.load(Ordering::SeqCst) {
320                    break;
321                }
322                if crate::config_reload::RELOAD_REQUESTED
323                    .compare_exchange(true, false, Ordering::SeqCst, Ordering::Relaxed)
324                    .is_ok()
325                {
326                    crate::config_reload::reload();
327                }
328                match listener.accept() {
329                    Ok((stream, peer_addr)) => {
330                        Server::dispatch_connection(stream, peer_addr, &pool, app.clone());
331                    }
332                    Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
333                        std::thread::sleep(Duration::from_millis(10));
334                    }
335                    Err(e) => {
336                        eprintln!("accept error: {}", e);
337                        break;
338                    }
339                }
340            }
341
342            crate::metrics::SERVER_READY.store(false, std::sync::atomic::Ordering::SeqCst);
343            println!("Shutting down — waiting for in-flight connections to finish");
344            pool.join();
345            println!("Server stopped");
346        }
347
348        #[cfg(not(feature = "http1"))]
349        {
350            for boxed_stream in listener.incoming() {
351                match boxed_stream {
352                    Err(e) => {
353                        eprintln!("unable to get TCP stream: {}", e);
354                        return;
355                    }
356                    Ok(stream) => {
357                        let peer_addr = match stream.peer_addr() {
358                            Ok(a) => a,
359                            Err(e) => {
360                                eprintln!("unable to read peer addr: {}", e);
361                                return;
362                            }
363                        };
364                        Server::dispatch_connection(stream, peer_addr, &pool, app.clone());
365                    }
366                }
367            }
368        }
369    }
370
371    fn dispatch_connection(
372        stream: std::net::TcpStream,
373        peer_addr: std::net::SocketAddr,
374        pool: &ThreadPool,
375        app: impl Application + Send + 'static + Clone,
376    ) {
377        print!("Connection established, ");
378        if let Ok(local) = stream.local_addr() {
379            print!("local addr: {}", local);
380        }
381        println!(", peer addr: {}", peer_addr);
382
383        let (server_ip, server_port, _thread_count) = get_ip_port_thread_count();
384        let connection = ConnectionInfo {
385            client: Address {
386                ip: peer_addr.ip().to_string(),
387                port: peer_addr.port() as i32,
388            },
389            server: Address {
390                ip: server_ip,
391                port: server_port,
392            },
393            request_size: get_request_allocation_size(),
394            sni_hostname: None,
395        };
396
397        if let Err(e) = stream.set_read_timeout(Some(Duration::from_secs(30))) {
398            eprintln!("failed to set read timeout: {}", e);
399        }
400
401        pool.execute(move || {
402            crate::metrics::connection_open();
403            let result = Server::process(stream, connection, app);
404            crate::metrics::connection_close();
405            if let Err(msg) = result {
406                crate::metrics::record_error();
407                eprintln!("{}", msg);
408            }
409        });
410    }
411
412}
413
414/// Network context for the current connection, passed into every [`Controller`](crate::controller::Controller).
415#[derive(Clone)]
416pub struct ConnectionInfo {
417    /// Client (peer) address.
418    pub client: Address,
419    /// Server (local) address.
420    pub server: Address,
421    /// Bytes allocated for reading the request.
422    pub request_size: i64,
423    /// SNI hostname sent by the client during the TLS handshake, if any.
424    /// `None` for plain-HTTP connections or when the client omits SNI.
425    pub sni_hostname: Option<String>,
426}
427
428/// IP address and port pair.
429#[derive(Clone)]
430pub struct Address {
431    pub ip: String,
432    pub port: i32
433}
434
435impl ConnectionInfo {
436    /// Parse the client address into a [`std::net::SocketAddr`], if the stored
437    /// IP and port are valid. Returns `None` if parsing fails.
438    pub fn peer_addr(&self) -> Option<std::net::SocketAddr> {
439        self.client.to_socket_addr()
440    }
441}
442
443impl Address {
444    /// Parse this address into a [`std::net::SocketAddr`]. Returns `None` if
445    /// the IP string or port value cannot be converted.
446    pub fn to_socket_addr(&self) -> Option<std::net::SocketAddr> {
447        let ip: std::net::IpAddr = self.ip.parse().ok()?;
448        let port = u16::try_from(self.port).ok()?;
449        Some(std::net::SocketAddr::new(ip, port))
450    }
451}
452
453/// Resolves when SIGTERM is received on Unix, or never on other platforms.
454/// Enables a single `select!` branch to handle both SIGTERM and Ctrl+C.
455#[cfg(feature = "http2")]
456async fn sigterm() {
457    #[cfg(unix)]
458    {
459        if let Ok(mut s) = tokio::signal::unix::signal(
460            tokio::signal::unix::SignalKind::terminate()
461        ) {
462            s.recv().await;
463        } else {
464            std::future::pending::<()>().await
465        }
466    }
467    #[cfg(not(unix))]
468    std::future::pending::<()>().await
469}
470
471/// Returns a stream that fires on each SIGHUP on Unix; never fires elsewhere.
472#[cfg(feature = "http2")]
473async fn sighup() {
474    #[cfg(unix)]
475    {
476        if let Ok(mut s) = tokio::signal::unix::signal(
477            tokio::signal::unix::SignalKind::hangup()
478        ) {
479            s.recv().await;
480        } else {
481            std::future::pending::<()>().await
482        }
483    }
484    #[cfg(not(unix))]
485    std::future::pending::<()>().await
486}
487
488#[cfg(feature = "http2")]
489impl Server {
490    pub async fn run_tls(
491        listener: TcpListener,
492        pool: ThreadPool,
493        app: impl Application + Send + 'static + Clone,
494    ) {
495        use crate::tls::create_tls_acceptor_from_vhosts;
496        use crate::h2_handler;
497
498        let cert_path = std::env::var(crate::entry_point::Config::RWS_CONFIG_TLS_CERT_FILE)
499            .unwrap_or_default();
500        let key_path = std::env::var(crate::entry_point::Config::RWS_CONFIG_TLS_KEY_FILE)
501            .unwrap_or_default();
502
503        if cert_path.is_empty() || key_path.is_empty() {
504            println!("No TLS certificate configured — serving plain HTTP/1.1.");
505            tokio::task::block_in_place(|| Server::run(listener, pool, app));
506            return;
507        }
508
509        let vhosts = crate::entry_point::get_virtual_hosts();
510        let mut tls_acceptor = match create_tls_acceptor_from_vhosts(&vhosts, &cert_path, &key_path) {
511            Ok(a) => a,
512            Err(e) => {
513                eprintln!("TLS setup failed: {}", e);
514                return;
515            }
516        };
517
518        listener
519            .set_nonblocking(true)
520            .expect("failed to set TCP listener to non-blocking");
521        let tokio_listener = tokio::net::TcpListener::from_std(listener)
522            .expect("failed to convert TCP listener to tokio");
523
524        println!("Listening for TLS connections (HTTP/1.1 + HTTP/2)...");
525
526        loop {
527            tokio::select! {
528                result = tokio_listener.accept() => {
529                    match result {
530                        Ok((tcp_stream, peer_addr)) => {
531                            let acceptor = tls_acceptor.clone();
532                            let app = app.clone();
533                            tokio::spawn(async move {
534                                match acceptor.accept(tcp_stream).await {
535                                    Ok(tls_stream) => {
536                                        let server_conn = tls_stream.get_ref().1;
537                                        let sni = server_conn.server_name().map(|s| s.to_string());
538                                        let protocol = server_conn
539                                            .alpn_protocol()
540                                            .map(|p| p.to_vec());
541
542                                        match protocol.as_deref() {
543                                            Some(b"h2") => {
544                                                if let Err(e) =
545                                                    h2_handler::handle_connection(tls_stream, peer_addr, sni, app)
546                                                        .await
547                                                {
548                                                    eprintln!("H2 connection error: {}", e);
549                                                }
550                                            }
551                                            _ => {
552                                                if let Err(e) =
553                                                    Server::process_h1_tls(tls_stream, peer_addr, sni, app).await
554                                                {
555                                                    eprintln!("H1 TLS error: {}", e);
556                                                }
557                                            }
558                                        }
559                                    }
560                                    Err(e) => eprintln!("TLS handshake failed: {}", e),
561                                }
562                            });
563                        }
564                        Err(e) => eprintln!("TCP accept error: {}", e),
565                    }
566                }
567                _ = tokio::signal::ctrl_c() => {
568                    crate::metrics::SERVER_READY.store(false, std::sync::atomic::Ordering::SeqCst);
569                    println!("\nShutting down gracefully (SIGINT).");
570                    break;
571                }
572                _ = sigterm() => {
573                    crate::metrics::SERVER_READY.store(false, std::sync::atomic::Ordering::SeqCst);
574                    println!("\nShutting down gracefully (SIGTERM).");
575                    break;
576                }
577                _ = sighup() => {
578                    crate::config_reload::reload();
579                    let vhosts = crate::entry_point::get_virtual_hosts();
580                    if let Ok(new_acceptor) = create_tls_acceptor_from_vhosts(&vhosts, &cert_path, &key_path) {
581                        tls_acceptor = new_acceptor;
582                        println!("[TLS] Certificates reloaded ({} virtual hosts).", vhosts.len());
583                    }
584                }
585            }
586        }
587    }
588
589    /// Binds a plain-HTTP listener on the port in `RWS_CONFIG_HTTP_REDIRECT_PORT` and sends
590    /// `301 Moved Permanently` to the HTTPS equivalent of every incoming URL.
591    /// Returns immediately if TLS is not configured or the redirect port is not set.
592    pub async fn run_redirect() {
593        use std::env;
594        use tokio::io::{AsyncReadExt, AsyncWriteExt};
595        use tokio::net::TcpListener as TokioListener;
596
597        let cert_path = env::var(crate::entry_point::Config::RWS_CONFIG_TLS_CERT_FILE)
598            .unwrap_or_default();
599        if cert_path.is_empty() {
600            return;
601        }
602
603        let redirect_port_str = env::var(crate::entry_point::Config::RWS_CONFIG_HTTP_REDIRECT_PORT)
604            .unwrap_or_default();
605        if redirect_port_str.is_empty() {
606            return;
607        }
608
609        let redirect_port: u16 = match redirect_port_str.parse() {
610            Ok(p) => p,
611            Err(_) => {
612                eprintln!("Invalid RWS_CONFIG_HTTP_REDIRECT_PORT: {}", redirect_port_str);
613                return;
614            }
615        };
616
617        let (server_ip, server_port, _) = get_ip_port_thread_count();
618        let bind_addr = format!("{}:{}", server_ip, redirect_port);
619
620        let listener = match TokioListener::bind(&bind_addr).await {
621            Ok(l) => l,
622            Err(e) => {
623                eprintln!("HTTP redirect listener error on {}: {}", bind_addr, e);
624                return;
625            }
626        };
627
628        println!("HTTP→HTTPS redirect listening on http://{}:{}", server_ip, redirect_port);
629
630        loop {
631            tokio::select! {
632                result = listener.accept() => {
633                    match result {
634                        Ok((mut stream, _peer)) => {
635                            let https_port = server_port;
636                            tokio::spawn(async move {
637                                let mut buf = vec![0u8; 4096];
638                                let n = match stream.read(&mut buf).await {
639                                    Ok(n) => n,
640                                    Err(_) => return,
641                                };
642                                let text = String::from_utf8_lossy(&buf[..n]);
643
644                                let uri = text.lines()
645                                    .next()
646                                    .and_then(|line| line.split_whitespace().nth(1))
647                                    .unwrap_or("/")
648                                    .to_string();
649
650                                let host_header = text.lines()
651                                    .find(|l| l.to_lowercase().starts_with("host:"))
652                                    .map(|l| l[5..].trim().to_string());
653
654                                let location = match host_header {
655                                    Some(h) => {
656                                        // strip existing port from Host header
657                                        let h_no_port = if h.starts_with('[') {
658                                            // IPv6: [::1] or [::1]:port
659                                            h.find(']')
660                                                .map(|i| h[..=i].to_string())
661                                                .unwrap_or(h.clone())
662                                        } else {
663                                            h.rfind(':')
664                                                .map(|i| h[..i].to_string())
665                                                .unwrap_or(h.clone())
666                                        };
667                                        if https_port == 443 {
668                                            format!("https://{}{}", h_no_port, uri)
669                                        } else {
670                                            format!("https://{}:{}{}", h_no_port, https_port, uri)
671                                        }
672                                    }
673                                    None => format!("https://localhost:{}{}", https_port, uri),
674                                };
675
676                                let response = format!(
677                                    "HTTP/1.1 301 Moved Permanently\r\nLocation: {}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
678                                    location
679                                );
680                                let _ = stream.write_all(response.as_bytes()).await;
681                            });
682                        }
683                        Err(e) => eprintln!("HTTP redirect accept error: {}", e),
684                    }
685                }
686                _ = tokio::signal::ctrl_c() => {
687                    println!("\nShutting down HTTP redirect listener (SIGINT).");
688                    break;
689                }
690                _ = sigterm() => {
691                    println!("\nShutting down HTTP redirect listener (SIGTERM).");
692                    break;
693                }
694                _ = sighup() => {
695                    crate::config_reload::reload();
696                }
697            }
698        }
699    }
700
701    async fn process_h1_tls(
702        mut stream: tokio_rustls::server::TlsStream<tokio::net::TcpStream>,
703        peer_addr: std::net::SocketAddr,
704        sni_hostname: Option<String>,
705        app: impl Application,
706    ) -> Result<(), String> {
707        use tokio::io::{AsyncReadExt, AsyncWriteExt};
708
709        let (server_ip, server_port, _) = get_ip_port_thread_count();
710        let request_allocation_size = get_request_allocation_size();
711
712        let mut buffer = vec![0u8; request_allocation_size as usize];
713        if let Err(e) = stream.read(&mut buffer).await {
714            let raw = Server::bad_request_response(e.to_string());
715            let _ = stream.write_all(&raw).await;
716            return Ok(());
717        }
718
719        let request = match Request::parse(&buffer) {
720            Ok(r) => r,
721            Err(message) => {
722                let raw = Server::bad_request_response(message);
723                let _ = stream.write_all(&raw).await;
724                return Ok(());
725            }
726        };
727
728        let connection = ConnectionInfo {
729            client: Address {
730                ip: peer_addr.ip().to_string(),
731                port: peer_addr.port() as i32,
732            },
733            server: Address {
734                ip: server_ip,
735                port: server_port,
736            },
737            request_size: request_allocation_size,
738            sni_hostname,
739        };
740
741        let mut response = match app.execute(&request, &connection) {
742            Ok(r) => r,
743            Err(message) => {
744                let raw = Server::bad_request_response(message);
745                let _ = stream.write_all(&raw).await;
746                return Ok(());
747            }
748        };
749
750        crate::metrics::record_request();
751        crate::compression::apply_gzip(&request, &mut response);
752        response.headers.push(Header::get_hsts_header());
753
754        #[cfg(feature = "http3")]
755        response.headers.push(Header {
756            name: Header::_ALT_SVC.to_string(),
757            value: format!("h3=\":{}\"", server_port),
758        });
759        #[cfg(not(feature = "http3"))]
760        response.headers.push(Header {
761            name: Header::_ALT_SVC.to_string(),
762            value: format!("h2=\":{}\"", server_port),
763        });
764
765        Log::log_access(&request, &response, &peer_addr);
766
767        let raw = Response::generate_response(response, request);
768        stream
769            .write_all(&raw)
770            .await
771            .map_err(|e| e.to_string())?;
772        stream.flush().await.map_err(|e| e.to_string())?;
773
774        Ok(())
775    }
776}
777
778#[cfg(feature = "http3")]
779impl Server {
780    pub async fn run_quic(
781        app: impl Application + Send + 'static + Clone,
782    ) {
783        use crate::tls::create_quinn_server_config_from_vhosts;
784        use crate::h3_handler;
785
786        let cert_path = std::env::var(crate::entry_point::Config::RWS_CONFIG_TLS_CERT_FILE)
787            .unwrap_or_default();
788        let key_path = std::env::var(crate::entry_point::Config::RWS_CONFIG_TLS_KEY_FILE)
789            .unwrap_or_default();
790
791        if cert_path.is_empty() || key_path.is_empty() {
792            return;
793        }
794
795        let vhosts = crate::entry_point::get_virtual_hosts();
796        let server_config = match create_quinn_server_config_from_vhosts(&vhosts, &cert_path, &key_path) {
797            Ok(c) => c,
798            Err(e) => {
799                eprintln!("QUIC TLS setup failed: {}", e);
800                return;
801            }
802        };
803
804        let (server_ip, server_port, _) = get_ip_port_thread_count();
805        let bind_addr = format!("{}:{}", server_ip, server_port);
806        let addr: std::net::SocketAddr = match bind_addr.parse() {
807            Ok(a) => a,
808            Err(e) => {
809                eprintln!("Invalid QUIC bind address '{}': {}", bind_addr, e);
810                return;
811            }
812        };
813
814        let endpoint = match quinn::Endpoint::server(server_config, addr) {
815            Ok(e) => e,
816            Err(e) => {
817                eprintln!("QUIC endpoint error: {}", e);
818                return;
819            }
820        };
821
822        println!("Listening for QUIC/HTTP3 on UDP {}:{}", server_ip, server_port);
823
824        loop {
825            tokio::select! {
826                maybe = endpoint.accept() => {
827                    match maybe {
828                        Some(incoming) => {
829                            let app = app.clone();
830                            tokio::spawn(async move {
831                                match incoming.await {
832                                    Ok(conn) => {
833                                        let peer_addr = conn.remote_address();
834                                        if let Err(e) = h3_handler::handle_connection(conn, peer_addr, app).await {
835                                            eprintln!("H3 connection error: {}", e);
836                                        }
837                                    }
838                                    Err(e) => eprintln!("QUIC connection error: {}", e),
839                                }
840                            });
841                        }
842                        None => break,
843                    }
844                }
845                _ = tokio::signal::ctrl_c() => {
846                    crate::metrics::SERVER_READY.store(false, std::sync::atomic::Ordering::SeqCst);
847                    println!("\nShutting down QUIC (SIGINT).");
848                    endpoint.close(0u32.into(), b"shutdown");
849                    break;
850                }
851                _ = sigterm() => {
852                    crate::metrics::SERVER_READY.store(false, std::sync::atomic::Ordering::SeqCst);
853                    println!("\nShutting down QUIC (SIGTERM).");
854                    endpoint.close(0u32.into(), b"shutdown");
855                    break;
856                }
857                _ = sighup() => {
858                    crate::config_reload::reload();
859                }
860            }
861        }
862    }
863}
864
865
rust_web_server/server/mod.rs

rust_web_server/server/
mod.rs
