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

rns_net/interface/
tcp_server.rs
