Github/aralez
1
0
Fork 0
mirror of https://github.com/sadoyan/aralez.git synced 2026-04-29 22:38:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

README.md

Browse Source
This commit is contained in:
Ara Sadoyan
2025-04-15 19:55:54 +02:00
parent 0b9b0f5704
commit a5c588c108
5 changed files with 1 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
src/backup/handler.rs
View File

@@ -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<Body>, db: Peers) -> Result<Response<Body>, 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<Response<Body>, 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::<hyper::Uri>().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);
// }
// });
// }

28
src/backup/peers.rs
View File

@@ -1,28 +0,0 @@
use dashmap::DashMap;
use rand::Rng;
use std::sync::Arc;
// type Db = Arc<RwLock<HashMap<Arc<str>, Arc<i32>>>>;
pub type Peers = Arc<DashMap<Arc<str>, Vec<Arc<str>>>>;
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<str> {
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")
}
}

45
src/backup/start.rs
View File

@@ -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);
}
}

113
src/backup/tcpcache.rs
View File

@@ -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<RwLock<HashMap<Arc<[u8]>, 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
}
*/