rns_net/interface/
tcp_server.rs

1//! TCP server interface with HDLC framing.
2//!
3//! Accepts client connections and spawns per-client reader threads.
4//! Each client gets a dynamically allocated InterfaceId.
5//! Matches Python `TCPServerInterface` from `TCPInterface.py`.
6
7use std::io::{self, Read, Write};
8use std::net::{TcpListener, TcpStream};
9use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
10use std::sync::{Arc, Mutex};
11use std::thread;
12
13use rns_core::constants;
14use rns_core::transport::types::{IngressControlConfig, InterfaceId, InterfaceInfo};
15
16use crate::event::{Event, EventSender};
17use crate::hdlc;
18use crate::interface::{lock_or_recover, ListenerControl, Writer};
19
20/// Configuration for a TCP server interface.
21#[derive(Debug, Clone)]
22pub struct TcpServerConfig {
23    pub name: String,
24    pub listen_ip: String,
25    pub listen_port: u16,
26    pub interface_id: InterfaceId,
27    pub max_connections: Option<usize>,
28    pub ingress_control: IngressControlConfig,
29    pub runtime: Arc<Mutex<TcpServerRuntime>>,
30}
31
32#[derive(Debug, Clone)]
33pub struct TcpServerRuntime {
34    pub max_connections: Option<usize>,
35}
36
37impl TcpServerRuntime {
38    pub fn from_config(config: &TcpServerConfig) -> Self {
39        Self {
40            max_connections: config.max_connections,
41        }
42    }
43}
44
45#[derive(Debug, Clone)]
46pub struct TcpServerRuntimeConfigHandle {
47    pub interface_name: String,
48    pub runtime: Arc<Mutex<TcpServerRuntime>>,
49    pub startup: TcpServerRuntime,
50}
51
52impl Default for TcpServerConfig {
53    fn default() -> Self {
54        let mut config = TcpServerConfig {
55            name: String::new(),
56            listen_ip: "0.0.0.0".into(),
57            listen_port: 4242,
58            interface_id: InterfaceId(0),
59            max_connections: None,
60            ingress_control: IngressControlConfig::enabled(),
61            runtime: Arc::new(Mutex::new(TcpServerRuntime {
62                max_connections: None,
63            })),
64        };
65        let startup = TcpServerRuntime::from_config(&config);
66        config.runtime = Arc::new(Mutex::new(startup));
67        config
68    }
69}
70
71/// Writer that sends HDLC-framed data over a TCP stream.
72struct TcpServerWriter {
73    stream: TcpStream,
74}
75
76impl Writer for TcpServerWriter {
77    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
78        self.stream.write_all(&hdlc::frame(data))
79    }
80}
81
82/// Start a TCP server. Spawns a listener thread that accepts connections
83/// and per-client reader threads. Returns immediately.
84///
85/// `next_id` is shared with the node for allocating unique InterfaceIds
86/// for each connected client.
87pub fn start(
88    config: TcpServerConfig,
89    tx: EventSender,
90    next_id: Arc<AtomicU64>,
91) -> io::Result<ListenerControl> {
92    let addr = format!("{}:{}", config.listen_ip, config.listen_port);
93    let listener = TcpListener::bind(&addr)?;
94    listener.set_nonblocking(true)?;
95
96    log::info!("[{}] TCP server listening on {}", config.name, addr);
97
98    let name = config.name.clone();
99    let runtime = Arc::clone(&config.runtime);
100    let ingress_control = config.ingress_control;
101    let active_connections = Arc::new(AtomicUsize::new(0));
102    let control = ListenerControl::new();
103    let listener_control = control.clone();
104    thread::Builder::new()
105        .name(format!("tcp-server-{}", config.interface_id.0))
106        .spawn(move || {
107            listener_loop(
108                listener,
109                name,
110                tx,
111                next_id,
112                runtime,
113                ingress_control,
114                active_connections,
115                listener_control,
116            );
117        })?;
118
119    Ok(control)
120}
121
122/// Listener thread: accepts connections and spawns reader threads.
123fn listener_loop(
124    listener: TcpListener,
125    name: String,
126    tx: EventSender,
127    next_id: Arc<AtomicU64>,
128    runtime: Arc<Mutex<TcpServerRuntime>>,
129    ingress_control: IngressControlConfig,
130    active_connections: Arc<AtomicUsize>,
131    control: ListenerControl,
132) {
133    loop {
134        if control.should_stop() {
135            log::info!("[{}] listener stopping", name);
136            return;
137        }
138
139        let stream_result = listener.accept().map(|(stream, _)| stream);
140        let stream = match stream_result {
141            Ok(s) => s,
142            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
143                thread::sleep(std::time::Duration::from_millis(50));
144                continue;
145            }
146            Err(e) => {
147                log::warn!("[{}] accept failed: {}", name, e);
148                continue;
149            }
150        };
151
152        let max_connections = lock_or_recover(&runtime, "tcp server runtime").max_connections;
153        if let Some(max) = max_connections {
154            if active_connections.load(Ordering::Relaxed) >= max {
155                let peer = stream.peer_addr().ok();
156                log::warn!(
157                    "[{}] max connections ({}) reached, rejecting {:?}",
158                    name,
159                    max,
160                    peer
161                );
162                drop(stream);
163                continue;
164            }
165        }
166
167        active_connections.fetch_add(1, Ordering::Relaxed);
168
169        let client_id = InterfaceId(next_id.fetch_add(1, Ordering::Relaxed));
170        let peer_addr = stream.peer_addr().ok();
171
172        log::info!(
173            "[{}] client connected: {:?} → id {}",
174            name,
175            peer_addr,
176            client_id.0
177        );
178
179        // Set TCP_NODELAY on the client socket
180        if let Err(e) = stream.set_nodelay(true) {
181            log::warn!("[{}] set_nodelay failed: {}", name, e);
182        }
183
184        // Clone stream for writer
185        let writer_stream = match stream.try_clone() {
186            Ok(s) => s,
187            Err(e) => {
188                log::warn!("[{}] failed to clone stream: {}", name, e);
189                continue;
190            }
191        };
192
193        let writer: Box<dyn Writer> = Box::new(TcpServerWriter {
194            stream: writer_stream,
195        });
196
197        let info = InterfaceInfo {
198            id: client_id,
199            name: format!("TCPServerInterface/Client-{}", client_id.0),
200            mode: constants::MODE_FULL,
201            recursive_prs: false,
202            announces_from_internal: true,
203            out_capable: true,
204            in_capable: true,
205            bitrate: None,
206            airtime_profile: None,
207            announce_rate_target: None,
208            announce_rate_grace: 0,
209            announce_rate_penalty: 0.0,
210            announce_cap: constants::ANNOUNCE_CAP,
211            is_local_client: false,
212            wants_tunnel: false,
213            tunnel_id: None,
214            mtu: 65535,
215            ia_freq: 0.0,
216            ip_freq: 0.0,
217            op_freq: 0.0,
218            op_samples: 0,
219            started: 0.0,
220            ingress_control,
221        };
222
223        // Send InterfaceUp with InterfaceInfo for dynamic registration
224        if tx
225            .send(Event::InterfaceUp(client_id, Some(writer), Some(info)))
226            .is_err()
227        {
228            // Driver shut down
229            return;
230        }
231
232        // Spawn reader thread for this client
233        let client_tx = tx.clone();
234        let client_name = name.clone();
235        let client_active = active_connections.clone();
236        thread::Builder::new()
237            .name(format!("tcp-server-reader-{}", client_id.0))
238            .spawn(move || {
239                client_reader_loop(stream, client_id, client_name, client_tx, client_active);
240            })
241            .ok();
242    }
243}
244
245/// Per-client reader thread: reads HDLC frames, sends to driver.
246fn client_reader_loop(
247    mut stream: TcpStream,
248    id: InterfaceId,
249    name: String,
250    tx: EventSender,
251    active_connections: Arc<AtomicUsize>,
252) {
253    let mut decoder = hdlc::Decoder::new();
254    let mut buf = [0u8; 4096];
255
256    loop {
257        match stream.read(&mut buf) {
258            Ok(0) => {
259                log::info!("[{}] client {} disconnected", name, id.0);
260                active_connections.fetch_sub(1, Ordering::Relaxed);
261                let _ = tx.send(Event::InterfaceDown(id));
262                return;
263            }
264            Ok(n) => {
265                for frame in decoder.feed(&buf[..n]) {
266                    if tx
267                        .send(Event::Frame {
268                            interface_id: id,
269                            data: frame,
270                            rssi: None,
271                            snr: None,
272                        })
273                        .is_err()
274                    {
275                        // Driver shut down
276                        active_connections.fetch_sub(1, Ordering::Relaxed);
277                        return;
278                    }
279                }
280            }
281            Err(e) => {
282                log::warn!("[{}] client {} read error: {}", name, id.0, e);
283                active_connections.fetch_sub(1, Ordering::Relaxed);
284                let _ = tx.send(Event::InterfaceDown(id));
285                return;
286            }
287        }
288    }
289}
290
291// --- Factory implementation ---
292
293use super::{InterfaceConfigData, InterfaceFactory, StartContext, StartResult};
294use std::collections::HashMap;
295
296/// Factory for `TCPServerInterface`.
297pub struct TcpServerFactory;
298
299impl InterfaceFactory for TcpServerFactory {
300    fn type_name(&self) -> &str {
301        "TCPServerInterface"
302    }
303
304    fn parse_config(
305        &self,
306        name: &str,
307        id: InterfaceId,
308        params: &HashMap<String, String>,
309    ) -> Result<Box<dyn InterfaceConfigData>, String> {
310        let listen_ip = params
311            .get("listen_ip")
312            .cloned()
313            .unwrap_or_else(|| "0.0.0.0".into());
314        let listen_port = params
315            .get("listen_port")
316            .and_then(|v| v.parse().ok())
317            .unwrap_or(4242);
318        let max_connections = params.get("max_connections").and_then(|v| v.parse().ok());
319        let mut config = TcpServerConfig {
320            name: name.to_string(),
321            listen_ip,
322            listen_port,
323            interface_id: id,
324            max_connections,
325            ingress_control: IngressControlConfig::enabled(),
326            runtime: Arc::new(Mutex::new(TcpServerRuntime {
327                max_connections: None,
328            })),
329        };
330        let startup = TcpServerRuntime::from_config(&config);
331        config.runtime = Arc::new(Mutex::new(startup));
332        Ok(Box::new(config))
333    }
334
335    fn start(
336        &self,
337        config: Box<dyn InterfaceConfigData>,
338        ctx: StartContext,
339    ) -> io::Result<StartResult> {
340        let mut cfg = *config
341            .into_any()
342            .downcast::<TcpServerConfig>()
343            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "wrong config type"))?;
344        cfg.ingress_control = ctx.ingress_control;
345        let control = start(cfg, ctx.tx, ctx.next_dynamic_id)?;
346        Ok(StartResult::Listener {
347            control: Some(control),
348        })
349    }
350}
351
352pub(crate) fn runtime_handle_from_config(config: &TcpServerConfig) -> TcpServerRuntimeConfigHandle {
353    TcpServerRuntimeConfigHandle {
354        interface_name: config.name.clone(),
355        runtime: Arc::clone(&config.runtime),
356        startup: TcpServerRuntime::from_config(config),
357    }
358}
359
360#[cfg(test)]
361mod tests {
362    use super::*;
363    use std::net::TcpStream;
364    use std::sync::mpsc::RecvTimeoutError;
365    use std::time::Duration;
366
367    fn find_free_port() -> u16 {
368        TcpListener::bind("127.0.0.1:0")
369            .unwrap()
370            .local_addr()
371            .unwrap()
372            .port()
373    }
374
375    fn make_server_config(
376        port: u16,
377        interface_id: u64,
378        max_connections: Option<usize>,
379    ) -> TcpServerConfig {
380        let mut config = TcpServerConfig {
381            name: "test-server".into(),
382            listen_ip: "127.0.0.1".into(),
383            listen_port: port,
384            interface_id: InterfaceId(interface_id),
385            max_connections,
386            ingress_control: IngressControlConfig::enabled(),
387            runtime: Arc::new(Mutex::new(TcpServerRuntime {
388                max_connections: None,
389            })),
390        };
391        let startup = TcpServerRuntime::from_config(&config);
392        config.runtime = Arc::new(Mutex::new(startup));
393        config
394    }
395
396    #[test]
397    fn accept_connection() {
398        let port = find_free_port();
399        let (tx, rx) = crate::event::channel();
400        let next_id = Arc::new(AtomicU64::new(1000));
401
402        let config = make_server_config(port, 1, None);
403
404        start(config, tx, next_id).unwrap();
405
406        // Give server time to start listening
407        thread::sleep(Duration::from_millis(50));
408
409        // Connect a client
410        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
411
412        // Should receive InterfaceUp with InterfaceInfo (dynamic)
413        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
414        match event {
415            Event::InterfaceUp(id, writer, info) => {
416                assert_eq!(id, InterfaceId(1000));
417                assert!(writer.is_some());
418                assert!(info.is_some());
419            }
420            other => panic!("expected InterfaceUp, got {:?}", other),
421        }
422    }
423
424    #[test]
425    fn spawned_client_inherits_ingress_control_config() {
426        let port = find_free_port();
427        let (tx, rx) = crate::event::channel();
428        let next_id = Arc::new(AtomicU64::new(1100));
429
430        let mut config = make_server_config(port, 11, None);
431        config.ingress_control = IngressControlConfig::disabled();
432        config.ingress_control.max_held_announces = 17;
433        config.ingress_control.burst_hold = 1.5;
434        config.ingress_control.burst_freq_new = 2.5;
435        config.ingress_control.burst_freq = 3.5;
436        config.ingress_control.new_time = 4.5;
437        config.ingress_control.burst_penalty = 5.5;
438        config.ingress_control.held_release_interval = 6.5;
439
440        start(config, tx, next_id).unwrap();
441        thread::sleep(Duration::from_millis(50));
442
443        let _client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
444
445        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
446        match event {
447            Event::InterfaceUp(_, _, Some(info)) => {
448                assert!(!info.ingress_control.enabled);
449                assert_eq!(info.ingress_control.max_held_announces, 17);
450                assert_eq!(info.ingress_control.burst_hold, 1.5);
451                assert_eq!(info.ingress_control.burst_freq_new, 2.5);
452                assert_eq!(info.ingress_control.burst_freq, 3.5);
453                assert_eq!(info.ingress_control.new_time, 4.5);
454                assert_eq!(info.ingress_control.burst_penalty, 5.5);
455                assert_eq!(info.ingress_control.held_release_interval, 6.5);
456            }
457            other => panic!("expected InterfaceUp with InterfaceInfo, got {:?}", other),
458        }
459    }
460
461    #[test]
462    fn listener_stop_prevents_new_accepts() {
463        let port = find_free_port();
464        let (tx, rx) = crate::event::channel();
465        let next_id = Arc::new(AtomicU64::new(1500));
466
467        let config = make_server_config(port, 15, None);
468        let control = start(config, tx, next_id).unwrap();
469
470        thread::sleep(Duration::from_millis(50));
471        control.request_stop();
472        thread::sleep(Duration::from_millis(120));
473
474        let connect_result = TcpStream::connect(format!("127.0.0.1:{}", port));
475        if let Ok(stream) = connect_result {
476            drop(stream);
477        }
478
479        match rx.recv_timeout(Duration::from_millis(200)) {
480            Err(RecvTimeoutError::Timeout) | Err(RecvTimeoutError::Disconnected) => {}
481            other => panic!(
482                "expected no InterfaceUp after listener stop, got {:?}",
483                other
484            ),
485        }
486    }
487
488    #[test]
489    fn receive_frame_from_client() {
490        let port = find_free_port();
491        let (tx, rx) = crate::event::channel();
492        let next_id = Arc::new(AtomicU64::new(2000));
493
494        let config = make_server_config(port, 2, None);
495
496        start(config, tx, next_id).unwrap();
497        thread::sleep(Duration::from_millis(50));
498
499        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
500
501        // Drain InterfaceUp
502        let _ = rx.recv_timeout(Duration::from_secs(1)).unwrap();
503
504        // Send an HDLC frame (>= 19 bytes)
505        let payload: Vec<u8> = (0..32).collect();
506        let framed = hdlc::frame(&payload);
507        client.write_all(&framed).unwrap();
508
509        // Should receive Frame event
510        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
511        match event {
512            Event::Frame {
513                interface_id,
514                data,
515                rssi: _,
516                snr: _,
517            } => {
518                assert_eq!(interface_id, InterfaceId(2000));
519                assert_eq!(data, payload);
520            }
521            other => panic!("expected Frame, got {:?}", other),
522        }
523    }
524
525    #[test]
526    fn send_frame_to_client() {
527        let port = find_free_port();
528        let (tx, rx) = crate::event::channel();
529        let next_id = Arc::new(AtomicU64::new(3000));
530
531        let config = make_server_config(port, 3, None);
532
533        start(config, tx, next_id).unwrap();
534        thread::sleep(Duration::from_millis(50));
535
536        let mut client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
537        client
538            .set_read_timeout(Some(Duration::from_secs(2)))
539            .unwrap();
540
541        // Get the writer from InterfaceUp
542        let event = rx.recv_timeout(Duration::from_secs(1)).unwrap();
543        let mut writer = match event {
544            Event::InterfaceUp(_, Some(w), _) => w,
545            other => panic!("expected InterfaceUp with writer, got {:?}", other),
546        };
547
548        // Send a frame via writer
549        let payload: Vec<u8> = (0..24).collect();
550        writer.send_frame(&payload).unwrap();
551
552        // Read from client side
553        let mut buf = [0u8; 256];
554        let n = client.read(&mut buf).unwrap();
555        let expected = hdlc::frame(&payload);
556        assert_eq!(&buf[..n], &expected[..]);
557    }
558
559    #[test]
560    fn multiple_clients() {
561        let port = find_free_port();
562        let (tx, rx) = crate::event::channel();
563        let next_id = Arc::new(AtomicU64::new(4000));
564
565        let config = make_server_config(port, 4, None);
566
567        start(config, tx, next_id).unwrap();
568        thread::sleep(Duration::from_millis(50));
569
570        // Connect two clients
571        let _client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
572        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
573
574        // Collect InterfaceUp events
575        let mut ids = Vec::new();
576        for _ in 0..2 {
577            let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
578            match event {
579                Event::InterfaceUp(id, _, _) => ids.push(id),
580                other => panic!("expected InterfaceUp, got {:?}", other),
581            }
582        }
583
584        // Should have unique IDs
585        assert_eq!(ids.len(), 2);
586        assert_ne!(ids[0], ids[1]);
587    }
588
589    #[test]
590    fn client_disconnect() {
591        let port = find_free_port();
592        let (tx, rx) = crate::event::channel();
593        let next_id = Arc::new(AtomicU64::new(5000));
594
595        let config = make_server_config(port, 5, None);
596
597        start(config, tx, next_id).unwrap();
598        thread::sleep(Duration::from_millis(50));
599
600        let client = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
601
602        // Drain InterfaceUp
603        let _ = rx.recv_timeout(Duration::from_secs(1)).unwrap();
604
605        // Disconnect
606        drop(client);
607
608        // Should receive InterfaceDown
609        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
610        assert!(
611            matches!(event, Event::InterfaceDown(InterfaceId(5000))),
612            "expected InterfaceDown(5000), got {:?}",
613            event
614        );
615    }
616
617    #[test]
618    fn server_bind_port() {
619        let port = find_free_port();
620        let (tx, _rx) = crate::event::channel();
621        let next_id = Arc::new(AtomicU64::new(6000));
622
623        let config = make_server_config(port, 6, None);
624
625        // Should not error
626        start(config, tx, next_id).unwrap();
627    }
628
629    #[test]
630    fn max_connections_rejects_excess() {
631        let port = find_free_port();
632        let (tx, rx) = crate::event::channel();
633        let next_id = Arc::new(AtomicU64::new(7000));
634
635        let config = make_server_config(port, 7, Some(2));
636
637        start(config, tx, next_id).unwrap();
638        thread::sleep(Duration::from_millis(50));
639
640        // Connect two clients (at limit)
641        let _client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
642        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
643
644        // Drain both InterfaceUp events
645        for _ in 0..2 {
646            let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
647            assert!(matches!(event, Event::InterfaceUp(_, _, _)));
648        }
649
650        // Third connection should be accepted at TCP level but immediately dropped by server
651        let client3 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
652        client3
653            .set_read_timeout(Some(Duration::from_millis(500)))
654            .unwrap();
655
656        // Give server time to reject
657        thread::sleep(Duration::from_millis(100));
658
659        // Should NOT receive a third InterfaceUp
660        let result = rx.recv_timeout(Duration::from_millis(500));
661        assert!(
662            result.is_err(),
663            "expected no InterfaceUp for rejected connection, got {:?}",
664            result
665        );
666    }
667
668    #[test]
669    fn max_connections_allows_after_disconnect() {
670        let port = find_free_port();
671        let (tx, rx) = crate::event::channel();
672        let next_id = Arc::new(AtomicU64::new(7100));
673
674        let config = make_server_config(port, 71, Some(1));
675
676        start(config, tx, next_id).unwrap();
677        thread::sleep(Duration::from_millis(50));
678
679        // Connect first client
680        let client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
681        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
682        assert!(matches!(event, Event::InterfaceUp(_, _, _)));
683
684        // Disconnect first client
685        drop(client1);
686
687        // Wait for InterfaceDown
688        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
689        assert!(matches!(event, Event::InterfaceDown(_)));
690
691        // Now a new connection should be accepted
692        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
693        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
694        assert!(
695            matches!(event, Event::InterfaceUp(_, _, _)),
696            "expected InterfaceUp after slot freed, got {:?}",
697            event
698        );
699    }
700
701    #[test]
702    fn runtime_max_connections_updates_live() {
703        let port = find_free_port();
704        let (tx, rx) = crate::event::channel();
705        let next_id = Arc::new(AtomicU64::new(7200));
706
707        let config = make_server_config(port, 72, None);
708        let runtime = Arc::clone(&config.runtime);
709
710        start(config, tx, next_id).unwrap();
711        thread::sleep(Duration::from_millis(50));
712
713        let _client1 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
714        let event = rx.recv_timeout(Duration::from_secs(2)).unwrap();
715        assert!(matches!(event, Event::InterfaceUp(_, _, _)));
716
717        {
718            let mut runtime = runtime.lock().unwrap();
719            runtime.max_connections = Some(1);
720        }
721
722        let _client2 = TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
723        let result = rx.recv_timeout(Duration::from_millis(400));
724        assert!(
725            result.is_err(),
726            "expected no InterfaceUp after lowering max_connections, got {:?}",
727            result
728        );
729    }
730}
rns_net/interface/tcp_server.rs

rns_net/interface/
tcp_server.rs
