async rewrite x2
This commit is contained in:
parent
0d5dd9281b
commit
5060e811a1
15
Cargo.lock
generated
15
Cargo.lock
generated
@ -87,6 +87,7 @@ dependencies = [
|
||||
"rusty_pool",
|
||||
"serde_json",
|
||||
"tokio",
|
||||
"tokio-io-timeout",
|
||||
"urlencoding",
|
||||
]
|
||||
|
||||
@ -374,9 +375,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "serde_json"
|
||||
version = "1.0.125"
|
||||
version = "1.0.127"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "83c8e735a073ccf5be70aa8066aa984eaf2fa000db6c8d0100ae605b366d31ed"
|
||||
checksum = "8043c06d9f82bd7271361ed64f415fe5e12a77fdb52e573e7f06a516dea329ad"
|
||||
dependencies = [
|
||||
"itoa",
|
||||
"memchr",
|
||||
@ -453,6 +454,16 @@ dependencies = [
|
||||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-io-timeout"
|
||||
version = "1.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "30b74022ada614a1b4834de765f9bb43877f910cc8ce4be40e89042c9223a8bf"
|
||||
dependencies = [
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-macros"
|
||||
version = "2.4.0"
|
||||
|
||||
@ -11,6 +11,10 @@ keywords = ["http", "server", "site", "async"]
|
||||
|
||||
[dependencies]
|
||||
urlencoding = "2.1.3"
|
||||
serde_json = "1.0.125"
|
||||
serde_json = "1.0.127"
|
||||
tokio = { version = "1.39.3", features = ["full"] }
|
||||
rusty_pool = "0.7.0"
|
||||
rusty_pool = "0.7.0"
|
||||
tokio-io-timeout = "1.2.0"
|
||||
|
||||
[features]
|
||||
http_rrs = []
|
||||
@ -13,6 +13,7 @@ pub enum HttpError {
|
||||
WriteHeadError,
|
||||
WriteBodyError,
|
||||
InvalidStatus,
|
||||
RequstError
|
||||
}
|
||||
|
||||
impl std::fmt::Display for HttpError {
|
||||
|
||||
86
src/ezhttp/handler.rs
Normal file
86
src/ezhttp/handler.rs
Normal file
@ -0,0 +1,86 @@
|
||||
use super::{HttpError, HttpRequest, HttpServer, Stream};
|
||||
|
||||
use std::{future::Future, pin::Pin, sync::Arc};
|
||||
use tokio::{net::TcpStream, sync::Mutex};
|
||||
use tokio_io_timeout::TimeoutStream;
|
||||
|
||||
#[cfg(feature = "http_rrs")]
|
||||
use {super::read_line_lf, std::net::{ToSocketAddrs, SocketAddr}};
|
||||
|
||||
pub type Handler<T> = Box<dyn Fn(Arc<Mutex<T>>, TimeoutStream<TcpStream>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;
|
||||
|
||||
|
||||
/// Default connection handler
|
||||
/// Turns input to request and response to output
|
||||
pub async fn handler_connection<S: HttpServer + Send + 'static>(
|
||||
server: Arc<Mutex<S>>,
|
||||
mut sock: Stream
|
||||
) {
|
||||
let Ok(addr) = sock.get_ref().peer_addr() else { return; };
|
||||
|
||||
let req = match HttpRequest::read(sock.get_mut(), &addr).await {
|
||||
Ok(i) => i,
|
||||
Err(e) => {
|
||||
server.lock().await.on_error(e).await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let resp = match server.lock().await.on_request(&req).await {
|
||||
Some(i) => i,
|
||||
None => {
|
||||
server.lock().await.on_error(HttpError::RequstError).await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
match resp.write(sock.get_mut()).await {
|
||||
Ok(_) => {},
|
||||
Err(e) => {
|
||||
server.lock().await.on_error(e).await;
|
||||
return;
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "http_rrs")]
|
||||
/// HTTP_RRS handler
|
||||
pub async fn handler_http_rrs<S: HttpServer + Send + 'static>(
|
||||
server: Arc<Mutex<S>>,
|
||||
mut sock: Stream,
|
||||
) {
|
||||
let addr = match read_line_lf(sock.get_mut()).await {
|
||||
Ok(i) => i,
|
||||
Err(e) => {
|
||||
server.lock().await.on_error(e).await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
.to_socket_addrs()
|
||||
.unwrap()
|
||||
.collect::<Vec<SocketAddr>>()[0];
|
||||
|
||||
let req = match HttpRequest::read(sock.get_mut(), &addr).await {
|
||||
Ok(i) => i,
|
||||
Err(e) => {
|
||||
server.lock().await.on_error(e).await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let resp = match server.lock().await.on_request(&req).await {
|
||||
Some(i) => i,
|
||||
None => {
|
||||
server.lock().await.on_error(HttpError::RequstError).await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
match resp.write(sock.get_mut()).await {
|
||||
Ok(_) => {},
|
||||
Err(e) => {
|
||||
server.lock().await.on_error(e).await;
|
||||
return;
|
||||
},
|
||||
}
|
||||
}
|
||||
@ -1,94 +1,64 @@
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::{
|
||||
boxed::Box,
|
||||
error::Error,
|
||||
future::Future,
|
||||
io::Read,
|
||||
net::{TcpListener, TcpStream},
|
||||
sync::{
|
||||
atomic::{AtomicBool, Ordering},
|
||||
Arc,
|
||||
},
|
||||
sync::Arc,
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use tokio::io::AsyncReadExt;
|
||||
use rusty_pool::ThreadPool;
|
||||
use tokio::net::{TcpListener, TcpStream};
|
||||
use tokio::sync::Mutex;
|
||||
use tokio_io_timeout::TimeoutStream;
|
||||
|
||||
pub mod error;
|
||||
pub mod headers;
|
||||
pub mod request;
|
||||
pub mod response;
|
||||
pub mod starter;
|
||||
pub mod handler;
|
||||
|
||||
pub use error::*;
|
||||
pub use headers::*;
|
||||
pub use request::*;
|
||||
pub use response::*;
|
||||
use rusty_pool::ThreadPool;
|
||||
pub use starter::*;
|
||||
use tokio::sync::Mutex;
|
||||
pub use handler::*;
|
||||
|
||||
fn read_line(data: &mut impl Read) -> Result<String, HttpError> {
|
||||
let mut bytes = Vec::new();
|
||||
|
||||
for byte in data.bytes() {
|
||||
let byte = match byte {
|
||||
Ok(i) => i,
|
||||
Err(_) => return Err(HttpError::ReadLineEof),
|
||||
};
|
||||
|
||||
bytes.push(byte);
|
||||
|
||||
if byte == 0x0A {
|
||||
async fn read_line(data: &mut (impl AsyncReadExt + Unpin)) -> Result<String, HttpError> {
|
||||
let mut line = Vec::new();
|
||||
loop {
|
||||
let mut buffer = vec![0;1];
|
||||
data.read_exact(&mut buffer).await.or(Err(HttpError::ReadLineEof))?;
|
||||
let char = buffer[0];
|
||||
line.push(char);
|
||||
if char == 0x0a {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
match String::from_utf8(bytes) {
|
||||
Ok(i) => Ok(i),
|
||||
Err(_) => Err(HttpError::ReadLineUnknown),
|
||||
}
|
||||
String::from_utf8(line).or(Err(HttpError::ReadLineUnknown))
|
||||
}
|
||||
|
||||
fn read_line_crlf(data: &mut impl Read) -> Result<String, HttpError> {
|
||||
match read_line(data) {
|
||||
async fn read_line_crlf(data: &mut (impl AsyncReadExt + Unpin)) -> Result<String, HttpError> {
|
||||
match read_line(data).await {
|
||||
Ok(i) => Ok(i[..i.len() - 2].to_string()),
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
fn read_line_lf(data: &mut impl Read) -> Result<String, HttpError> {
|
||||
match read_line(data) {
|
||||
#[cfg(feature = "http_rrs")]
|
||||
async fn read_line_lf(data: &mut (impl AsyncReadExt + Unpin)) -> Result<String, HttpError> {
|
||||
match read_line(data).await {
|
||||
Ok(i) => Ok(i[..i.len() - 1].to_string()),
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
fn rem_first(value: &str) -> &str {
|
||||
let mut chars = value.chars();
|
||||
chars.next();
|
||||
chars.as_str()
|
||||
}
|
||||
|
||||
fn split(text: String, delimiter: &str, times: usize) -> Vec<String> {
|
||||
match times {
|
||||
0 => text.split(delimiter).map(|v| v.to_string()).collect(),
|
||||
1 => {
|
||||
let mut v: Vec<String> = Vec::new();
|
||||
match text.split_once(delimiter) {
|
||||
Some(i) => {
|
||||
v.push(i.0.to_string());
|
||||
v.push(i.1.to_string());
|
||||
}
|
||||
None => {
|
||||
v.push(text);
|
||||
}
|
||||
}
|
||||
v
|
||||
}
|
||||
_ => text
|
||||
.splitn(times, delimiter)
|
||||
.map(|v| v.to_string())
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
pub type Stream = TimeoutStream<TcpStream>;
|
||||
|
||||
/// Async http server trait
|
||||
pub trait HttpServer {
|
||||
@ -98,45 +68,43 @@ pub trait HttpServer {
|
||||
&mut self,
|
||||
req: &HttpRequest,
|
||||
) -> impl Future<Output = Option<HttpResponse>> + Send;
|
||||
fn on_error(
|
||||
&mut self,
|
||||
_: HttpError
|
||||
) -> impl Future<Output = ()> + Send {
|
||||
async {}
|
||||
}
|
||||
}
|
||||
|
||||
async fn start_server_with_threadpool<S>(
|
||||
server: S,
|
||||
async fn start_server_with_threadpool<T>(
|
||||
server: T,
|
||||
host: &str,
|
||||
timeout: Option<Duration>,
|
||||
threads: usize,
|
||||
rrs: bool,
|
||||
handler: Handler<T>,
|
||||
running: Arc<AtomicBool>,
|
||||
) -> Result<(), Box<dyn Error>>
|
||||
where
|
||||
S: HttpServer + Send + 'static,
|
||||
T: HttpServer + Send + 'static,
|
||||
{
|
||||
let threadpool = ThreadPool::new(threads, threads * 10, Duration::from_secs(60));
|
||||
let server = Arc::new(Mutex::new(server));
|
||||
let listener = TcpListener::bind(host)?;
|
||||
let listener = TcpListener::bind(host).await?;
|
||||
|
||||
let host_clone = String::from(host).clone();
|
||||
let server_clone = server.clone();
|
||||
server_clone.lock().await.on_start(&host_clone).await;
|
||||
|
||||
while running.load(Ordering::Acquire) {
|
||||
let (sock, _) = match listener.accept() {
|
||||
Ok(i) => i,
|
||||
Err(_) => {
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let Ok((sock, _)) = listener.accept().await else { continue; };
|
||||
let mut sock = TimeoutStream::new(sock);
|
||||
|
||||
sock.set_read_timeout(timeout).unwrap();
|
||||
sock.set_write_timeout(timeout).unwrap();
|
||||
sock.set_read_timeout(timeout);
|
||||
sock.set_write_timeout(timeout);
|
||||
|
||||
let now_server = Arc::clone(&server);
|
||||
|
||||
if !rrs {
|
||||
threadpool.spawn(handle_connection(now_server, sock));
|
||||
} else {
|
||||
threadpool.spawn(handle_connection_rrs(now_server, sock));
|
||||
}
|
||||
threadpool.spawn((&handler)(now_server, sock));
|
||||
}
|
||||
|
||||
threadpool.join();
|
||||
@ -146,41 +114,33 @@ where
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn start_server_new_thread<S>(
|
||||
server: S,
|
||||
async fn start_server_new_thread<T>(
|
||||
server: T,
|
||||
host: &str,
|
||||
timeout: Option<Duration>,
|
||||
rrs: bool,
|
||||
handler: Handler<T>,
|
||||
running: Arc<AtomicBool>,
|
||||
) -> Result<(), Box<dyn Error>>
|
||||
where
|
||||
S: HttpServer + Send + 'static,
|
||||
T: HttpServer + Send + 'static,
|
||||
{
|
||||
let server = Arc::new(Mutex::new(server));
|
||||
let listener = TcpListener::bind(host)?;
|
||||
let listener = TcpListener::bind(host).await?;
|
||||
|
||||
let host_clone = String::from(host).clone();
|
||||
let server_clone = server.clone();
|
||||
server_clone.lock().await.on_start(&host_clone).await;
|
||||
|
||||
while running.load(Ordering::Acquire) {
|
||||
let (sock, _) = match listener.accept() {
|
||||
Ok(i) => i,
|
||||
Err(_) => {
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let Ok((sock, _)) = listener.accept().await else { continue; };
|
||||
let mut sock = TimeoutStream::new(sock);
|
||||
|
||||
sock.set_read_timeout(timeout).unwrap();
|
||||
sock.set_write_timeout(timeout).unwrap();
|
||||
sock.set_read_timeout(timeout);
|
||||
sock.set_write_timeout(timeout);
|
||||
|
||||
let now_server = Arc::clone(&server);
|
||||
|
||||
if !rrs {
|
||||
tokio::spawn(handle_connection(now_server, sock));
|
||||
} else {
|
||||
tokio::spawn(handle_connection_rrs(now_server, sock));
|
||||
}
|
||||
tokio::spawn((&handler)(now_server, sock));
|
||||
}
|
||||
|
||||
server.lock().await.on_close().await;
|
||||
@ -188,41 +148,33 @@ where
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn start_server_sync<S>(
|
||||
server: S,
|
||||
async fn start_server_sync<T>(
|
||||
server: T,
|
||||
host: &str,
|
||||
timeout: Option<Duration>,
|
||||
rrs: bool,
|
||||
handler: Handler<T>,
|
||||
running: Arc<AtomicBool>,
|
||||
) -> Result<(), Box<dyn Error>>
|
||||
where
|
||||
S: HttpServer + Send + 'static,
|
||||
T: HttpServer + Send + 'static,
|
||||
{
|
||||
let server = Arc::new(Mutex::new(server));
|
||||
let listener = TcpListener::bind(host)?;
|
||||
let listener = TcpListener::bind(host).await?;
|
||||
|
||||
let host_clone = String::from(host).clone();
|
||||
let server_clone = server.clone();
|
||||
server_clone.lock().await.on_start(&host_clone).await;
|
||||
|
||||
while running.load(Ordering::Acquire) {
|
||||
let (sock, _) = match listener.accept() {
|
||||
Ok(i) => i,
|
||||
Err(_) => {
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let Ok((sock, _)) = listener.accept().await else { continue; };
|
||||
let mut sock = TimeoutStream::new(sock);
|
||||
|
||||
sock.set_read_timeout(timeout).unwrap();
|
||||
sock.set_write_timeout(timeout).unwrap();
|
||||
sock.set_read_timeout(timeout);
|
||||
sock.set_write_timeout(timeout);
|
||||
|
||||
let now_server = Arc::clone(&server);
|
||||
|
||||
if !rrs {
|
||||
handle_connection(now_server, sock).await;
|
||||
} else {
|
||||
handle_connection_rrs(now_server, sock).await;
|
||||
}
|
||||
handler(now_server, sock).await;
|
||||
}
|
||||
|
||||
server.lock().await.on_close().await;
|
||||
@ -230,60 +182,18 @@ where
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn handle_connection<S: HttpServer + Send + 'static>(
|
||||
server: Arc<Mutex<S>>,
|
||||
mut sock: TcpStream
|
||||
) {
|
||||
let Ok(addr) = sock.peer_addr() else { return; };
|
||||
|
||||
let req = match HttpRequest::read(&mut sock, &addr) {
|
||||
Ok(i) => i,
|
||||
Err(_) => {
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let resp = match server.lock().await.on_request(&req).await {
|
||||
Some(i) => i,
|
||||
None => {
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let _ = resp.write(&mut sock);
|
||||
}
|
||||
|
||||
async fn handle_connection_rrs<S: HttpServer + Send + 'static>(
|
||||
server: Arc<Mutex<S>>,
|
||||
mut sock: TcpStream,
|
||||
) {
|
||||
let req = match HttpRequest::read_with_rrs(&mut sock) {
|
||||
Ok(i) => i,
|
||||
Err(_) => {
|
||||
return;
|
||||
}
|
||||
};
|
||||
let resp = match server.lock().await.on_request(&req).await {
|
||||
Some(i) => i,
|
||||
None => {
|
||||
return;
|
||||
}
|
||||
};
|
||||
let _ = resp.write(&mut sock);
|
||||
}
|
||||
|
||||
/// Start [`HttpServer`](HttpServer) on some host
|
||||
///
|
||||
/// Use [`HttpServerStarter`](HttpServerStarter) to set more options
|
||||
pub async fn start_server<S: HttpServer + Send + 'static>(
|
||||
server: S,
|
||||
pub async fn start_server<T: HttpServer + Send + 'static>(
|
||||
server: T,
|
||||
host: &str
|
||||
) -> Result<(), Box<dyn Error>> {
|
||||
start_server_new_thread(
|
||||
server,
|
||||
host,
|
||||
None,
|
||||
false,
|
||||
Box::new(move |a, b| Box::pin(handler_connection(a, b))),
|
||||
Arc::new(AtomicBool::new(true)),
|
||||
).await
|
||||
}
|
||||
|
||||
@ -1,11 +1,11 @@
|
||||
use super::{read_line_crlf, read_line_lf, rem_first, split, Headers, HttpError};
|
||||
use super::{read_line_crlf, Headers, HttpError};
|
||||
|
||||
use serde_json::Value;
|
||||
use std::{
|
||||
fmt::{Debug, Display},
|
||||
io::{Read, Write},
|
||||
net::{IpAddr, SocketAddr, ToSocketAddrs},
|
||||
net::SocketAddr,
|
||||
};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
/// Http request
|
||||
#[derive(Debug, Clone)]
|
||||
@ -38,39 +38,28 @@ impl HttpRequest {
|
||||
}
|
||||
|
||||
/// Read http request from stream
|
||||
pub fn read(data: &mut impl Read, addr: &SocketAddr) -> Result<HttpRequest, HttpError> {
|
||||
let octets = match addr.ip() {
|
||||
IpAddr::V4(ip) => ip.octets(),
|
||||
_ => [127, 0, 0, 1],
|
||||
};
|
||||
pub async fn read(data: &mut (impl AsyncReadExt + Unpin), addr: &SocketAddr) -> Result<HttpRequest, HttpError> {
|
||||
let ip_str = addr.to_string();
|
||||
|
||||
let ip_str = octets[0].to_string()
|
||||
+ "."
|
||||
+ &octets[1].to_string()
|
||||
+ "."
|
||||
+ &octets[2].to_string()
|
||||
+ "."
|
||||
+ &octets[3].to_string();
|
||||
|
||||
let status = split(
|
||||
match read_line_crlf(data) {
|
||||
Ok(i) => i,
|
||||
Err(e) => return Err(e),
|
||||
let status: Vec<String> = match read_line_crlf(data).await {
|
||||
Ok(i) => {
|
||||
i.splitn(3, " ")
|
||||
.map(|s| s.to_string())
|
||||
.collect()
|
||||
},
|
||||
" ",
|
||||
3,
|
||||
);
|
||||
Err(e) => return Err(e),
|
||||
};
|
||||
|
||||
let method = status[0].clone();
|
||||
let (page, query) = match status[1].split_once("?") {
|
||||
Some(i) => (i.0.to_string(), Some(i.1)),
|
||||
None => (status[1].clone(), None),
|
||||
None => (status[1].to_string(), None),
|
||||
};
|
||||
|
||||
let mut headers = Headers::new();
|
||||
|
||||
loop {
|
||||
let text = match read_line_crlf(data) {
|
||||
let text = match read_line_crlf(data).await {
|
||||
Ok(i) => i,
|
||||
Err(_) => return Err(HttpError::InvalidHeaders),
|
||||
};
|
||||
@ -121,7 +110,7 @@ impl HttpRequest {
|
||||
let mut buf: Vec<u8> = Vec::new();
|
||||
buf.resize(content_size - reqdata.len(), 0);
|
||||
|
||||
match data.read_exact(&mut buf) {
|
||||
match data.read_exact(&mut buf).await {
|
||||
Ok(i) => i,
|
||||
Err(_) => return Err(HttpError::InvalidContent),
|
||||
};
|
||||
@ -166,7 +155,7 @@ impl HttpRequest {
|
||||
}
|
||||
"application/x-www-form-urlencoded" => {
|
||||
if body.starts_with("?") {
|
||||
body = rem_first(body.as_str()).to_string()
|
||||
body = body.as_str()[1..].to_string()
|
||||
}
|
||||
|
||||
for ele in body.split("&") {
|
||||
@ -201,20 +190,6 @@ impl HttpRequest {
|
||||
})
|
||||
}
|
||||
|
||||
/// Read http request with http_rrs support
|
||||
pub fn read_with_rrs(data: &mut impl Read) -> Result<HttpRequest, HttpError> {
|
||||
let addr = match read_line_lf(data) {
|
||||
Ok(i) => i,
|
||||
Err(e) => {
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
.to_socket_addrs()
|
||||
.unwrap()
|
||||
.collect::<Vec<SocketAddr>>()[0];
|
||||
HttpRequest::read(data, &addr)
|
||||
}
|
||||
|
||||
/// Set params to query in url
|
||||
pub fn params_to_page(&mut self) {
|
||||
let mut query = String::new();
|
||||
@ -246,7 +221,7 @@ impl HttpRequest {
|
||||
/// Write http request to stream
|
||||
///
|
||||
/// [`params`](Self::params) is not written to the stream, you need to use [`params_to_json`](Self::params_to_json) or [`params_to_page`](Self::params_to_page)
|
||||
pub fn write(self, data: &mut impl Write) -> Result<(), HttpError> {
|
||||
pub async fn write(self, data: &mut (impl AsyncWriteExt + Unpin)) -> Result<(), HttpError> {
|
||||
let mut head: String = String::new();
|
||||
head.push_str(&self.method);
|
||||
head.push_str(" ");
|
||||
@ -263,13 +238,13 @@ impl HttpRequest {
|
||||
|
||||
head.push_str("\r\n");
|
||||
|
||||
match data.write_all(head.as_bytes()) {
|
||||
match data.write_all(head.as_bytes()).await {
|
||||
Ok(i) => i,
|
||||
Err(_) => return Err(HttpError::WriteHeadError),
|
||||
};
|
||||
|
||||
if !self.data.is_empty() {
|
||||
match data.write_all(&self.data) {
|
||||
match data.write_all(&self.data).await {
|
||||
Ok(i) => i,
|
||||
Err(_) => return Err(HttpError::WriteBodyError),
|
||||
};
|
||||
|
||||
@ -1,10 +1,8 @@
|
||||
use super::{read_line_crlf, Headers, HttpError};
|
||||
|
||||
use serde_json::Value;
|
||||
use std::{
|
||||
fmt::{Debug, Display},
|
||||
io::{Read, Write},
|
||||
};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use std::fmt::{Debug, Display};
|
||||
|
||||
/// Http response
|
||||
#[derive(Debug, Clone)]
|
||||
@ -61,8 +59,8 @@ impl HttpResponse {
|
||||
}
|
||||
|
||||
/// Read http response from stream
|
||||
pub fn read(data: &mut impl Read) -> Result<HttpResponse, HttpError> {
|
||||
let status = match read_line_crlf(data) {
|
||||
pub async fn read(data: &mut (impl AsyncReadExt + Unpin)) -> Result<HttpResponse, HttpError> {
|
||||
let status = match read_line_crlf(data).await {
|
||||
Ok(i) => i,
|
||||
Err(e) => {
|
||||
return Err(e);
|
||||
@ -77,7 +75,7 @@ impl HttpResponse {
|
||||
let mut headers = Headers::new();
|
||||
|
||||
loop {
|
||||
let text = match read_line_crlf(data) {
|
||||
let text = match read_line_crlf(data).await {
|
||||
Ok(i) => i,
|
||||
Err(_) => return Err(HttpError::InvalidHeaders),
|
||||
};
|
||||
@ -106,7 +104,7 @@ impl HttpResponse {
|
||||
let mut buf: Vec<u8> = Vec::new();
|
||||
buf.resize(content_size - reqdata.len(), 0);
|
||||
|
||||
match data.read_exact(&mut buf) {
|
||||
match data.read_exact(&mut buf).await {
|
||||
Ok(i) => i,
|
||||
Err(_) => return Err(HttpError::InvalidContent),
|
||||
};
|
||||
@ -117,7 +115,7 @@ impl HttpResponse {
|
||||
loop {
|
||||
let mut buf: Vec<u8> = vec![0; 1024 * 32];
|
||||
|
||||
let buf_len = match data.read(&mut buf) {
|
||||
let buf_len = match data.read(&mut buf).await {
|
||||
Ok(i) => i,
|
||||
Err(_) => {
|
||||
break;
|
||||
@ -138,7 +136,7 @@ impl HttpResponse {
|
||||
}
|
||||
|
||||
/// Write http response to stream
|
||||
pub fn write(self, data: &mut impl Write) -> Result<(), &str> {
|
||||
pub async fn write(self, data: &mut (impl AsyncWriteExt + Unpin)) -> Result<(), HttpError> {
|
||||
let mut head: String = String::new();
|
||||
head.push_str("HTTP/1.1 ");
|
||||
head.push_str(&self.status_code);
|
||||
@ -153,14 +151,14 @@ impl HttpResponse {
|
||||
|
||||
head.push_str("\r\n");
|
||||
|
||||
match data.write_all(head.as_bytes()) {
|
||||
match data.write_all(head.as_bytes()).await {
|
||||
Ok(i) => i,
|
||||
Err(_) => return Err("write head error"),
|
||||
Err(_) => return Err(HttpError::WriteHeadError),
|
||||
};
|
||||
|
||||
match data.write_all(&self.data) {
|
||||
match data.write_all(&self.data).await {
|
||||
Ok(i) => i,
|
||||
Err(_) => return Err("write body error"),
|
||||
Err(_) => return Err(HttpError::WriteHeadError),
|
||||
};
|
||||
|
||||
Ok(())
|
||||
|
||||
@ -1,28 +1,25 @@
|
||||
use tokio::task::JoinHandle;
|
||||
use tokio::sync::Mutex;
|
||||
use tokio::net::TcpStream;
|
||||
use tokio_io_timeout::TimeoutStream;
|
||||
|
||||
use super::{
|
||||
start_server_new_thread, start_server_sync,
|
||||
start_server_with_threadpool, HttpServer,
|
||||
start_server_new_thread,
|
||||
start_server_sync,
|
||||
start_server_with_threadpool,
|
||||
handler_connection,
|
||||
Handler,
|
||||
HttpServer,
|
||||
};
|
||||
|
||||
use std::pin::Pin;
|
||||
use std::{
|
||||
error::Error,
|
||||
sync::{
|
||||
error::Error, future::Future, sync::{
|
||||
atomic::{AtomicBool, Ordering},
|
||||
Arc,
|
||||
},
|
||||
time::Duration,
|
||||
}, time::Duration
|
||||
};
|
||||
|
||||
/// Http server start builder
|
||||
pub struct HttpServerStarter<T: HttpServer + Send + 'static> {
|
||||
http_server: T,
|
||||
support_http_rrs: bool,
|
||||
timeout: Option<Duration>,
|
||||
host: String,
|
||||
threads: usize,
|
||||
}
|
||||
|
||||
/// Running http server
|
||||
pub struct RunningHttpServer {
|
||||
thread: JoinHandle<()>,
|
||||
@ -41,12 +38,21 @@ impl RunningHttpServer {
|
||||
}
|
||||
}
|
||||
|
||||
/// Http server start builder
|
||||
pub struct HttpServerStarter<T: HttpServer + Send + 'static> {
|
||||
http_server: T,
|
||||
handler: Handler<T>,
|
||||
timeout: Option<Duration>,
|
||||
host: String,
|
||||
threads: usize,
|
||||
}
|
||||
|
||||
impl<T: HttpServer + Send + 'static> HttpServerStarter<T> {
|
||||
/// Create new HttpServerStarter
|
||||
pub fn new(http_server: T, host: &str) -> Self {
|
||||
HttpServerStarter {
|
||||
http_server,
|
||||
support_http_rrs: false,
|
||||
handler: Box::new(move |a, b| Box::pin(handler_connection(a, b))),
|
||||
timeout: None,
|
||||
host: host.to_string(),
|
||||
threads: 0,
|
||||
@ -56,25 +62,25 @@ impl<T: HttpServer + Send + 'static> HttpServerStarter<T> {
|
||||
/// Set http server
|
||||
pub fn http_server(mut self, http_server: T) -> Self {
|
||||
self.http_server = http_server;
|
||||
return self;
|
||||
self
|
||||
}
|
||||
|
||||
/// Set if http_rrs is supported
|
||||
pub fn support_http_rrs(mut self, support_http_rrs: bool) -> Self {
|
||||
self.support_http_rrs = support_http_rrs;
|
||||
return self;
|
||||
pub fn handler(mut self, handler: impl Fn(Arc<Mutex<T>>, TimeoutStream<TcpStream>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static) -> Self {
|
||||
self.handler = Box::new(move |a, b| Box::pin(handler(a, b)));
|
||||
self
|
||||
}
|
||||
|
||||
/// Set timeout for read & write
|
||||
pub fn timeout(mut self, timeout: Option<Duration>) -> Self {
|
||||
self.timeout = timeout;
|
||||
return self;
|
||||
self
|
||||
}
|
||||
|
||||
/// Set host
|
||||
pub fn host(mut self, host: String) -> Self {
|
||||
self.host = host;
|
||||
return self;
|
||||
self
|
||||
}
|
||||
|
||||
/// Set threads in threadpool and return builder
|
||||
@ -83,17 +89,17 @@ impl<T: HttpServer + Send + 'static> HttpServerStarter<T> {
|
||||
/// 1 thread means that all connections are processed in the main thread
|
||||
pub fn threads(mut self, threads: usize) -> Self {
|
||||
self.threads = threads;
|
||||
return self;
|
||||
self
|
||||
}
|
||||
|
||||
/// Get http server
|
||||
pub fn get_http_server(self) -> T {
|
||||
self.http_server
|
||||
pub fn get_http_server(&self) -> &T {
|
||||
&self.http_server
|
||||
}
|
||||
|
||||
/// Get if http_rrs is supported
|
||||
pub fn get_support_http_rrs(&self) -> bool {
|
||||
self.support_http_rrs
|
||||
pub fn get_handler(&self) -> &Handler<T> {
|
||||
&self.handler
|
||||
}
|
||||
|
||||
/// Get timeout for read & write
|
||||
@ -109,7 +115,7 @@ impl<T: HttpServer + Send + 'static> HttpServerStarter<T> {
|
||||
/// Get threads in threadpool
|
||||
///
|
||||
/// 0 threads means that a new thread is created for each connection \
|
||||
/// 1 thread means that all connections are processed in the main thread
|
||||
/// 1 thread means that all connections are processed in the one thread
|
||||
pub fn get_threads(&self) -> usize {
|
||||
self.threads
|
||||
}
|
||||
@ -119,16 +125,16 @@ impl<T: HttpServer + Send + 'static> HttpServerStarter<T> {
|
||||
let running = Arc::new(AtomicBool::new(true));
|
||||
|
||||
if self.threads == 0 {
|
||||
start_server_new_thread(self.http_server, &self.host, self.timeout, self.support_http_rrs, running).await
|
||||
start_server_new_thread(self.http_server, &self.host, self.timeout, self.handler, running).await
|
||||
} else if self.threads == 1 {
|
||||
start_server_sync(self.http_server, &self.host, self.timeout, self.support_http_rrs, running).await
|
||||
start_server_sync(self.http_server, &self.host, self.timeout, self.handler, running).await
|
||||
} else {
|
||||
start_server_with_threadpool(
|
||||
self.http_server,
|
||||
&self.host,
|
||||
self.timeout,
|
||||
self.threads,
|
||||
self.support_http_rrs,
|
||||
self.handler,
|
||||
running,
|
||||
).await
|
||||
}
|
||||
@ -145,7 +151,7 @@ impl<T: HttpServer + Send + 'static> HttpServerStarter<T> {
|
||||
self.http_server,
|
||||
&self.host,
|
||||
self.timeout,
|
||||
self.support_http_rrs,
|
||||
self.handler,
|
||||
running_clone,
|
||||
).await
|
||||
.expect("http server error");
|
||||
@ -156,7 +162,7 @@ impl<T: HttpServer + Send + 'static> HttpServerStarter<T> {
|
||||
self.http_server,
|
||||
&self.host,
|
||||
self.timeout,
|
||||
self.support_http_rrs,
|
||||
self.handler,
|
||||
running_clone,
|
||||
).await
|
||||
.expect("http server error");
|
||||
@ -168,7 +174,7 @@ impl<T: HttpServer + Send + 'static> HttpServerStarter<T> {
|
||||
&self.host,
|
||||
self.timeout,
|
||||
self.threads,
|
||||
self.support_http_rrs,
|
||||
self.handler,
|
||||
running_clone,
|
||||
).await
|
||||
.expect("http server error")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user