Reload Endpoint Reload Keybind Reload Logic
Remove TU Endpoint
This commit is contained in:
parent
fe0b3f0315
commit
7bf90a3699
10 changed files with 385 additions and 201 deletions
|
|
@ -18,6 +18,9 @@ pub async fn add_community(community: Arc<Community>) {
|
|||
pub async fn remove_community(name: &str) {
|
||||
COMMUNITY_REGISTRY.lock().await.remove(name);
|
||||
}
|
||||
pub async fn clear() {
|
||||
COMMUNITY_REGISTRY.lock().await.clear();
|
||||
}
|
||||
pub async fn get_community(name: &str) -> Option<Arc<Community>> {
|
||||
if let Some(c) = COMMUNITY_REGISTRY.lock().await.get(name) {
|
||||
Some(c.clone())
|
||||
|
|
|
|||
|
|
@ -1,11 +1,16 @@
|
|||
use crate::{SHUTDOWN, gui::tui::UNIQUE, util::config_util::CONFIG};
|
||||
use crate::{ACTIVE_TASKS, RELOAD, SHUTDOWN, gui::tui::UNIQUE, util::config_util::CONFIG};
|
||||
use crossterm::event::{KeyEvent, KeyModifiers};
|
||||
use ratatui::crossterm::event::{Event, KeyCode, read};
|
||||
use tokio::{self};
|
||||
use tokio_util::sync::WaitForCancellationFutureOwned;
|
||||
|
||||
pub fn setup_input_handler() {
|
||||
tokio::spawn(async move {
|
||||
{
|
||||
ACTIVE_TASKS
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push("Input Handler".to_string());
|
||||
}
|
||||
while let Ok(event) = read() {
|
||||
if *SHUTDOWN.read().await {
|
||||
break;
|
||||
|
|
@ -13,6 +18,15 @@ pub fn setup_input_handler() {
|
|||
if let Event::Key(key) = event {
|
||||
handle_input(key).await;
|
||||
}
|
||||
if *SHUTDOWN.read().await {
|
||||
break;
|
||||
}
|
||||
}
|
||||
{
|
||||
ACTIVE_TASKS
|
||||
.lock()
|
||||
.unwrap()
|
||||
.retain(|t| !t.eq(&"Input Handler".to_string()));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
|
@ -22,6 +36,14 @@ pub async fn handle_input(key: KeyEvent) {
|
|||
(KeyCode::Char('c'), KeyModifiers::CONTROL) => {
|
||||
*SHUTDOWN.write().await = true;
|
||||
}
|
||||
(KeyCode::Char('r'), KeyModifiers::CONTROL) => {
|
||||
{
|
||||
*RELOAD.write().await = true;
|
||||
}
|
||||
{
|
||||
*SHUTDOWN.write().await = true;
|
||||
}
|
||||
}
|
||||
(KeyCode::Backspace, KeyModifiers::NONE) => {
|
||||
let password: String = {
|
||||
let cfg = CONFIG.read().await;
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
use crate::ACTIVE_TASKS;
|
||||
use crate::APP_STATE;
|
||||
use crate::SHUTDOWN;
|
||||
use crate::gui::tui::UNIQUE;
|
||||
|
|
@ -36,6 +37,9 @@ pub fn log_message(msg: impl Into<String>) {
|
|||
pub fn setup() {
|
||||
// Start a background thread to sample metrics
|
||||
tokio::spawn(async move {
|
||||
{
|
||||
ACTIVE_TASKS.lock().unwrap().push("metrics".to_string());
|
||||
}
|
||||
let mut sys = System::new_with_specifics(RefreshKind::new());
|
||||
let mut last_total_received = 0u64;
|
||||
let mut last_total_transmitted = 0u64;
|
||||
|
|
@ -86,5 +90,11 @@ pub fn setup() {
|
|||
*UNIQUE.write().await = true;
|
||||
thread::sleep(Duration::from_millis(1000));
|
||||
}
|
||||
{
|
||||
ACTIVE_TASKS
|
||||
.lock()
|
||||
.unwrap()
|
||||
.retain(|t| !t.eq(&"metrics".to_string()));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ use std::{
|
|||
};
|
||||
|
||||
use crate::{
|
||||
APP_STATE, SHUTDOWN,
|
||||
ACTIVE_TASKS, APP_STATE, SHUTDOWN,
|
||||
gui::{settings_panel, widgets::betterblock::draw_block_joins},
|
||||
util::config_util::CONFIG,
|
||||
};
|
||||
|
|
@ -48,6 +48,10 @@ pub static TERMINAL: Lazy<Arc<Mutex<Terminal<CrosstermBackend<Stdout>>>>> = Lazy
|
|||
pub fn start_tui() {
|
||||
tokio::spawn(async move {
|
||||
init_terminal();
|
||||
|
||||
{
|
||||
ACTIVE_TASKS.lock().unwrap().push("UI".to_string());
|
||||
}
|
||||
loop {
|
||||
if *SHUTDOWN.read().await {
|
||||
break;
|
||||
|
|
@ -58,6 +62,12 @@ pub fn start_tui() {
|
|||
thread::sleep(Duration::from_millis(50));
|
||||
}
|
||||
}
|
||||
{
|
||||
ACTIVE_TASKS
|
||||
.lock()
|
||||
.unwrap()
|
||||
.retain(|t| !t.eq(&"UI".to_string()));
|
||||
}
|
||||
});
|
||||
}
|
||||
pub async fn render_tui() {
|
||||
|
|
|
|||
242
src/main.rs
242
src/main.rs
|
|
@ -1,4 +1,5 @@
|
|||
use json::{self, JsonValue::String};
|
||||
use json::JsonValue;
|
||||
use json::{self};
|
||||
use once_cell::sync::Lazy;
|
||||
use pnet::datalink::NetworkInterface;
|
||||
use std::sync::Arc;
|
||||
|
|
@ -40,140 +41,159 @@ pub static APP_STATE: LazyLock<Arc<Mutex<AppState>>> =
|
|||
LazyLock::new(|| Arc::new(Mutex::new(AppState::new())));
|
||||
|
||||
pub static SHUTDOWN: Lazy<RwLock<bool>> = Lazy::new(|| RwLock::new(false));
|
||||
pub static RELOAD: Lazy<RwLock<bool>> = Lazy::new(|| RwLock::new(true));
|
||||
pub static ACTIVE_TASKS: Lazy<Mutex<Vec<String>>> = Lazy::new(|| Mutex::new(Vec::new()));
|
||||
pub static RECONNECT: Lazy<RwLock<bool>> = Lazy::new(|| RwLock::new(false));
|
||||
|
||||
#[tokio::main(flavor = "multi_thread", worker_threads = 8)]
|
||||
#[allow(unused_must_use, dead_code)]
|
||||
async fn main() {
|
||||
// EULA
|
||||
//if !eula_checker::check_eula() {
|
||||
// println!("Please accept the end user license agreement before launching!");
|
||||
// return;
|
||||
//}
|
||||
while *RELOAD.read().await {
|
||||
*RELOAD.write().await = false;
|
||||
*SHUTDOWN.write().await = false;
|
||||
|
||||
// LANGUAGE PACK
|
||||
if let Err(e) = language_creator::create_languages() {
|
||||
println!("Language pack creation failed: {}", e);
|
||||
return;
|
||||
}
|
||||
// EULA
|
||||
//if !eula_checker::check_eula() {
|
||||
// println!("Please accept the end user license agreement before launching!");
|
||||
// return;
|
||||
//}
|
||||
|
||||
// UI
|
||||
log_panel::setup();
|
||||
tui::start_tui();
|
||||
input_handler::setup_input_handler();
|
||||
// LANGUAGE PACK
|
||||
if let Err(e) = language_creator::create_languages() {
|
||||
println!("Language pack creation failed: {}", e);
|
||||
return;
|
||||
}
|
||||
|
||||
// BASIC CONFIGURATION
|
||||
&CONFIG.write().await.load();
|
||||
if !CONFIG.read().await.config.has_key("iota_id") {
|
||||
CONFIG
|
||||
.write()
|
||||
.await
|
||||
.change("iota_id", &Uuid::new_v4().to_string());
|
||||
CONFIG.write().await.update();
|
||||
}
|
||||
// UI
|
||||
log_panel::setup();
|
||||
tui::start_tui();
|
||||
input_handler::setup_input_handler();
|
||||
|
||||
// USER MANAGEMENT
|
||||
if let Err(_) = user_manager::load_users().await {
|
||||
log_message_trans("user_load_failed");
|
||||
}
|
||||
// BASIC CONFIGURATION
|
||||
&CONFIG.write().await.load();
|
||||
if !CONFIG.read().await.config.has_key("iota_id") {
|
||||
CONFIG
|
||||
.write()
|
||||
.await
|
||||
.change("iota_id", &Uuid::new_v4().to_string());
|
||||
CONFIG.write().await.update();
|
||||
}
|
||||
|
||||
let mut sb = "".to_string();
|
||||
// USER MANAGEMENT
|
||||
if let Err(_) = user_manager::load_users().await {
|
||||
log_message_trans("user_load_failed");
|
||||
}
|
||||
|
||||
for up in user_manager::get_users() {
|
||||
sb = sb + "," + &up.user_id.to_string().as_str();
|
||||
}
|
||||
let mut sb = "".to_string();
|
||||
|
||||
if !sb.is_empty() {
|
||||
{}
|
||||
sb.remove(0);
|
||||
sb = sb + ",";
|
||||
}
|
||||
log_message(format!(
|
||||
"IOTA ID: {}-####-####-####-############",
|
||||
CONFIG
|
||||
.read()
|
||||
.await
|
||||
.get_iota_id()
|
||||
.to_string()
|
||||
.split("-")
|
||||
.next()
|
||||
.unwrap()
|
||||
));
|
||||
log_message(format!("User IDS: {}", sb));
|
||||
for up in user_manager::get_users() {
|
||||
sb = sb + "," + &up.user_id.to_string().as_str();
|
||||
}
|
||||
|
||||
// COMMUNITY MANAGEMENT
|
||||
registry::load_interactables().await;
|
||||
community_manager::load_communities().await;
|
||||
community_manager::save_communities().await;
|
||||
let mut sb1 = "".to_string();
|
||||
for cp in community_manager::get_communities().await {
|
||||
sb1 = sb1 + "," + &cp.get_name().to_string().as_str();
|
||||
}
|
||||
if !sb.is_empty() {
|
||||
{}
|
||||
sb.remove(0);
|
||||
sb = sb + ",";
|
||||
}
|
||||
log_message(format!(
|
||||
"IOTA ID: {}-####-####-####-############",
|
||||
CONFIG
|
||||
.read()
|
||||
.await
|
||||
.get_iota_id()
|
||||
.to_string()
|
||||
.split("-")
|
||||
.next()
|
||||
.unwrap()
|
||||
));
|
||||
log_message(format!("User IDS: {}", sb));
|
||||
|
||||
if !sb1.is_empty() {
|
||||
sb1.remove(0);
|
||||
sb1 = sb1 + ",";
|
||||
}
|
||||
log_message(format!("Community IDS: {}", sb1));
|
||||
let port = CONFIG.read().await.get_port();
|
||||
let mut ip = "0.0.0.0".to_string();
|
||||
for iface in pnet::datalink::interfaces() {
|
||||
let iface: NetworkInterface = iface;
|
||||
if iface.ips.len() > 0 {
|
||||
let ipsv = format!("{}", iface.ips[0]);
|
||||
let ips: &str = ipsv.split('/').next().unwrap();
|
||||
if format!("{}", ips).starts_with("10.") || format!("{}", ips).starts_with("192.") {
|
||||
ip = ips.to_string();
|
||||
// COMMUNITY MANAGEMENT
|
||||
registry::load_interactables().await;
|
||||
community_manager::load_communities().await;
|
||||
community_manager::save_communities().await;
|
||||
let mut sb1 = "".to_string();
|
||||
for cp in community_manager::get_communities().await {
|
||||
sb1 = sb1 + "," + &cp.get_name().to_string().as_str();
|
||||
}
|
||||
|
||||
if !sb1.is_empty() {
|
||||
sb1.remove(0);
|
||||
sb1 = sb1 + ",";
|
||||
}
|
||||
log_message(format!("Community IDS: {}", sb1));
|
||||
let port = CONFIG.read().await.get_port();
|
||||
let mut ip = "0.0.0.0".to_string();
|
||||
for iface in pnet::datalink::interfaces() {
|
||||
let iface: NetworkInterface = iface;
|
||||
if iface.ips.len() > 0 {
|
||||
let ipsv = format!("{}", iface.ips[0]);
|
||||
let ips: &str = ipsv.split('/').next().unwrap();
|
||||
if format!("{}", ips).starts_with("10.") || format!("{}", ips).starts_with("192.") {
|
||||
ip = ips.to_string();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if start(port).await {
|
||||
log_message(format("community_active", &[&ip, &port.to_string()]));
|
||||
} else {
|
||||
if port < 1024 {
|
||||
log_message(format("community_start_error_admin", &[&port.to_string()]));
|
||||
if start(port).await {
|
||||
log_message(format("community_active", &[&ip, &port.to_string()]));
|
||||
} else {
|
||||
log_message(format("community_start_error", &[&port.to_string()]));
|
||||
if port < 1024 {
|
||||
log_message(format("community_start_error_admin", &[&port.to_string()]));
|
||||
} else {
|
||||
log_message(format("community_start_error", &[&port.to_string()]));
|
||||
}
|
||||
}
|
||||
}
|
||||
if !has_dir("web") {
|
||||
/*download_and_extract_zip(
|
||||
"weblink",
|
||||
"web",
|
||||
)
|
||||
.await;*/
|
||||
}
|
||||
loop {
|
||||
if *SHUTDOWN.read().await {
|
||||
break;
|
||||
}
|
||||
let omikron: Arc<OmikronConnection> = Arc::new(OmikronConnection::new());
|
||||
omikron.connect().await;
|
||||
omikron
|
||||
.send_message(
|
||||
CommunicationValue::new(CommunicationType::identification)
|
||||
.add_data(DataTypes::user_ids, String(sb.to_string()))
|
||||
.add_data(
|
||||
DataTypes::iota_id,
|
||||
String(CONFIG.read().await.get_iota_id().to_string()),
|
||||
)
|
||||
.to_json()
|
||||
.to_string()
|
||||
.as_mut()
|
||||
.to_string(),
|
||||
if !has_dir("web") {
|
||||
/*download_and_extract_zip(
|
||||
"weblink",
|
||||
"web",
|
||||
)
|
||||
.await;
|
||||
let mut omikron_connection = OMIKRON_CONNECTION.write().await;
|
||||
*omikron_connection = Some(omikron.clone());
|
||||
log_message_trans("setup_completed");
|
||||
.await;*/
|
||||
}
|
||||
loop {
|
||||
if *SHUTDOWN.read().await {
|
||||
break;
|
||||
}
|
||||
if !omikron.is_connected().await {
|
||||
break;
|
||||
let omikron: Arc<OmikronConnection> = Arc::new(OmikronConnection::new());
|
||||
omikron.connect().await;
|
||||
omikron
|
||||
.send_message(
|
||||
CommunicationValue::new(CommunicationType::identification)
|
||||
.add_data(DataTypes::user_ids, JsonValue::String(sb.to_string()))
|
||||
.add_data(
|
||||
DataTypes::iota_id,
|
||||
JsonValue::String(CONFIG.read().await.get_iota_id().to_string()),
|
||||
)
|
||||
.to_json()
|
||||
.to_string()
|
||||
.as_mut()
|
||||
.to_string(),
|
||||
)
|
||||
.await;
|
||||
let mut omikron_connection = OMIKRON_CONNECTION.write().await;
|
||||
*omikron_connection = Some(omikron.clone());
|
||||
log_message_trans("setup_completed");
|
||||
loop {
|
||||
if *SHUTDOWN.read().await {
|
||||
break;
|
||||
}
|
||||
if !omikron.is_connected().await {
|
||||
break;
|
||||
}
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
if *RELOAD.read().await {
|
||||
loop {
|
||||
if ACTIVE_TASKS.lock().unwrap().is_empty() {
|
||||
break;
|
||||
}
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
&CONFIG.write().await.clear();
|
||||
user_manager::clear();
|
||||
community_manager::clear();
|
||||
*APP_STATE.lock().unwrap() = AppState::new();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,13 +1,12 @@
|
|||
use crate::auth::local_auth;
|
||||
use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes};
|
||||
use crate::gui::log_panel::{log_cv, log_message, log_message_trans};
|
||||
use crate::langu::language_manager::format;
|
||||
use crate::users::contact::Contact;
|
||||
use crate::users::user_community_util::UserCommunityUtil;
|
||||
use crate::util::chat_files;
|
||||
use crate::util::chats_util::{get_user, get_users, mod_user};
|
||||
use crate::util::file_util::{get_children, load_file, save_file};
|
||||
use crate::{RECONNECT, SHUTDOWN};
|
||||
use crate::{ACTIVE_TASKS, RECONNECT, SHUTDOWN};
|
||||
use futures::Stream;
|
||||
use futures::stream::{SplitSink, SplitStream};
|
||||
use futures_util::sink::Sink;
|
||||
|
|
@ -103,6 +102,10 @@ impl OmikronConnection {
|
|||
> = Box::new(read_half);
|
||||
self.clone().spawn_listener(boxed_reader).await;
|
||||
let cloned_self = self.clone();
|
||||
|
||||
{
|
||||
ACTIVE_TASKS.lock().unwrap().push("PingPong".to_string());
|
||||
}
|
||||
let handle = tokio::spawn(async move {
|
||||
loop {
|
||||
if *SHUTDOWN.read().await {
|
||||
|
|
@ -115,13 +118,18 @@ impl OmikronConnection {
|
|||
sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
});
|
||||
{
|
||||
ACTIVE_TASKS
|
||||
.lock()
|
||||
.unwrap()
|
||||
.retain(|t| !t.eq(&"PingPong".to_string()));
|
||||
}
|
||||
|
||||
*self.is_connected.lock().await = true;
|
||||
*self.pingpong.lock().await = Some(handle);
|
||||
break;
|
||||
}
|
||||
Err(e) => {
|
||||
log_message(format("connection_failed", &[&e.to_string().as_str()]));
|
||||
Err(_) => {
|
||||
*self.is_connected.lock().await = false;
|
||||
sleep(Duration::from_secs(2)).await;
|
||||
}
|
||||
|
|
@ -150,6 +158,10 @@ impl OmikronConnection {
|
|||
let sel_out = self.clone();
|
||||
let variant = self.variant.clone();
|
||||
let sel_arc_out = self.clone();
|
||||
|
||||
{
|
||||
ACTIVE_TASKS.lock().unwrap().push("Listener".to_string());
|
||||
}
|
||||
tokio::spawn(async move {
|
||||
while let Some(msg) = read_half.next().await {
|
||||
if *SHUTDOWN.read().await {
|
||||
|
|
@ -568,6 +580,12 @@ impl OmikronConnection {
|
|||
});
|
||||
}
|
||||
});
|
||||
{
|
||||
ACTIVE_TASKS
|
||||
.lock()
|
||||
.unwrap()
|
||||
.retain(|t| !t.eq(&"Listener".to_string()));
|
||||
}
|
||||
}
|
||||
pub async fn send_message_static(
|
||||
writer: &Arc<
|
||||
|
|
|
|||
|
|
@ -1,10 +1,11 @@
|
|||
use std::sync::Arc;
|
||||
|
||||
use crate::SHUTDOWN;
|
||||
use crate::auth::auth_connector::unregister_user;
|
||||
use crate::communities::community::Community;
|
||||
use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes};
|
||||
use crate::gui::log_panel::log_message;
|
||||
use crate::util::file_util::delete_file;
|
||||
use crate::{RELOAD, SHUTDOWN};
|
||||
use axum::http::HeaderValue;
|
||||
use http_body_util::Full;
|
||||
use hyper::body::Bytes;
|
||||
|
|
@ -48,6 +49,15 @@ pub async fn handle(
|
|||
"{\"type\":\"success\"}".to_string(),
|
||||
)
|
||||
}
|
||||
"reload" => {
|
||||
*SHUTDOWN.write().await = true;
|
||||
*RELOAD.write().await = true;
|
||||
(
|
||||
StatusCode::OK,
|
||||
"application/json",
|
||||
"{\"type\":\"success\"}".to_string(),
|
||||
)
|
||||
}
|
||||
"app_state" => (StatusCode::OK, "application/json", {
|
||||
let with = headers
|
||||
.get("size")
|
||||
|
|
@ -82,6 +92,18 @@ pub async fn handle(
|
|||
}
|
||||
}
|
||||
}
|
||||
"remove_tu" => {
|
||||
if body.is_none() {
|
||||
"{\"type\":\"error\"}".to_string()
|
||||
} else {
|
||||
let username =
|
||||
body.unwrap()["username"].as_str().unwrap().to_string();
|
||||
|
||||
delete_file("", &format!("{}.tu", username));
|
||||
|
||||
"{\"type\":\"success\"}".to_string()
|
||||
}
|
||||
}
|
||||
"remove" => {
|
||||
if body.is_none() {
|
||||
"{\"type\":\"error\"}".to_string()
|
||||
|
|
@ -94,6 +116,7 @@ pub async fn handle(
|
|||
)
|
||||
.await;
|
||||
user_manager::remove_user(uuid);
|
||||
user_manager::save_users();
|
||||
"{}".to_string()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,8 +1,8 @@
|
|||
use crate::SHUTDOWN;
|
||||
use crate::gui::log_panel::log_message;
|
||||
use crate::server::api;
|
||||
use crate::server::socket::handle;
|
||||
use crate::util::file_util::{load_file_buf, load_file_vec};
|
||||
use crate::{ACTIVE_TASKS, SHUTDOWN};
|
||||
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::STANDARD;
|
||||
|
|
@ -29,6 +29,7 @@ use std::result::Result::Ok;
|
|||
use std::sync::Arc;
|
||||
use std::{future::Future, pin::Pin, time::Duration};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::broadcast; // Import broadcast for the kill switch
|
||||
use tokio_rustls::TlsAcceptor;
|
||||
use tokio_tungstenite::WebSocketStream;
|
||||
use tower::Service;
|
||||
|
|
@ -201,7 +202,7 @@ impl Service<HttpRequest<Incoming>> for HttpService {
|
|||
}
|
||||
}
|
||||
|
||||
fn is_local_network(addr: IpAddr) -> bool {
|
||||
fn is_local_network(_addr: IpAddr) -> bool {
|
||||
// PLACEHOLDER
|
||||
return true;
|
||||
}
|
||||
|
|
@ -229,45 +230,81 @@ async fn run_http_server(port: u16) -> bool {
|
|||
ip, port
|
||||
));
|
||||
|
||||
// Create a broadcast channel for graceful shutdown signal
|
||||
let (shutdown_tx, _) = broadcast::channel::<()>(1);
|
||||
|
||||
ACTIVE_TASKS.lock().unwrap().push("WebServer".to_string());
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
if *SHUTDOWN.read().await {
|
||||
break;
|
||||
}
|
||||
match listener.accept().await {
|
||||
std::result::Result::Ok((stream, addr)) => {
|
||||
let service = HttpService { peer_addr: addr };
|
||||
let io = TokioIo::new(stream);
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(err) = http1::Builder::new()
|
||||
.preserve_header_case(true)
|
||||
.title_case_headers(true)
|
||||
.serve_connection(io, TowerToHyperService::new(service))
|
||||
.with_upgrades()
|
||||
.await
|
||||
{
|
||||
if let Some(io_err) =
|
||||
err.source().and_then(|e| e.downcast_ref::<io::Error>())
|
||||
{
|
||||
if io_err.kind() != io::ErrorKind::ConnectionReset
|
||||
&& io_err.kind() != io::ErrorKind::BrokenPipe
|
||||
{
|
||||
log_message(format!("Error serving connection: {:?}", err));
|
||||
}
|
||||
} else {
|
||||
log_message(format!("Error serving connection: {:?}", err));
|
||||
}
|
||||
}
|
||||
});
|
||||
tokio::select! {
|
||||
// Monitor for shutdown signal
|
||||
_ = async {
|
||||
loop {
|
||||
if *SHUTDOWN.read().await { return; }
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
} => {
|
||||
log_message("Standard Server received shutdown signal.");
|
||||
// Send kill signal to all active connection tasks
|
||||
let _ = shutdown_tx.send(());
|
||||
break;
|
||||
}
|
||||
Err(e) => {
|
||||
log_message(format!("Error accepting connection: {:?}", e));
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
|
||||
// Accept new connections
|
||||
accepted = listener.accept() => {
|
||||
match accepted {
|
||||
std::result::Result::Ok((stream, addr)) => {
|
||||
let service = HttpService { peer_addr: addr };
|
||||
let io = TokioIo::new(stream);
|
||||
|
||||
// Subscribe to the shutdown signal for this specific connection
|
||||
let mut rx = shutdown_tx.subscribe();
|
||||
|
||||
tokio::spawn(async move {
|
||||
// Prepare the connection future
|
||||
let conn = http1::Builder::new()
|
||||
.preserve_header_case(true)
|
||||
.title_case_headers(true)
|
||||
.serve_connection(io, TowerToHyperService::new(service))
|
||||
.with_upgrades();
|
||||
|
||||
// Wait for either the connection to finish naturally OR the shutdown signal
|
||||
tokio::select! {
|
||||
res = conn => {
|
||||
if let Err(err) = res {
|
||||
if let Some(io_err) = err.source().and_then(|e| e.downcast_ref::<io::Error>()) {
|
||||
if io_err.kind() != io::ErrorKind::ConnectionReset
|
||||
&& io_err.kind() != io::ErrorKind::BrokenPipe
|
||||
{
|
||||
log_message(format!("Error serving connection: {:?}", err));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
_ = rx.recv() => {
|
||||
// Shutdown signal received.
|
||||
// Dropping the 'conn' future here closes the socket immediately.
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
log_message(format!("Error accepting connection: {:?}", e));
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ACTIVE_TASKS
|
||||
.lock()
|
||||
.unwrap()
|
||||
.retain(|t| !t.eq(&"WebServer".to_string()));
|
||||
log_message("Standard Server shutdown complete.");
|
||||
});
|
||||
|
||||
true
|
||||
}
|
||||
|
||||
|
|
@ -297,57 +334,91 @@ async fn run_tls_server(port: u16, tls_config: Arc<ServerConfig>) -> bool {
|
|||
ip, port
|
||||
));
|
||||
|
||||
// Create a broadcast channel for graceful shutdown signal
|
||||
let (shutdown_tx, _) = broadcast::channel::<()>(1);
|
||||
|
||||
ACTIVE_TASKS.lock().unwrap().push("WebServer".to_string());
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
if *SHUTDOWN.read().await {
|
||||
break;
|
||||
}
|
||||
match listener.accept().await {
|
||||
std::result::Result::Ok((stream, addr)) => {
|
||||
let service = HttpService { peer_addr: addr };
|
||||
let acceptor = acceptor.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
// Perform TLS handshake
|
||||
let tls_stream = match acceptor.accept(stream).await {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
// Ignore non-TLS clients connecting to the TLS port
|
||||
if e.kind() != io::ErrorKind::Interrupted {
|
||||
log_message(format!("TLS Handshake error: {:?}", e));
|
||||
}
|
||||
return;
|
||||
}
|
||||
};
|
||||
let io = TokioIo::new(tls_stream);
|
||||
|
||||
if let Err(err) = http1::Builder::new()
|
||||
.preserve_header_case(true)
|
||||
.title_case_headers(true)
|
||||
.serve_connection(io, TowerToHyperService::new(service))
|
||||
.with_upgrades()
|
||||
.await
|
||||
{
|
||||
if let Some(io_err) =
|
||||
err.source().and_then(|e| e.downcast_ref::<io::Error>())
|
||||
{
|
||||
if io_err.kind() != io::ErrorKind::ConnectionReset
|
||||
&& io_err.kind() != io::ErrorKind::BrokenPipe
|
||||
{
|
||||
log_message(format!("Error serving connection: {:?}", err));
|
||||
}
|
||||
} else {
|
||||
log_message(format!("Error serving connection: {:?}", err));
|
||||
}
|
||||
}
|
||||
});
|
||||
tokio::select! {
|
||||
// Monitor for shutdown signal
|
||||
_ = async {
|
||||
loop {
|
||||
if *SHUTDOWN.read().await { return; }
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
} => {
|
||||
log_message("Encrypted Server received shutdown signal.");
|
||||
// Send kill signal to all active connection tasks
|
||||
let _ = shutdown_tx.send(());
|
||||
break;
|
||||
}
|
||||
Err(e) => {
|
||||
log_message(format!("Error accepting connection: {:?}", e));
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
|
||||
// Accept new connections
|
||||
accepted = listener.accept() => {
|
||||
match accepted {
|
||||
std::result::Result::Ok((stream, addr)) => {
|
||||
let service = HttpService { peer_addr: addr };
|
||||
let acceptor = acceptor.clone();
|
||||
|
||||
// Subscribe to the shutdown signal for this specific connection
|
||||
let mut rx = shutdown_tx.subscribe();
|
||||
|
||||
tokio::spawn(async move {
|
||||
// Perform TLS handshake
|
||||
let tls_stream = match acceptor.accept(stream).await {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
if e.kind() != io::ErrorKind::Interrupted {
|
||||
log_message(format!("TLS Handshake error: {:?}", e));
|
||||
}
|
||||
return;
|
||||
}
|
||||
};
|
||||
let io = TokioIo::new(tls_stream);
|
||||
|
||||
// Prepare connection future
|
||||
let conn = http1::Builder::new()
|
||||
.preserve_header_case(true)
|
||||
.title_case_headers(true)
|
||||
.serve_connection(io, TowerToHyperService::new(service))
|
||||
.with_upgrades();
|
||||
|
||||
// Wait for either the connection to finish naturally OR the shutdown signal
|
||||
tokio::select! {
|
||||
res = conn => {
|
||||
if let Err(err) = res {
|
||||
if let Some(io_err) = err.source().and_then(|e| e.downcast_ref::<io::Error>()) {
|
||||
if io_err.kind() != io::ErrorKind::ConnectionReset
|
||||
&& io_err.kind() != io::ErrorKind::BrokenPipe
|
||||
{
|
||||
log_message(format!("Error serving connection: {:?}", err));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
_ = rx.recv() => {
|
||||
// Shutdown signal received.
|
||||
// Dropping the 'conn' future here closes the socket immediately.
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
log_message(format!("Error accepting connection: {:?}", e));
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ACTIVE_TASKS
|
||||
.lock()
|
||||
.unwrap()
|
||||
.retain(|t| !t.eq(&"WebServer".to_string()));
|
||||
log_message("Encrypted Server shutdown complete.");
|
||||
});
|
||||
true
|
||||
}
|
||||
|
|
|
|||
|
|
@ -82,7 +82,7 @@ pub async fn create_user(username: &str) -> (Option<UserProfile>, Option<String>
|
|||
);
|
||||
|
||||
USERS.lock().unwrap().push(up.clone());
|
||||
save_users().ok();
|
||||
save_users();
|
||||
(Some(up), Some(STANDARD.encode(&private_key.as_bytes())))
|
||||
}
|
||||
|
||||
|
|
@ -105,14 +105,19 @@ pub fn remove_user(user_id: Uuid) {
|
|||
*UNIQUE.lock().unwrap() = true;
|
||||
}
|
||||
|
||||
pub fn save_users() -> io::Result<()> {
|
||||
pub fn save_users() {
|
||||
*UNIQUE.lock().unwrap() = false;
|
||||
let users = USERS.lock().unwrap();
|
||||
let arr: Vec<JsonValue> = users.iter().map(|u| u.to_json()).collect();
|
||||
let json_str = JsonValue::Array(arr).dump();
|
||||
|
||||
save_file("", "users.json", &json_str);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn clear() {
|
||||
let mut users = USERS.lock().unwrap();
|
||||
users.clear();
|
||||
*UNIQUE.lock().unwrap() = true;
|
||||
}
|
||||
|
||||
pub async fn load_users() -> io::Result<()> {
|
||||
|
|
@ -132,7 +137,7 @@ pub async fn load_users() -> io::Result<()> {
|
|||
}
|
||||
}
|
||||
if *UNIQUE.lock().unwrap() {
|
||||
save_users().ok();
|
||||
save_users();
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,7 +18,9 @@ impl ConfigUtil {
|
|||
unique: false,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn clear(&mut self) {
|
||||
self.config = JsonValue::new_object();
|
||||
}
|
||||
pub fn load(&mut self) {
|
||||
let s = load_file("", "config.json");
|
||||
if !s.is_empty() {
|
||||
|
|
|
|||
Loading…
Reference in a new issue