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

rust_web_server/server/
mod.rs
