diff --git a/src/server.rs b/src/server.rs deleted file mode 100644 index f56496f..0000000 --- a/src/server.rs +++ /dev/null @@ -1,133 +0,0 @@ -use tokio::net::TcpListener; -use std::sync::Arc; -use crate::config::ServerConfig; -use log::{info, error}; -use colored::Colorize; - -pub fn print_startup_info(config: &ServerConfig) { - println!(); - println!("{}", "Futriix-Distributed JSON-server".bold().green()); - println!("Process ID: {}", std::process::id()); - println!("Port: {}", config.port); - println!("Mode: {}", if config.cluster_mode { - "Cluster-mode".bold().blue() - } else { - "Standalone-mode".bold().yellow() - }); - println!("Log file: {}", config.log_path.display()); - println!(); -} - -async fn handle_standalone_connection( - mut stream: tokio::net::TcpStream, - client: redis::Client, - _config: Arc, -) -> anyhow::Result<()> { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - let mut conn = client.get_connection()?; - let mut buffer = [0; 1024]; - - loop { - let n = stream.read(&mut buffer).await?; - if n == 0 { - break; - } - - let input = String::from_utf8_lossy(&buffer[..n]); - info!("Received command: {}", input.trim()); - - if let Some(cmd) = crate::commands::Command::parse(&input) { - match cmd.execute_standalone(&mut conn) { - Ok(result) => { - let response = serde_json::to_string(&result)?; - stream.write_all(response.as_bytes()).await?; - } - Err(e) => { - let error_msg = format!("Error: {}", e); - stream.write_all(error_msg.as_bytes()).await?; - } - } - } else { - let error_msg = "Error: Invalid command"; - stream.write_all(error_msg.as_bytes()).await?; - } - } - - Ok(()) -} - -async fn handle_cluster_connection( - mut stream: tokio::net::TcpStream, - client: redis::cluster::ClusterClient, - _config: Arc, -) -> anyhow::Result<()> { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - let mut conn = client.get_connection()?; - let mut buffer = [0; 1024]; - - loop { - let n = stream.read(&mut buffer).await?; - if n == 0 { - break; - } - - let input = String::from_utf8_lossy(&buffer[..n]); - info!("Received command (cluster): {}", input.trim()); - - if let Some(cmd) = crate::commands::Command::parse(&input) { - match cmd.execute_cluster(&mut conn) { - Ok(result) => { - let response = serde_json::to_string(&result)?; - stream.write_all(response.as_bytes()).await?; - } - Err(e) => { - let error_msg = format!("Error: {}", e); - stream.write_all(error_msg.as_bytes()).await?; - } - } - } else { - let error_msg = "Error: Invalid command"; - stream.write_all(error_msg.as_bytes()).await?; - } - } - - Ok(()) -} - -pub async fn start_server(config: Arc) -> anyhow::Result<()> { - let listener = TcpListener::bind(format!("0.0.0.0:{}", config.port)).await?; - - info!("Server started on port {}", config.port); - - if config.cluster_mode { - info!("Running in Cluster mode"); - let client = redis::cluster::ClusterClient::new(config.redis_nodes.clone())?; - - while let Ok((stream, _)) = listener.accept().await { - let client = client.clone(); - let config_clone = config.clone(); - - tokio::spawn(async move { - if let Err(e) = handle_cluster_connection(stream, client, config_clone).await { - error!("Connection error: {}", e); - } - }); - } - } else { - info!("Running in Standalone mode"); - let client = redis::Client::open(config.redis_nodes[0].clone())?; - - while let Ok((stream, _)) = listener.accept().await { - let client = client.clone(); - let config_clone = config.clone(); - - tokio::spawn(async move { - if let Err(e) = handle_standalone_connection(stream, client, config_clone).await { - error!("Connection error: {}", e); - } - }); - } - } - - Ok(()) -}