1use std::collections::HashMap;
9use std::fs::File;
10use std::io::{BufRead, BufReader as StdBufReader};
11use std::path::Path;
12use std::sync::Arc;
13use std::time::Duration;
14use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
15use tokio::net::{TcpStream, UnixStream};
16use tokio::sync::{mpsc, oneshot, Mutex};
17use tokio::time::sleep;
18use tokio_native_tls::native_tls::TlsConnector;
19
20type IoHalves = (
21 Box<dyn tokio::io::AsyncRead + Send + Unpin>,
22 Box<dyn tokio::io::AsyncWrite + Send + Unpin>,
23);
24
25#[derive(Clone)]
26pub enum ConnectionAddress {
27 Tcp(String, u16),
28 Unix(String),
29}
30
31fn load_external_config(config_path: &str) -> HashMap<String, String> {
32 let mut config = HashMap::new();
33 let path = Path::new(config_path);
34 if !path.exists() {
35 return config;
36 }
37
38 if let Ok(file) = File::open(path) {
39 if let Ok(json_val) = serde_json::from_reader::<_, serde_json::Value>(&file) {
40 if let Some(obj) = json_val.as_object() {
41 for (k, v) in obj {
42 if let Some(s) = v.as_str() {
43 config.insert(k.clone(), s.to_string());
44 } else if let Some(i) = v.as_i64() {
45 config.insert(k.clone(), i.to_string());
46 } else if let Some(b) = v.as_bool() {
47 config.insert(k.clone(), b.to_string());
48 }
49 }
50 return config;
51 }
52 }
53 }
54
55 if let Ok(file) = File::open(path) {
56 let reader = StdBufReader::new(file);
57 for line in reader.lines().map_while(Result::ok) {
58 let line = line.trim();
59 if line.is_empty() || line.starts_with('#') {
60 continue;
61 }
62 if let Some(pos) = line.find('=') {
63 let key = line[..pos].trim().to_string();
64 let val = line[pos + 1..].trim().to_string();
65 config.insert(key, val);
66 }
67 }
68 }
69 config
70}
71
72pub struct NexusClient {
73 client_id: Option<String>,
74 retry_interval: Duration,
75 auth_token: Option<String>,
76 pub address: ConnectionAddress,
77 tx: Arc<Mutex<Option<mpsc::UnboundedSender<String>>>>,
78 list_resolvers: Arc<Mutex<Vec<oneshot::Sender<Vec<String>>>>>,
79 listeners: Arc<Mutex<Vec<mpsc::UnboundedSender<serde_json::Value>>>>,
80 on_reconnect_listeners: Arc<Mutex<Vec<mpsc::UnboundedSender<()>>>>,
81 closed: Arc<Mutex<bool>>,
82 pub is_windows: bool,
83 pub use_ssl: bool,
84 pub ssl_insecure: bool,
85}
86
87impl NexusClient {
88 pub fn new(retry_interval: Duration, config_path: &str, auth_token: Option<String>) -> Self {
89 let config = load_external_config(config_path);
90 let is_windows = cfg!(target_os = "windows");
91
92 let token = auth_token.or_else(|| config.get("auth_token").cloned());
93
94 let address = if is_windows {
95 let host = config
96 .get("host")
97 .cloned()
98 .unwrap_or_else(|| "127.0.0.1".to_string());
99 let port = config
100 .get("port")
101 .and_then(|p| p.parse::<u16>().ok())
102 .unwrap_or(50505);
103 ConnectionAddress::Tcp(host, port)
104 } else {
105 let path = config
106 .get("socket_path")
107 .or_else(|| config.get("socketPath"))
108 .cloned()
109 .unwrap_or_else(|| "/tmp/nerve.sock".to_string());
110 ConnectionAddress::Unix(path)
111 };
112
113 Self {
114 client_id: None,
115 retry_interval,
116 auth_token: token,
117 address,
118 tx: Arc::new(Mutex::new(None)),
119 list_resolvers: Arc::new(Mutex::new(Vec::new())),
120 listeners: Arc::new(Mutex::new(Vec::new())),
121 on_reconnect_listeners: Arc::new(Mutex::new(Vec::new())),
122 closed: Arc::new(Mutex::new(false)),
123 is_windows,
124 use_ssl: config
125 .get("use_ssl")
126 .or_else(|| config.get("useSSL"))
127 .map(|v| v.to_lowercase() == "true" || v == "1")
128 .unwrap_or(false),
129 ssl_insecure: config
130 .get("ssl_insecure")
131 .or_else(|| config.get("sslInsecure"))
132 .map(|v| v.to_lowercase() == "true" || v == "1")
133 .unwrap_or(false),
134 }
135 }
136
137 pub async fn connect(
138 &mut self,
139 client_id: &str,
140 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
141 self.client_id = Some(client_id.to_string());
142 let (connect_ok_tx, connect_ok_rx) = oneshot::channel();
143
144 let client_id_str = client_id.to_string();
145 let retry_interval = self.retry_interval;
146 let address = self.address.clone();
147 let auth_token = self.auth_token.clone();
148 let list_resolvers = self.list_resolvers.clone();
149 let listeners = self.listeners.clone();
150 let on_reconnect_listeners = self.on_reconnect_listeners.clone();
151 let tx_shared = self.tx.clone();
152 let closed = self.closed.clone();
153 let is_windows = self.is_windows;
154 let use_ssl = self.use_ssl;
155 let ssl_insecure = self.ssl_insecure;
156
157 tokio::spawn(async move {
158 let mut first_connect = Some(connect_ok_tx);
159
160 while !*closed.lock().await {
161 let conn_result: Result<IoHalves, Box<dyn std::error::Error + Send + Sync>> =
162 if is_windows {
163 match &address {
164 ConnectionAddress::Tcp(host, port) => {
165 match TcpStream::connect(format!("{}:{}", host, port)).await {
166 Ok(stream) => {
167 if use_ssl {
168 let mut builder = TlsConnector::builder();
169 if ssl_insecure {
170 builder
171 .danger_accept_invalid_certs(true)
172 .danger_accept_invalid_hostnames(true);
173 }
174 match builder.build() {
175 Ok(cx) => {
176 let connector =
177 tokio_native_tls::TlsConnector::from(cx);
178 match connector.connect(host, stream).await {
179 Ok(tls_stream) => {
180 let (r, w) =
181 tokio::io::split(tls_stream);
182 Ok((Box::new(r), Box::new(w)))
183 }
184 Err(e) => Err(e.into()),
185 }
186 }
187 Err(e) => Err(e.into()),
188 }
189 } else {
190 let (r, w) = tokio::io::split(stream);
191 Ok((Box::new(r), Box::new(w)))
192 }
193 }
194 Err(e) => Err(e.into()),
195 }
196 }
197 _ => Err(std::io::Error::other("Invalid address for Windows").into()),
198 }
199 } else {
200 match &address {
201 ConnectionAddress::Unix(path) => {
202 match UnixStream::connect(path).await {
203 Ok(stream) => {
204 let (r, w) = tokio::io::split(stream);
205 Ok((Box::new(r), Box::new(w)))
206 }
207 Err(e) => Err(e.into()),
208 }
209 }
210 _ => Err(std::io::Error::other("Invalid address for Unix").into()),
211 }
212 };
213
214 let (read_half, mut write_half) = match conn_result {
215 Ok(halves) => halves,
216 Err(e) => {
217 println!(
218 "[NERVE] Hub unavailable ({}). Retrying in {}s...",
219 e,
220 retry_interval.as_secs_f32()
221 );
222 sleep(retry_interval).await;
223 continue;
224 }
225 };
226
227 let reg_msg = if let Some(token) = &auth_token {
228 serde_json::json!({
229 "type": "register",
230 "id": client_id_str,
231 "token": token
232 })
233 } else {
234 serde_json::json!({
235 "type": "register",
236 "id": client_id_str
237 })
238 };
239
240 if write_half
241 .write_all(format!("{}\n", reg_msg).as_bytes())
242 .await
243 .is_err()
244 {
245 sleep(retry_interval).await;
246 continue;
247 }
248
249 let mut reader = BufReader::new(read_half);
250 let mut line = String::new();
251
252 let handshake_timeout = sleep(Duration::from_secs(5));
253 tokio::pin!(handshake_timeout);
254
255 let mut handshake_ok = false;
256 let mut auth_failed = false;
257
258 tokio::select! {
259 res = reader.read_line(&mut line) => {
260 if let Ok(n) = res {
261 if n > 0 {
262 if let Ok(val) = serde_json::from_str::<serde_json::Value>(&line) {
263 if val["type"] == "registered" {
264 if val["status"] == "success" {
265 handshake_ok = true;
266 } else if val["status"] == "failed" && val["reason"] == "auth" {
267 auth_failed = true;
268 }
269 }
270 }
271 }
272 }
273 }
274 _ = &mut handshake_timeout => {}
275 }
276
277 if auth_failed {
278 println!(
279 "[NERVE] Connected to hub as '{}' failed (auth).",
280 client_id_str
281 );
282 *closed.lock().await = true;
283 if let Some(chan) = first_connect.take() {
284 let _ = chan.send(Err("Authentication failed".into()));
285 }
286 break;
287 }
288
289 if !handshake_ok {
290 sleep(retry_interval).await;
291 continue;
292 }
293
294 println!("[NERVE] Connected to hub as '{}'.", client_id_str);
295
296 let (write_tx, mut write_rx) = mpsc::unbounded_channel::<String>();
297 *tx_shared.lock().await = Some(write_tx);
298
299 let is_reconnect = first_connect.is_none();
300 if let Some(chan) = first_connect.take() {
301 let _ = chan.send(Ok(()));
302 }
303
304 if is_reconnect {
305 let r_listeners = on_reconnect_listeners.lock().await;
306 for listener in r_listeners.iter() {
307 let _ = listener.send(());
308 }
309 }
310
311 let read_resolvers = list_resolvers.clone();
312 let read_listeners = listeners.clone();
313 let read_closed = closed.clone();
314
315 let read_handle = tokio::spawn(async move {
316 let mut reader_line = String::new();
317 while !*read_closed.lock().await {
318 reader_line.clear();
319 match reader.read_line(&mut reader_line).await {
320 Ok(n) if n > 0 => {
321 if let Ok(val) =
322 serde_json::from_str::<serde_json::Value>(&reader_line)
323 {
324 let msg_type = val["type"].as_str().unwrap_or("");
325 if msg_type == "ping" || msg_type == "pong" {
326 continue;
327 }
328 if msg_type == "list" {
329 let clients: Vec<String> = val["clients"]
330 .as_array()
331 .unwrap_or(&Vec::new())
332 .iter()
333 .filter_map(|c| c.as_str().map(|s| s.to_string()))
334 .collect();
335
336 let mut resolvers = read_resolvers.lock().await;
337 if !resolvers.is_empty() {
338 let res = resolvers.remove(0);
339 let _ = res.send(clients);
340 }
341 } else {
342 let l_list = read_listeners.lock().await;
343 for listener in l_list.iter() {
344 let _ = listener.send(val.clone());
345 }
346 }
347 }
348 }
349 _ => break,
350 }
351 }
352 });
353
354 let write_closed = closed.clone();
355 let write_handle = tokio::spawn(async move {
356 while !*write_closed.lock().await {
357 if let Some(msg) = write_rx.recv().await {
358 if write_half.write_all(msg.as_bytes()).await.is_err() {
359 break;
360 }
361 } else {
362 break;
363 }
364 }
365 });
366
367 tokio::select! {
368 _ = read_handle => {}
369 _ = write_handle => {}
370 }
371
372 *tx_shared.lock().await = None;
373 if *closed.lock().await {
374 break;
375 }
376 println!("[NERVE] Hub connection lost. Reconnecting...");
377 }
378 });
379
380 match connect_ok_rx.await {
381 Ok(Ok(())) => Ok(()),
382 Ok(Err(e)) => Err(e),
383 Err(_) => Err("Connection thread panicked".into()),
384 }
385 }
386
387 pub fn disconnect(&self) {
388 if let Ok(mut closed_guard) = self.closed.try_lock() {
389 *closed_guard = true;
390 }
391 if let Ok(mut tx_guard) = self.tx.try_lock() {
392 *tx_guard = None;
393 }
394 if let Some(id) = &self.client_id {
395 println!("[NERVE] '{}' disconnected.", id);
396 }
397 }
398
399 fn send_raw(
400 &self,
401 msg: serde_json::Value,
402 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
403 let tx_lock = self.tx.try_lock()?;
404 if let Some(tx) = &*tx_lock {
405 tx.send(format!("{}\n", msg))?;
406 Ok(())
407 } else {
408 Err("Not connected to hub.".into())
409 }
410 }
411
412 pub fn send(
413 &self,
414 to: &str,
415 payload: serde_json::Value,
416 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
417 self.send_raw(serde_json::json!({
418 "type": "send",
419 "to": to,
420 "payload": payload
421 }))
422 }
423
424 pub fn broadcast(
425 &self,
426 payload: serde_json::Value,
427 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
428 self.send_raw(serde_json::json!({
429 "type": "broadcast",
430 "payload": payload
431 }))
432 }
433
434 pub async fn list_clients(
435 &self,
436 ) -> Result<Vec<String>, Box<dyn std::error::Error + Send + Sync>> {
437 let (res_tx, res_rx) = oneshot::channel();
438 {
439 let mut resolvers = self.list_resolvers.lock().await;
440 resolvers.push(res_tx);
441 }
442
443 self.send_raw(serde_json::json!({ "type": "list" }))?;
444
445 tokio::select! {
446 res = res_rx => {
447 match res {
448 Ok(clients) => Ok(clients),
449 Err(_) => Err("Request cancelled".into())
450 }
451 }
452 _ = sleep(Duration::from_secs(2)) => {
453 Err("Timeout listing clients".into())
454 }
455 }
456 }
457
458 pub async fn listen<F>(
459 &self,
460 callback: F,
461 on_reconnect: Option<Box<dyn Fn() + Send + Sync + 'static>>,
462 ) where
463 F: Fn(serde_json::Value) + Send + Sync + 'static,
464 {
465 let (msg_tx, mut msg_rx) = mpsc::unbounded_channel::<serde_json::Value>();
466 {
467 let mut list = self.listeners.lock().await;
468 list.push(msg_tx);
469 }
470
471 tokio::spawn(async move {
472 while let Some(msg) = msg_rx.recv().await {
473 callback(msg);
474 }
475 });
476
477 if let Some(cb) = on_reconnect {
478 let (rec_tx, mut rec_rx) = mpsc::unbounded_channel::<()>();
479 {
480 let mut list = self.on_reconnect_listeners.lock().await;
481 list.push(rec_tx);
482 }
483 tokio::spawn(async move {
484 while rec_rx.recv().await.is_some() {
485 cb();
486 }
487 });
488 }
489 }
490}
491
492pub struct SyncNexusClient {
493 async_client: Arc<Mutex<NexusClient>>,
494 rt: tokio::runtime::Runtime,
495}
496
497impl SyncNexusClient {
498 pub fn new(retry_interval: Duration, config_path: &str, auth_token: Option<String>) -> Self {
499 let rt = tokio::runtime::Builder::new_multi_thread()
500 .enable_all()
501 .build()
502 .unwrap();
503 let async_client =
504 rt.block_on(async { NexusClient::new(retry_interval, config_path, auth_token) });
505 Self {
506 async_client: Arc::new(Mutex::new(async_client)),
507 rt,
508 }
509 }
510
511 pub fn connect(&self, client_id: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
512 let client = self.async_client.clone();
513 let client_id = client_id.to_string();
514 self.rt.block_on(async move {
515 let mut guard = client.lock().await;
516 guard.connect(&client_id).await
517 })
518 }
519
520 pub fn disconnect(&self) {
521 let client = self.async_client.clone();
522 self.rt.block_on(async move {
523 let guard = client.lock().await;
524 guard.disconnect();
525 });
526 }
527
528 pub fn send(
529 &self,
530 to: &str,
531 payload: serde_json::Value,
532 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
533 let client = self.async_client.clone();
534 let to = to.to_string();
535 self.rt.block_on(async move {
536 let guard = client.lock().await;
537 guard.send(&to, payload)
538 })
539 }
540
541 pub fn broadcast(
542 &self,
543 payload: serde_json::Value,
544 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
545 let client = self.async_client.clone();
546 self.rt.block_on(async move {
547 let guard = client.lock().await;
548 guard.broadcast(payload)
549 })
550 }
551
552 pub fn list_clients(&self) -> Result<Vec<String>, Box<dyn std::error::Error + Send + Sync>> {
553 let client = self.async_client.clone();
554 self.rt.block_on(async move {
555 let guard = client.lock().await;
556 guard.list_clients().await
557 })
558 }
559
560 pub fn listen<F>(
561 &self,
562 callback: F,
563 on_reconnect: Option<Box<dyn Fn() + Send + Sync + 'static>>,
564 ) where
565 F: Fn(serde_json::Value) + Send + Sync + 'static,
566 {
567 let client = self.async_client.clone();
568
569 let (tx, rx) = std::sync::mpsc::channel();
570 let (rec_tx, rec_rx) = std::sync::mpsc::channel();
571
572 let has_reconnect = on_reconnect.is_some();
573
574 self.rt.block_on(async move {
575 let guard = client.lock().await;
576
577 let on_rec: Option<Box<dyn Fn() + Send + Sync + 'static>> = if has_reconnect {
578 Some(Box::new(move || {
579 let _ = rec_tx.send(());
580 }))
581 } else {
582 None
583 };
584
585 guard
586 .listen(
587 move |msg| {
588 let _ = tx.send(msg);
589 },
590 on_rec,
591 )
592 .await;
593 });
594
595 std::thread::spawn(move || {
596 while let Ok(msg) = rx.recv() {
597 callback(msg);
598 }
599 });
600
601 if let Some(cb) = on_reconnect {
602 std::thread::spawn(move || {
603 while rec_rx.recv().is_ok() {
604 cb();
605 }
606 });
607 }
608 }
609}
610
611#[cfg(test)]
612mod tests {
613 use super::*;
614 use std::io::Write;
615 use tempfile::NamedTempFile;
616
617 #[test]
618 fn test_load_external_config_json() {
619 let mut file = NamedTempFile::new().unwrap();
620 let json_content = r#"{
621 "str_key": "value",
622 "int_key": 42,
623 "bool_key": true
624 }"#;
625 writeln!(file, "{}", json_content).unwrap();
626
627 let config = load_external_config(file.path().to_str().unwrap());
628 assert_eq!(config.get("str_key").unwrap(), "value");
629 assert_eq!(config.get("int_key").unwrap(), "42");
630 assert_eq!(config.get("bool_key").unwrap(), "true");
631 }
632
633 #[test]
634 fn test_load_external_config_env() {
635 let mut file = NamedTempFile::new().unwrap();
636 let env_content = "
637# this is a comment
638KEY1=value1
639KEY2 = value2
640";
641 writeln!(file, "{}", env_content).unwrap();
642
643 let config = load_external_config(file.path().to_str().unwrap());
644 assert_eq!(config.get("KEY1").unwrap(), "value1");
645 assert_eq!(config.get("KEY2").unwrap(), "value2");
646 }
647
648 #[test]
649 fn test_load_external_config_non_existent() {
650 let config = load_external_config("/non/existent/path/that/should/not/exist.json");
651 assert!(config.is_empty());
652 }
653}