diff --git a/src/http.rs b/src/http.rs index 1379016..0e401f4 100644 --- a/src/http.rs +++ b/src/http.rs @@ -1,9 +1,5 @@ use std::{ - collections::HashMap, - io::{BufRead, BufReader, Error, ErrorKind, Result, Write}, - net::{IpAddr, SocketAddr, TcpListener, TcpStream}, - str::FromStr, - sync::Arc, + collections::HashMap, io::{BufRead, BufReader, Error, ErrorKind, Result, Write}, net::{IpAddr, SocketAddr, TcpListener, TcpStream}, str::FromStr, sync::Arc, thread, }; use chrono::Local; @@ -35,27 +31,28 @@ impl StatusCode { StatusCode::OK => "OK", StatusCode::BadRequest => "Bad Request", StatusCode::NotFound => "Not Found", - // StatusCode::ImATeapot => "I'm a teapot", + StatusCode::ImATeapot => "I'm a teapot", StatusCode::InternalServerError => "Internal Server Error", + StatusCode::BadGateway => "Bad Gateway", } } } type HttpHandlerFunc = fn(&Request, Response, bool) -> Result; -pub struct Request<'a> { - stream: &'a TcpStream, - path: &'a str, - method: &'a str, - version: &'a str, +pub struct Request<'r> { + stream: &'r TcpStream, + path: &'r str, + method: &'r str, + version: &'r str, headers: HashMap, query: HashMap, body: Option, real_address: IpAddr, } -impl<'a> Request<'a> { - pub fn new(stream: &'a TcpStream, lines: &'a Vec, trusted_proxies: Vec) -> Result> { +impl<'r> Request<'r> { + pub fn new(stream: &'r TcpStream, lines: &'r Vec, trusted_proxies: Vec) -> Result> { let request_line = lines[0].as_str(); let request_line_split: Vec<&str> = request_line.split(" ").collect(); if request_line_split.len() < 3 { @@ -139,13 +136,13 @@ impl<'a> Request<'a> { pub fn real_address(&self) -> &IpAddr { &self.real_address } - pub fn path(&self) -> &'a str { + pub fn path(&self) -> &'r str { self.path } - pub fn method(&self) -> &'a str { + pub fn method(&self) -> &'r str { self.method } - pub fn version(&self) -> &'a str { + pub fn version(&self) -> &'r str { self.version } pub fn body(&self) -> &Option { @@ -159,15 +156,15 @@ impl<'a> Request<'a> { } } -pub struct Response<'a> { - stream: &'a TcpStream, +pub struct Response<'r> { + stream: &'r TcpStream, status: StatusCode, - headers: HashMap<&'a str, String>, + headers: HashMap<&'r str, String>, body: Option, } -impl<'a> Response<'a> { - pub fn new(stream: &'a TcpStream) -> Response<'a> { +impl<'r> Response<'r> { + pub fn new(stream: &'r TcpStream) -> Response<'r> { Response { stream, status: StatusCode::OK, @@ -182,10 +179,10 @@ impl<'a> Response<'a> { pub fn status(&mut self, status: StatusCode) { self.status = status; } - pub fn headers(&self) -> &HashMap<&'a str, String> { + pub fn headers(&self) -> &HashMap<&'r str, String> { &self.headers } - pub fn set_header(&mut self, name: &'a str, value: String) { + pub fn set_header(&mut self, name: &'r str, value: String) { self.headers.insert(name, value); } pub fn body(&mut self, body: String) { @@ -213,16 +210,16 @@ impl<'a> Response<'a> { } } -pub struct HttpServer<'a> { - address: &'a str, +pub struct HttpServer<'s> { + address: &'s str, port: u16, trusted_proxies: Arc>, max_connections: usize, verbose: bool, } -impl HttpServer <'_> { - pub fn new(address: &'_ str, max_connections: usize, trusted_proxies: Vec, verbose: bool) -> HttpServer<'_> { +impl<'s> HttpServer <'s> { + pub fn new(address: &'s str, max_connections: usize, trusted_proxies: Vec, verbose: bool) -> HttpServer<'s> { let mut _address = address; let mut _port: u16 = 8080; match address.split_once(":") { @@ -242,25 +239,27 @@ impl HttpServer <'_> { } pub fn start(&self, handler: HttpHandlerFunc) -> Result<()> { - let pool = ThreadPool::new(self.max_connections); - let listener = TcpListener::bind(format!("{}:{}", self.address, self.port)).expect("Failed to bind to port"); - let verbose = self.verbose; + thread::scope(|scope| { + let pool = ThreadPool::new(scope, self.max_connections); + let listener = TcpListener::bind(format!("{}:{}", self.address, self.port)).expect("Failed to bind to port"); + let verbose = self.verbose; - println!("Now listening on {}:{}", self.address, self.port); + println!("Now listening on {}:{}", self.address, self.port); - for stream in listener.incoming() { - match stream { - Ok(stream) => { - let trusted_proxies = self.trusted_proxies.clone(); - pool.execute(move || { - HttpServer::handle_client(&stream, handler, trusted_proxies, verbose); - }); - } - Err(e) => { - eprintln!("Failed to handle incoming connection: {e}"); + for stream in listener.incoming() { + match stream { + Ok(stream) => { + let trusted_proxies = self.trusted_proxies.clone(); + pool.execute(move || { + HttpServer::handle_client(&stream, handler, trusted_proxies, verbose); + }); + } + Err(e) => { + eprintln!("Failed to handle incoming connection: {e}"); + } } } - } + }); Ok(()) } diff --git a/src/thread.rs b/src/thread.rs index a6cae8f..c4332d8 100644 --- a/src/thread.rs +++ b/src/thread.rs @@ -1,19 +1,19 @@ -use std::{sync::{mpsc, Arc, Mutex}, thread}; +use std::{marker::PhantomData, sync::{Arc, Mutex, mpsc}, thread}; -pub struct ThreadPool { - workers: Vec, - sender: Option>, +pub struct ThreadPool<'s> { + workers: Vec>, + sender: mpsc::Sender>, } -type Job = Box; +type Job<'s> = Box; -impl ThreadPool { +impl<'s> ThreadPool<'s> { // Create a new ThreadPool with `size` available threads. // // # Panics // // `new` will panic if `size` is zero. - pub fn new(size: usize) -> ThreadPool { + pub fn new(scope: &'s thread::Scope<'s, '_>, size: usize) -> ThreadPool<'s> { assert!(size > 0); let (sender, receiver) = mpsc::channel(); @@ -23,59 +23,64 @@ impl ThreadPool { let mut workers = Vec::with_capacity(size); for id in 0..size { - workers.push(ThreadWorker::new(id, Arc::clone(&receiver))); + let worker = ThreadWorker::new(id, receiver.clone(), scope); + workers.push(worker); } ThreadPool { workers, - sender: Some(sender) + sender: sender, } } pub fn execute(&self, f: F) where - F: FnOnce() + Send + 'static + F: FnOnce() + Send + 's { let job = Box::new(f); - self.sender.as_ref().unwrap().send(job).unwrap(); + self.sender.send(job).expect("Failed to execute worker job") } } -impl Drop for ThreadPool { +impl<'s> Drop for ThreadPool<'s> { fn drop(&mut self) { - drop(self.sender.take()); - - for worker in &mut self.workers.drain(..) { - worker.thread.join().expect( - format!("Error in worker {}", worker.id).as_str()); + for worker in self.workers.drain(..) { + worker.close(); } } } -struct ThreadWorker { - id: usize, - thread: thread::JoinHandle<()>, +struct ThreadWorker<'s> { + _lifetime: PhantomData<&'s ()>, + _id: usize, + thread: thread::ScopedJoinHandle<'s, ()>, } -impl ThreadWorker { - fn new(id: usize, receiver: Arc>>) -> ThreadWorker { - let thread = thread::spawn(move || loop { - let msg = receiver.lock().unwrap().recv(); +impl<'s> ThreadWorker<'s> { + fn new(_id: usize, receiver: Arc>>>, scope: &'s thread::Scope<'s, '_>) -> ThreadWorker<'s> { + let thread = scope.spawn(move || loop { + let receiver_locked = receiver.lock().expect("Failed to acquire lock"); + let msg = receiver_locked.recv(); match msg { Ok(job) => { - // println!("Job received by worker {id}"); + // println!("Job received by worker {_id}"); job(); } Err(_) => { - // println!("Worker {id} disconnected. Shutting down..."); + // println!("Worker {_id} disconnected. Shutting down..."); break; } } }); - ThreadWorker { id, thread } - } -} + ThreadWorker { _lifetime: PhantomData, _id, thread } + } + fn close(self) { + self.thread.join().unwrap_or_else(|err| { + println!("Failed to properly close thread: {err:?}") + }) + } +}