1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
use crate::OutputProtocol;
use crate::debug_log;
use std::sync::Arc;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tokio::sync::Mutex;
#[derive(Debug)]
pub struct TCPOutputProtocol {
port: u16,
clients: Arc<Mutex<Vec<tokio::net::tcp::OwnedWriteHalf>>>,
}
impl TCPOutputProtocol {
pub fn new(port: u16) -> Self {
Self {
port,
clients: Arc::new(Mutex::new(Vec::new())),
}
}
}
impl OutputProtocol for TCPOutputProtocol {
async fn init(&self) {
let port = self.port;
let clients = self.clients.clone();
tokio::spawn(async move {
let addr = format!("0.0.0.0:{}", port);
let listener = match TcpListener::bind(&addr).await {
Ok(l) => l,
Err(e) => {
eprintln!("[TCP] Failed to bind to {}: {}", addr, e);
return;
}
};
debug_log!("[TCP] Listening on {}", addr);
loop {
match listener.accept().await {
Ok((stream, peer)) => {
debug_log!("[TCP] Client connected: {}", peer);
let (_, write_half) = stream.into_split();
clients.lock().await.push(write_half);
}
Err(e) => {
debug_log!("[TCP] Accept error: {}", e);
}
}
}
});
}
async fn send(self: Arc<Self>, message: &str) {
let mut clients = self.clients.lock().await;
let mut i = 0;
while i < clients.len() {
let mut write_half = clients.remove(i);
let bytes = format!("{}\n", message);
match write_half.write_all(bytes.as_bytes()).await {
Ok(_) => {
// Re-insert at the end if successful
clients.insert(i, write_half);
i += 1;
}
Err(_) => {
// Client disconnected, drop it
debug_log!("[TCP] Client disconnected, removing");
}
}
}
}
}
|
