made the server and the client/server communication functionality work.
This commit is contained in:
+178
-241
@@ -1,31 +1,20 @@
|
||||
use crate::{ToolMessage::Input, *};
|
||||
use crate::*;
|
||||
use crossterm::{
|
||||
cursor,
|
||||
event::{self, DisableMouseCapture, EnableMouseCapture, Event, KeyCode, KeyEventKind},
|
||||
execute,
|
||||
terminal::{EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode},
|
||||
};
|
||||
use iced::keyboard::key::Code::Sleep;
|
||||
use ratatui::{
|
||||
prelude::*,
|
||||
widgets::{Block, Borders, List, ListItem, ListState, Paragraph},
|
||||
};
|
||||
use std::sync::Arc;
|
||||
use std::sync::Mutex;
|
||||
use std::time::Duration;
|
||||
use std::{error::Error, time::Instant};
|
||||
use std::{
|
||||
io::{Read, Write},
|
||||
usize,
|
||||
};
|
||||
use std::{net::TcpStream, thread};
|
||||
|
||||
pub enum ServerBrokerCmd {
|
||||
ConnectServer(usize),
|
||||
RegisterActionOutput((Server, String)),
|
||||
Disconnect(Server),
|
||||
DisconnectAll,
|
||||
Exit,
|
||||
}
|
||||
use std::{io::Write, usize};
|
||||
use std::{sync::Arc, thread::sleep};
|
||||
use std::{sync::Mutex, thread::spawn};
|
||||
|
||||
pub fn get_user_input(prompt: &str) -> Result<String, Box<dyn Error>> {
|
||||
println!("{}", prompt);
|
||||
@@ -34,158 +23,6 @@ pub fn get_user_input(prompt: &str) -> Result<String, Box<dyn Error>> {
|
||||
return Ok(response.trim().to_string());
|
||||
}
|
||||
|
||||
pub fn server_broker(
|
||||
rx: Receiver<ServerBrokerCmd>,
|
||||
tx: Sender<ToolMessage>,
|
||||
servers: Arc<Mutex<Vec<Server>>>,
|
||||
) {
|
||||
loop {
|
||||
if let Ok(msg) = rx.try_recv() {
|
||||
match msg {
|
||||
ServerBrokerCmd::ConnectServer(id) => {
|
||||
let _ = tx.send(ToolMessage::Output((
|
||||
0,
|
||||
format!("attempting to connect to server {}", id),
|
||||
)));
|
||||
if let Ok(mut lock) = servers.lock() {
|
||||
lock.iter_mut().for_each(|s| {
|
||||
let _ =
|
||||
tx.send(ToolMessage::Output((0, format!("got server list lock!"))));
|
||||
if s.id == id {
|
||||
let _ = tx.send(ToolMessage::Output((
|
||||
0,
|
||||
format!("Found the server for {}: {}", s.id, s.address),
|
||||
)));
|
||||
if let Ok(mut stream) = TcpStream::connect(s.address.clone()) {
|
||||
let _ = tx.send(ToolMessage::Output((
|
||||
0,
|
||||
String::from("TCP successful to server!"),
|
||||
)));
|
||||
let _ = stream.write(format!("HELLO|{}\n", s.id).as_bytes());
|
||||
let _ = tx.send(ToolMessage::Output((
|
||||
0,
|
||||
String::from("Hello message sent!"),
|
||||
)));
|
||||
let mut buf = [0; 1024];
|
||||
if let Ok(bytes_read) = stream.read(&mut buf) {
|
||||
let response = String::from_utf8_lossy(&buf[..bytes_read]);
|
||||
if let Some((_, id)) = response.split_once("|") {
|
||||
if let Ok(id) = id.trim().parse::<usize>() {
|
||||
s.client_id = id;
|
||||
s.connected = true;
|
||||
let _ = tx.send(ToolMessage::Output((
|
||||
0,
|
||||
String::from("got client ID from server!"),
|
||||
)));
|
||||
} else {
|
||||
let _ = tx.send(ToolMessage::Output((
|
||||
0,
|
||||
String::from(
|
||||
"failed to get response from server!",
|
||||
),
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
ServerBrokerCmd::Disconnect(server) => {
|
||||
if let Ok(mut lock) = servers.lock() {
|
||||
let mut rid = 0;
|
||||
let mut found = false;
|
||||
lock.iter().enumerate().for_each(|(id, s)| {
|
||||
if s.id == server.id {
|
||||
rid = id;
|
||||
found = true;
|
||||
}
|
||||
});
|
||||
if found {
|
||||
lock.remove(rid);
|
||||
}
|
||||
}
|
||||
}
|
||||
ServerBrokerCmd::DisconnectAll => {
|
||||
println!("todo");
|
||||
}
|
||||
ServerBrokerCmd::Exit => {
|
||||
println!("todo");
|
||||
break;
|
||||
}
|
||||
ServerBrokerCmd::RegisterActionOutput(text) => {
|
||||
if let Ok(mut lock) = servers.lock() {
|
||||
lock.iter_mut().for_each(|s| {
|
||||
if s.id == text.0.id {
|
||||
s.message_que.push(text.1.clone());
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Ok(mut lock) = servers.try_lock() {
|
||||
lock.iter_mut().for_each(|s| {
|
||||
if s.connected {
|
||||
if Instant::now().duration_since(s.last_check) >= s.timer {
|
||||
if let Ok(mut stream) = TcpStream::connect(s.address.clone()) {
|
||||
s.last_check = Instant::now();
|
||||
let mut success_ids = Vec::new();
|
||||
if s.message_que.len() > 0 {
|
||||
s.message_que.iter().enumerate().for_each(|(id, msg)| {
|
||||
if let Ok(_) = stream.write(
|
||||
format!("{}|{}\n", s.client_id.clone(), msg).as_bytes(),
|
||||
) {
|
||||
success_ids.push(id);
|
||||
}
|
||||
});
|
||||
} else {
|
||||
let _ = stream.write(format!("{}|PING", s.client_id).as_bytes());
|
||||
}
|
||||
for id in success_ids {
|
||||
s.message_que.remove(id);
|
||||
}
|
||||
let mut buffer = [0; 1024];
|
||||
if let Ok(bytes_read) = stream.read(&mut buffer) {
|
||||
let response =
|
||||
String::from_utf8_lossy(&buffer[..bytes_read]).to_string();
|
||||
let (_, action_string) =
|
||||
response.split_once("|").unwrap_or(("ACTIONS", "NONE"));
|
||||
let actions: Vec<&str> = action_string.split(",").collect();
|
||||
for action in actions {
|
||||
let (action, data) =
|
||||
action.split_once("|").unwrap_or(("PING", "NONE"));
|
||||
match action {
|
||||
"OUTPUT" => {
|
||||
let _ = tx.send(ToolMessage::Output((
|
||||
0,
|
||||
format!("FROM SERVER: {}", data),
|
||||
)));
|
||||
}
|
||||
"CMD" => {
|
||||
let (rid_str, cmd) =
|
||||
data.split_once("|").unwrap_or(("0", "NONE"));
|
||||
if let Ok(rid) = rid_str.parse::<usize>() {
|
||||
let _ = tx.send(ToolMessage::Input((
|
||||
rid,
|
||||
cmd.to_string(),
|
||||
)));
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
}
|
||||
|
||||
pub fn run_tui(
|
||||
mut state: AppState,
|
||||
main_rx: Receiver<ToolMessage>,
|
||||
@@ -198,12 +35,6 @@ pub fn run_tui(
|
||||
cursor::Hide,
|
||||
EnableMouseCapture
|
||||
)?;
|
||||
let (server_tx, server_rx) = channel();
|
||||
let main_tx = state.main_tx.clone();
|
||||
let server_clone = state.servers.clone();
|
||||
std::thread::spawn(move || {
|
||||
server_broker(server_rx, main_tx, server_clone);
|
||||
});
|
||||
let backend = CrosstermBackend::new(&stdout);
|
||||
let mut terminal = Terminal::new(backend)?;
|
||||
let (event_tx, event_rx) = channel::<AppEvent>();
|
||||
@@ -250,6 +81,7 @@ pub fn run_tui(
|
||||
project_list_state.select(Some(0));
|
||||
}
|
||||
let mut history_index = state.history.len();
|
||||
let mut handles = Vec::new();
|
||||
loop {
|
||||
terminal.draw(|f| {
|
||||
let main_chunks = Layout::default()
|
||||
@@ -402,17 +234,22 @@ pub fn run_tui(
|
||||
}
|
||||
ToolMessage::Input(cmd) => {
|
||||
state.log.push(cmd.1.clone());
|
||||
if cmd.1.contains(" ") {
|
||||
let (gcmd, args) = cmd.1.split_once(" ").unwrap();
|
||||
state.execute_command(gcmd, Some(args.to_string()), cmd.0)?;
|
||||
} else {
|
||||
state.execute_command(cmd.1.as_str(), None, cmd.0)?;
|
||||
}
|
||||
}
|
||||
ToolMessage::Output((rid, txt)) => {
|
||||
state.log.push(txt.clone());
|
||||
state.output.push(txt.clone());
|
||||
state.output_scroll = u16::MAX;
|
||||
if rid != 0 {
|
||||
if let Some(server) = state.selected_server.clone() {
|
||||
let _ = server_tx.send(ServerBrokerCmd::RegisterActionOutput((
|
||||
server,
|
||||
format!("OUTPUT|{}|{}", rid, txt),
|
||||
)));
|
||||
if let Some(server) = state.servers.get(state.selected_server) {
|
||||
if let Ok(mut server) = server.lock() {
|
||||
server.message_que.push(format!("OUTPUT|{}", txt));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -482,12 +319,121 @@ pub fn run_tui(
|
||||
terminal.clear()?;
|
||||
}
|
||||
ToolMessage::ConnectServer => {
|
||||
if let Some(server) = state.selected_server.clone() {
|
||||
let _ = server_tx.send(ServerBrokerCmd::ConnectServer(server.id));
|
||||
if let Some(server) = state.servers.get_mut(state.selected_server) {
|
||||
let mut address = String::new();
|
||||
let lock_res = server.try_lock();
|
||||
let mut connected = false;
|
||||
match lock_res {
|
||||
Ok(mut locked_server) => match locked_server.connect() {
|
||||
Ok(id) => {
|
||||
state.output.push(format!(
|
||||
"Server Connected! ID:{} ADDRESS:{}",
|
||||
id,
|
||||
locked_server.address.clone()
|
||||
));
|
||||
address = locked_server.address.clone();
|
||||
connected = true;
|
||||
}
|
||||
Err(e) => {
|
||||
state
|
||||
.output
|
||||
.push(format!("Error connecting to server {e}"));
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
state
|
||||
.output
|
||||
.push(format!("Error locking server state! {e}"));
|
||||
}
|
||||
}
|
||||
if connected {
|
||||
let server_clone = server.clone();
|
||||
let tx_clone = state.main_tx.clone();
|
||||
let mut timer = Duration::from_secs(5);
|
||||
let mut last_check = Instant::now();
|
||||
let server_handle = spawn(move || {
|
||||
let server = server_clone;
|
||||
loop {
|
||||
let mut stop = false;
|
||||
if Instant::now().duration_since(last_check) >= timer {
|
||||
let lock_res = server.try_lock();
|
||||
match lock_res {
|
||||
Ok(mut locked_server) => {
|
||||
locked_server.checkin();
|
||||
last_check = locked_server.last_check.clone();
|
||||
for action in locked_server.action_que.clone() {
|
||||
if action.contains("|") {
|
||||
let (action, data) =
|
||||
action.split_once("|").unwrap();
|
||||
match action.trim() {
|
||||
"OUTPUT" => {
|
||||
let _ = tx_clone.send(
|
||||
ToolMessage::Output((
|
||||
0,
|
||||
data.trim().to_string(),
|
||||
)),
|
||||
);
|
||||
}
|
||||
"CMD" => {
|
||||
let _ = tx_clone.send(
|
||||
ToolMessage::Input((
|
||||
locked_server
|
||||
.client_id
|
||||
.clone(),
|
||||
data.trim().to_string(),
|
||||
)),
|
||||
);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
} else {
|
||||
match action.as_str() {
|
||||
"disconnect" => {
|
||||
locked_server.connected = false;
|
||||
stop = true;
|
||||
}
|
||||
_ => {
|
||||
let _ = tx_clone.send(
|
||||
ToolMessage::Output((
|
||||
0,
|
||||
action.clone(),
|
||||
)),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
timer = locked_server.timer.clone();
|
||||
locked_server.action_que.clear();
|
||||
if locked_server.message_que.is_empty() {
|
||||
let id = locked_server.client_id.clone();
|
||||
locked_server
|
||||
.message_que
|
||||
.push(format!("{}|||PING\n", id));
|
||||
}
|
||||
}
|
||||
Err(_) => {}
|
||||
}
|
||||
}
|
||||
if stop {
|
||||
println!("{} disconnected!", address);
|
||||
break;
|
||||
}
|
||||
sleep(Duration::from_secs(1));
|
||||
}
|
||||
});
|
||||
handles.push(server_handle);
|
||||
}
|
||||
}
|
||||
}
|
||||
ToolMessage::SendServer(data) => {
|
||||
let _ = server_tx.send(ServerBrokerCmd::RegisterActionOutput(data));
|
||||
if let Some(server) = state.servers.get(state.selected_server) {
|
||||
if let Ok(mut server) = server.lock() {
|
||||
server
|
||||
.message_que
|
||||
.push(format!("{}|||{}\n", data.0, data.1));
|
||||
}
|
||||
}
|
||||
}
|
||||
ToolMessage::DisconnectAllServers => {
|
||||
println!("todo");
|
||||
@@ -530,73 +476,55 @@ pub fn run_tui(
|
||||
}
|
||||
let (command, args) =
|
||||
trimmed.split_once(' ').unwrap_or((&trimmed, ""));
|
||||
if state.selected_client != 0 {
|
||||
if let Some(server) = state.selected_server.clone() {
|
||||
let _ = server_tx.send(
|
||||
ServerBrokerCmd::RegisterActionOutput((
|
||||
server.clone(),
|
||||
format!(
|
||||
"CMD|{}|{}|{} {}",
|
||||
server.client_id,
|
||||
state.selected_client,
|
||||
command,
|
||||
args
|
||||
),
|
||||
)),
|
||||
);
|
||||
match command {
|
||||
"exit" | "quit" => break,
|
||||
"reload-modules" => {
|
||||
state.initialize_modules();
|
||||
state.output.push("Reloading module paths...".into());
|
||||
}
|
||||
} else {
|
||||
match command {
|
||||
"exit" | "quit" => break,
|
||||
"reload-modules" => {
|
||||
state.initialize_modules();
|
||||
state.output.push("Reloading module paths...".into());
|
||||
"help" => {
|
||||
let help_text = state.help.clone().join("\n");
|
||||
for line in help_text.lines() {
|
||||
state.output.push(line.to_string());
|
||||
}
|
||||
"help" => {
|
||||
let help_text = state.help.clone().join("\n");
|
||||
for line in help_text.lines() {
|
||||
state.output.push(line.to_string());
|
||||
}
|
||||
"new_project" | "np" => {
|
||||
if args.split_once(' ').is_some() {
|
||||
let _ = state.execute_command(
|
||||
command,
|
||||
Some(args.to_string()),
|
||||
0,
|
||||
);
|
||||
} else {
|
||||
state
|
||||
.output
|
||||
.push("Error: USAGE -> np <org> <name>".into());
|
||||
}
|
||||
}
|
||||
command_name => {
|
||||
if state.module_loader.commands.contains_key(command) {
|
||||
state.output.push(format!(
|
||||
"[Worker] Executing script '{}'...",
|
||||
command_name
|
||||
));
|
||||
if let Err(e) = state.execute_command(
|
||||
command,
|
||||
Some(args.to_string()),
|
||||
0,
|
||||
) {
|
||||
state
|
||||
.output
|
||||
.push(format!("[Error] Pipeline fail: {}", e));
|
||||
}
|
||||
}
|
||||
"new_project" | "np" => {
|
||||
if args.split_once(' ').is_some() {
|
||||
} else {
|
||||
if args == "" {
|
||||
let _ = state.execute_command(command, None, 0);
|
||||
} else {
|
||||
let _ = state.execute_command(
|
||||
command,
|
||||
Some(args.to_string()),
|
||||
0,
|
||||
);
|
||||
} else {
|
||||
state
|
||||
.output
|
||||
.push("Error: USAGE -> np <org> <name>".into());
|
||||
}
|
||||
}
|
||||
command_name => {
|
||||
if state.module_loader.commands.contains_key(command) {
|
||||
state.output.push(format!(
|
||||
"[Worker] Executing script '{}'...",
|
||||
command_name
|
||||
));
|
||||
if let Err(e) = state.execute_command(
|
||||
command,
|
||||
Some(args.to_string()),
|
||||
0,
|
||||
) {
|
||||
state.output.push(format!(
|
||||
"[Error] Pipeline fail: {}",
|
||||
e
|
||||
));
|
||||
}
|
||||
} else {
|
||||
if args == "" {
|
||||
let _ = state.execute_command(command, None, 0);
|
||||
} else {
|
||||
let _ = state.execute_command(
|
||||
command,
|
||||
Some(args.to_string()),
|
||||
0,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -700,5 +628,14 @@ pub fn run_tui(
|
||||
cursor::Show,
|
||||
DisableMouseCapture
|
||||
)?;
|
||||
println!("disconnecting from servers...");
|
||||
state.servers.iter().for_each(|server| {
|
||||
if let Ok(mut server_lock) = server.lock() {
|
||||
server_lock.action_que.push("disconnect".to_string());
|
||||
}
|
||||
});
|
||||
for handle in handles {
|
||||
handle.join().unwrap();
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user