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

rns_net/interface/
tcp_server.rs
