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

rns_net/interface/
tcp_server.rs
