diff --git a/README.md b/README.md index d1f97f9..8b6ab3c 100644 --- a/README.md +++ b/README.md @@ -130,4 +130,4 @@ The product is distributed under [Apache License Version 2.0](https://www.apache - Uses Pingora under the hood for efficiency and flexibility. - Designed for edge proxying, internal routing, or hybrid cloud scenarios. -- WebSocket upgrade support is fully automatic. \ No newline at end of file +- Transparent, fully automatic WebSocket upgrade support. \ No newline at end of file diff --git a/src/backup/handler.rs b/src/backup/handler.rs deleted file mode 100644 index 9a4da39..0000000 --- a/src/backup/handler.rs +++ /dev/null @@ -1,54 +0,0 @@ -use crate::peers::Peers; -use hyper::{Body, Request, Response, StatusCode}; -use std::convert::Infallible; -use std::net::IpAddr; -use std::thread::sleep; -use std::time::Duration; - -pub async fn proxy_http(client_ip: IpAddr, req: Request, db: Peers) -> Result, Infallible> { - let p = db.clone(); - let rurl = req.uri().path(); - let yoyo = req.uri().path().to_string(); // Bad thing, only for debug - let peer = crate::peers::return_peer(p, rurl); - match hyper_reverse_proxy::call(client_ip, peer.as_ref(), req).await { - Ok(response) => { - println!("Peer: {}, Client: {}, Path: {}, Status: {}", peer.as_ref(), client_ip, yoyo, response.status()); - Ok(response) - } - Err(_error) => { - println!("Error: no live peers for: {}", yoyo); - Ok(Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::empty()).unwrap()) - } - } -} - -pub async fn healthcheck(peers: Peers, work: Peers) -> Result, Infallible> { - loop { - println!("Main Peers -> {:?}", peers); - println!("Work Peers -> {:?}", work); - sleep(Duration::from_secs(10)); - - peers.clone().iter().for_each(|peer| { - work.insert(peer.key().clone(), peer.value().clone()); - // work[peer.key()] = peer.value(); - }); - } -} - -// #[tokio::main] -// async fn client_check(url: &str) { -// let url = url.parse::().unwrap(); -// let host = url.host().expect("uri has no host"); -// let port = url.port_u16().unwrap_or(80); -// let address = format!("{}:{}", host, port); -// let stream = TcpStream::connect(address).await.unwrap(); -// let io = TokioIo::new(stream); -// let (mut sender, conn) = hyper::client::conn::http1::handshake(io).await.unwrap(); -// -// // Spawn a task to poll the connection, driving the HTTP state -// tokio::task::spawn(async move { -// if let Err(err) = conn.await { -// println!("Connection failed: {:?}", err); -// } -// }); -// } diff --git a/src/backup/peers.rs b/src/backup/peers.rs deleted file mode 100644 index 2437d27..0000000 --- a/src/backup/peers.rs +++ /dev/null @@ -1,28 +0,0 @@ -use dashmap::DashMap; -use rand::Rng; -use std::sync::Arc; - -// type Db = Arc, Arc>>>; -pub type Peers = Arc, Vec>>>; - -pub fn add_peers(peers: Peers, path: &str) { - if let Some(mut peers_list) = peers.get_mut(&Arc::from(path)) { - peers_list.push(Arc::from("http://192.168.1.1:8000")); - peers_list.push(Arc::from("http://192.168.1.10:8000")); - } - println!("Adding peers {} -> {:?}", peers.get(path).unwrap().key(), peers.get(path).unwrap().value()); -} -pub fn return_peer(peers: Peers, path: &str) -> Arc { - if let Some(peer_list) = peers.get(&Arc::from(path)) { - let mut rng = rand::thread_rng(); - let r = rng.gen_range(0..peer_list.len()); - - if let Some(selected_peer) = peer_list.get(r) { - selected_peer.clone() - } else { - Arc::from("https://127.0.0.1:8443") - } - } else { - Arc::from("https://127.0.0.1:8443") - } -} diff --git a/src/backup/start.rs b/src/backup/start.rs deleted file mode 100644 index 47a66f0..0000000 --- a/src/backup/start.rs +++ /dev/null @@ -1,45 +0,0 @@ -use dashmap::DashMap; -use hyper::server::conn::AddrStream; -use hyper::service::{make_service_fn, service_fn}; -use hyper::Server; -use web::handler; -use web::peers; - -#[tokio::main] -async fn main() { - let bind_addr = "0.0.0.0:6193"; - let addr: SocketAddr = bind_addr.parse().expect("Could not parse ip:port."); - - let main_peers: peers::Peers = Arc::new(DashMap::new()); - let work_peers: peers::Peers = Arc::new(DashMap::new()); - - main_peers.insert(Arc::from("/first"), vec![]); - let p = main_peers.clone(); - peers::add_peers(p, "/first"); - - main_peers.insert(Arc::from("/second"), vec![]); - let r = main_peers.clone(); - peers::add_peers(r, "/second"); - - let main_prs = main_peers.clone(); - let work_prs = work_peers.clone(); - tokio::spawn(async move { - let h = handler::healthcheck(main_prs.clone(), work_prs.clone()); - h.await.expect("health check failed"); - // handler::healthcheck(main_prs.clone(), work_prs.clone()).await; - }); - - let make_svc = make_service_fn(|conn: &AddrStream| { - let remote_addr = conn.remote_addr().ip(); - let value = work_peers.clone(); - async move { Ok::<_, Infallible>(service_fn(move |req| handler::proxy_http(remote_addr, req, value.clone()))) } - }); - - let server = Server::bind(&addr).serve(make_svc); - - println!("Running server on {:?}", addr); - - if let Err(e) = server.await { - println!("server error: {}", e); - } -} diff --git a/src/backup/tcpcache.rs b/src/backup/tcpcache.rs deleted file mode 100644 index 10e361c..0000000 --- a/src/backup/tcpcache.rs +++ /dev/null @@ -1,113 +0,0 @@ -/* -use std::collections::HashMap; -use std::sync::Arc; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::{TcpListener, TcpStream}; -use tokio::sync::RwLock; - -type Db = Arc, Arc<[u8]>>>>; -const DBG: bool = true; -#[tokio::main] -pub async fn run() { - println!("\n= = = = = = = = ASYNC TOKIO = = = = = = = = =\n"); - if 1 == 1 { - return; - } - let listener = TcpListener::bind("0.0.0.0:6379").await.unwrap(); - println!("Server is running on 0.0.0.0:6379 !\n"); - let hashmap: Db = Arc::new(RwLock::new(HashMap::new())); - loop { - let (socket, _) = listener.accept().await.unwrap(); - let hashmap = hashmap.clone(); - tokio::spawn(async move { - process(socket, hashmap).await; - }); - } -} - -async fn process(mut socket: TcpStream, db: Db) { - let mut buf = vec![0; 1024]; - loop { - let n = socket.read(&mut buf).await.expect("failed to read data from socket"); - match n > 3 { - true => { - if let Some((action, key, value)) = process_data(&buf[..n - 1], b' ') { - let mut map = db.write().await; - match action { - [115, 101, 116] => { - // SET - map.insert(Arc::from(key), Arc::from(value)); - socket.write_all("Done!\n".as_ref()).await.expect("failed to write"); - } - [103, 101, 116] => { - // GET - let t = map.get(&Arc::from(key)); - match t { - Some(t) => { - socket.write_all(t.as_ref()).await.expect("failed to write"); - } - None => { - socket.write_all("Not Found !\n".as_ref()).await.expect("failed to read"); - } - } - } - [100, 101, 108] => { - // DEL - let y = map.remove(&Arc::from(key)); - let mut _mssg = ""; - match y { - Some(_) => { - _mssg = "Deleted !\n"; - } - None => _mssg = "Not Found !\n", - } - socket.write_all(_mssg.as_ref()).await.expect("failed to write"); - } - [98, 121, 101] => { - //BYE - socket.write_all("Bye!: Closing the connection\n".as_ref()).await.expect("failed"); - socket.shutdown().await.expect("shutdown socket error"); - return; - } - _ => socket.write_all("Unknown command: send (get/set/del)\n".as_ref()).await.expect("failed to read"), - } - } - } - false => { - socket.write_all("Only get/set/del commands are accepted\n".as_ref()).await.expect("failed to respond"); - } - } - } -} - -fn process_data(data: &[u8], delim: u8) -> Option<(&[u8], &[u8], &[u8])> { - let action_bytes = &data[..3]; - - if DBG { - match data.get(4..) { - Some(_d) => { - println!(" DEBUG => {} : {} ", std::str::from_utf8(action_bytes).ok()?, std::str::from_utf8(&data[4..]).ok()?); - } - None => println!(" DEBUG => Goodbye"), - } - } - - match action_bytes { - [103, 101, 116] | [100, 101, 108] => { - let key = &data[4..]; - let val = &[]; - return Some((action_bytes, key, val)); - } - [115, 101, 116] => { - if let Some(pos) = data[4..].iter().position(|&b| b == delim) { - let (key_bytes, value_bytes) = data[4..].split_at(pos); - let value_bytes = &value_bytes[1..]; - return Some((action_bytes, key_bytes, value_bytes)); - } - } - _ => return Some((action_bytes, &[], &[])), - } - - None -} -*/