rns_net/
shared_client.rs

1//! Shared instance client mode.
2//!
3//! Allows an RnsNode to connect as a client to an already-running Reticulum
4//! daemon, proxying operations through it. The client runs a minimal transport
5//! engine with `transport_enabled: false` — it does no routing of its own, but
6//! registers local destinations and sends/receives packets via the local
7//! connection.
8//!
9//! This matches Python's behavior when `share_instance = True` and a daemon
10//! is already running: the new process connects as a client rather than
11//! starting its own interfaces.
12
13use std::io;
14use std::io::Read;
15use std::path::Path;
16use std::sync::atomic::{AtomicU64, Ordering};
17use std::sync::Arc;
18use std::thread;
19use std::time::Duration;
20
21use rns_core::destination::{destination_hash, name_hash};
22use rns_core::packet::RawPacket;
23use rns_core::transport::types::TransportConfig;
24use rns_crypto::identity::Identity;
25
26use crate::driver::{Callbacks, Driver};
27use crate::event;
28use crate::event::Event;
29use crate::hdlc;
30use crate::interface::local::LocalClientConfig;
31use crate::interface::{InterfaceEntry, InterfaceStats};
32use crate::node::RnsNode;
33use crate::storage;
34use crate::time;
35
36/// Configuration for connecting as a shared instance client.
37pub struct SharedClientConfig {
38    /// Instance name for Unix socket namespace (e.g. "default" → `\0rns/default`).
39    pub instance_name: String,
40    /// TCP port to try if Unix socket fails (default 37428).
41    pub port: u16,
42    /// RPC control port for queries (default 37429).
43    pub rpc_port: u16,
44}
45
46impl Default for SharedClientConfig {
47    fn default() -> Self {
48        SharedClientConfig {
49            instance_name: "default".into(),
50            port: 37428,
51            rpc_port: 37429,
52        }
53    }
54}
55
56impl RnsNode {
57    /// Connect to an existing shared instance as a client.
58    ///
59    /// The client runs `transport_enabled: false` — it does no routing,
60    /// but can register destinations and send/receive packets through
61    /// the daemon.
62    pub fn connect_shared(
63        config: SharedClientConfig,
64        callbacks: Box<dyn Callbacks>,
65    ) -> io::Result<Self> {
66        Self::connect_shared_with_reconnect_wait(config, callbacks, Duration::from_secs(8))
67    }
68
69    fn connect_shared_with_reconnect_wait(
70        config: SharedClientConfig,
71        callbacks: Box<dyn Callbacks>,
72        reconnect_wait: Duration,
73    ) -> io::Result<Self> {
74        let transport_config = TransportConfig {
75            transport_enabled: false,
76            identity_hash: None,
77            prefer_shorter_path: false,
78            max_paths_per_destination: 1,
79            packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
80            max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
81            max_path_destinations: rns_core::transport::types::DEFAULT_MAX_PATH_DESTINATIONS,
82            max_tunnel_destinations_total: usize::MAX,
83            destination_timeout_secs: rns_core::constants::DESTINATION_TIMEOUT,
84            announce_table_ttl_secs: rns_core::constants::ANNOUNCE_TABLE_TTL,
85            announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
86            announce_sig_cache_enabled: true,
87            announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
88            announce_sig_cache_ttl_secs: rns_core::constants::ANNOUNCE_SIG_CACHE_TTL,
89            announce_queue_max_entries: 256,
90            announce_queue_max_interfaces: 1024,
91        };
92
93        let (tx, rx) = event::channel();
94        let tick_interval_ms = Arc::new(AtomicU64::new(1000));
95        let mut driver = Driver::new(transport_config, rx, tx.clone(), callbacks);
96        driver.set_tick_interval_handle(Arc::clone(&tick_interval_ms));
97
98        // Connect to the daemon via LocalClientInterface
99        let local_config = LocalClientConfig {
100            name: "Local shared instance".into(),
101            instance_name: config.instance_name.clone(),
102            port: config.port,
103            interface_id: rns_core::transport::types::InterfaceId(1),
104            reconnect_wait,
105        };
106
107        let id = local_config.interface_id;
108        let info = rns_core::transport::types::InterfaceInfo {
109            id,
110            name: "LocalInterface".into(),
111            mode: rns_core::constants::MODE_FULL,
112            out_capable: true,
113            in_capable: true,
114            bitrate: Some(1_000_000_000),
115            airtime_profile: None,
116            announce_rate_target: None,
117            announce_rate_grace: 0,
118            announce_rate_penalty: 0.0,
119            announce_cap: rns_core::constants::ANNOUNCE_CAP,
120            is_local_client: true,
121            wants_tunnel: false,
122            tunnel_id: None,
123            mtu: 65535,
124            ia_freq: 0.0,
125            started: time::now(),
126            ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
127        };
128
129        let writer = crate::interface::local::start_client(local_config, tx.clone())?;
130
131        driver.engine.register_interface(info.clone());
132        driver.interfaces.insert(
133            id,
134            InterfaceEntry {
135                id,
136                info,
137                writer,
138                async_writer_metrics: None,
139                enabled: true,
140                online: false,
141                dynamic: false,
142                ifac: None,
143                stats: InterfaceStats {
144                    started: time::now(),
145                    ..Default::default()
146                },
147                interface_type: "LocalClientInterface".to_string(),
148                send_retry_at: None,
149                send_retry_backoff: Duration::ZERO,
150            },
151        );
152
153        // Spawn timer thread with configurable tick interval
154        let timer_tx = tx.clone();
155        let timer_interval = Arc::clone(&tick_interval_ms);
156        thread::Builder::new()
157            .name("rns-timer-client".into())
158            .spawn(move || loop {
159                let ms = timer_interval.load(Ordering::Relaxed);
160                thread::sleep(Duration::from_millis(ms));
161                if timer_tx.send(event::Event::Tick).is_err() {
162                    break;
163                }
164            })?;
165
166        // Spawn driver thread
167        let driver_handle = thread::Builder::new()
168            .name("rns-driver-client".into())
169            .spawn(move || {
170                driver.run();
171            })?;
172
173        Ok(RnsNode::from_parts(
174            tx,
175            driver_handle,
176            None,
177            tick_interval_ms,
178        ))
179    }
180
181    /// Connect to a shared instance, with config loaded from a config directory.
182    ///
183    /// Reads the config file to determine instance_name and ports.
184    pub fn connect_shared_from_config(
185        config_path: Option<&Path>,
186        callbacks: Box<dyn Callbacks>,
187    ) -> io::Result<Self> {
188        let config_dir = storage::resolve_config_dir(config_path);
189
190        // Parse config file for instance settings
191        let config_file = config_dir.join("config");
192        let rns_config = if config_file.exists() {
193            crate::config::parse_file(&config_file)
194                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?
195        } else {
196            crate::config::parse("")
197                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?
198        };
199
200        let shared_config = SharedClientConfig {
201            instance_name: rns_config.reticulum.instance_name.clone(),
202            port: rns_config.reticulum.shared_instance_port,
203            rpc_port: rns_config.reticulum.instance_control_port,
204        };
205
206        Self::connect_shared(shared_config, callbacks)
207    }
208}
209
210#[doc(hidden)]
211pub fn bench_shared_client_replay_once(
212    announce_count: usize,
213    reconnect_wait: Duration,
214) -> io::Result<usize> {
215    struct BenchNoopCallbacks;
216    impl Callbacks for BenchNoopCallbacks {
217        fn on_announce(&mut self, _: crate::destination::AnnouncedIdentity) {}
218        fn on_path_updated(&mut self, _: rns_core::types::DestHash, _: u8) {}
219        fn on_local_delivery(
220            &mut self,
221            _: rns_core::types::DestHash,
222            _: Vec<u8>,
223            _: rns_core::types::PacketHash,
224        ) {
225        }
226    }
227
228    fn build_shared_announce_raw(
229        dest_hash: &[u8; 16],
230        name_hash: &[u8; 10],
231        identity_prv_key: &[u8; 64],
232        app_data: Option<&[u8]>,
233        path_response: bool,
234    ) -> Vec<u8> {
235        let identity = Identity::from_private_key(identity_prv_key);
236        let mut random_hash = [0u8; 10];
237        random_hash[..5].copy_from_slice(&[0xA5; 5]);
238        random_hash[5..10].copy_from_slice(&[0, 0, 0, 0, 1]);
239
240        let (announce_data, _) = rns_core::announce::AnnounceData::pack(
241            &identity,
242            dest_hash,
243            name_hash,
244            &random_hash,
245            None,
246            app_data,
247        )
248        .unwrap();
249
250        let flags = rns_core::packet::PacketFlags {
251            header_type: rns_core::constants::HEADER_1,
252            context_flag: rns_core::constants::FLAG_UNSET,
253            transport_type: rns_core::constants::TRANSPORT_BROADCAST,
254            destination_type: rns_core::constants::DESTINATION_SINGLE,
255            packet_type: rns_core::constants::PACKET_TYPE_ANNOUNCE,
256        };
257        let context = if path_response {
258            rns_core::constants::CONTEXT_PATH_RESPONSE
259        } else {
260            rns_core::constants::CONTEXT_NONE
261        };
262
263        rns_core::packet::RawPacket::pack(flags, 0, dest_hash, None, context, &announce_data)
264            .unwrap()
265            .raw
266    }
267
268    fn read_until_frames(
269        stream: &mut std::net::TcpStream,
270        expected: usize,
271        expected_context: u8,
272    ) -> io::Result<Vec<Vec<u8>>> {
273        let mut decoder = hdlc::Decoder::new();
274        let mut buf = [0u8; 4096];
275        let mut frames = Vec::new();
276        let deadline = std::time::Instant::now() + Duration::from_secs(2);
277        while frames.len() < expected {
278            let n = match stream.read(&mut buf) {
279                Ok(n) => n,
280                Err(e)
281                    if e.kind() == io::ErrorKind::WouldBlock
282                        || e.kind() == io::ErrorKind::TimedOut =>
283                {
284                    if std::time::Instant::now() >= deadline {
285                        return Err(io::Error::new(
286                            io::ErrorKind::TimedOut,
287                            format!(
288                                "timed out waiting for {} frames, got {}",
289                                expected,
290                                frames.len()
291                            ),
292                        ));
293                    }
294                    continue;
295                }
296                Err(e) => return Err(e),
297            };
298            for frame in decoder.feed(&buf[..n]) {
299                let packet = RawPacket::unpack(&frame)
300                    .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{e}")))?;
301                if packet.context == expected_context {
302                    frames.push(frame);
303                }
304            }
305        }
306        Ok(frames)
307    }
308
309    let port = {
310        let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
311        listener.local_addr()?.port()
312    };
313    let instance_name = format!("bench-shared-replay-{port}");
314
315    let listener1 = std::net::TcpListener::bind(format!("127.0.0.1:{port}"))?;
316    let (accepted1_tx, accepted1_rx) = std::sync::mpsc::channel();
317    thread::spawn(move || {
318        let (stream, _) = listener1.accept().unwrap();
319        accepted1_tx.send(stream).unwrap();
320    });
321
322    let transport_config = TransportConfig {
323        transport_enabled: false,
324        identity_hash: None,
325        prefer_shorter_path: false,
326        max_paths_per_destination: 1,
327        packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
328        max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
329        max_path_destinations: rns_core::transport::types::DEFAULT_MAX_PATH_DESTINATIONS,
330        max_tunnel_destinations_total: usize::MAX,
331        destination_timeout_secs: rns_core::constants::DESTINATION_TIMEOUT,
332        announce_table_ttl_secs: rns_core::constants::ANNOUNCE_TABLE_TTL,
333        announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
334        announce_sig_cache_enabled: true,
335        announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
336        announce_sig_cache_ttl_secs: rns_core::constants::ANNOUNCE_SIG_CACHE_TTL,
337        announce_queue_max_entries: 256,
338        announce_queue_max_interfaces: 1024,
339    };
340
341    let (tx, rx) = event::channel();
342    let tick_interval_ms = Arc::new(AtomicU64::new(1000));
343    let mut driver = Driver::new(
344        transport_config,
345        rx,
346        tx.clone(),
347        Box::new(BenchNoopCallbacks),
348    );
349    driver.set_tick_interval_handle(Arc::clone(&tick_interval_ms));
350
351    let local_config = LocalClientConfig {
352        name: "Shared replay bench".into(),
353        instance_name: instance_name.clone(),
354        port,
355        interface_id: rns_core::transport::types::InterfaceId(1),
356        reconnect_wait,
357    };
358
359    let id = local_config.interface_id;
360    let info = rns_core::transport::types::InterfaceInfo {
361        id,
362        name: "LocalInterface".into(),
363        mode: rns_core::constants::MODE_FULL,
364        out_capable: true,
365        in_capable: true,
366        bitrate: Some(1_000_000_000),
367        airtime_profile: None,
368        announce_rate_target: None,
369        announce_rate_grace: 0,
370        announce_rate_penalty: 0.0,
371        announce_cap: rns_core::constants::ANNOUNCE_CAP,
372        is_local_client: true,
373        wants_tunnel: false,
374        tunnel_id: None,
375        mtu: 65535,
376        ia_freq: 0.0,
377        started: time::now(),
378        ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
379    };
380
381    let writer = crate::interface::local::start_client(local_config, tx.clone())?;
382    driver.engine.register_interface(info.clone());
383    driver.interfaces.insert(
384        id,
385        InterfaceEntry {
386            id,
387            info,
388            writer,
389            async_writer_metrics: None,
390            enabled: true,
391            online: false,
392            dynamic: false,
393            ifac: None,
394            stats: InterfaceStats {
395                started: time::now(),
396                ..Default::default()
397            },
398            interface_type: "LocalClientInterface".to_string(),
399            send_retry_at: None,
400            send_retry_backoff: Duration::ZERO,
401        },
402    );
403
404    let driver_handle = thread::Builder::new()
405        .name("rns-driver-bench-client".into())
406        .spawn(move || {
407            driver.run();
408        })?;
409
410    let mut stream1 = accepted1_rx
411        .recv_timeout(Duration::from_secs(2))
412        .map_err(|e| {
413            io::Error::new(
414                io::ErrorKind::TimedOut,
415                format!("shared bench initial accept failed: {e}"),
416            )
417        })?;
418    stream1.set_read_timeout(Some(Duration::from_secs(2)))?;
419
420    let mut records = Vec::new();
421    for i in 0..announce_count {
422        let mut prv_key = [0u8; 64];
423        for (j, byte) in prv_key.iter_mut().enumerate() {
424            *byte = (i as u8)
425                .wrapping_mul(23)
426                .wrapping_add(j as u8)
427                .wrapping_add(5);
428        }
429        let identity = Identity::from_private_key(&prv_key);
430        let aspect = format!("echo-{i}");
431        let name_hash = name_hash("shared-bench", &[&aspect]);
432        let dest_hash = destination_hash("shared-bench", &[&aspect], Some(identity.hash()));
433        let app_data = format!("hello-{i}").into_bytes();
434        records.push((dest_hash, name_hash, prv_key, app_data));
435    }
436
437    for (dest_hash, name_hash, prv_key, app_data) in &records {
438        let raw = build_shared_announce_raw(dest_hash, name_hash, prv_key, Some(app_data), false);
439        tx.send(Event::StoreSharedAnnounce {
440            dest_hash: *dest_hash,
441            name_hash: *name_hash,
442            identity_prv_key: *prv_key,
443            app_data: Some(app_data.clone()),
444        })
445        .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, format!("{e}")))?;
446        tx.send(Event::SendOutbound {
447            raw,
448            dest_type: rns_core::constants::DESTINATION_SINGLE,
449            attached_interface: None,
450        })
451        .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, format!("{e}")))?;
452    }
453
454    let _ = read_until_frames(
455        &mut stream1,
456        announce_count,
457        rns_core::constants::CONTEXT_NONE,
458    )?;
459    drop(stream1);
460
461    let listener2 = std::net::TcpListener::bind(format!("127.0.0.1:{port}"))?;
462    let (accepted2_tx, accepted2_rx) = std::sync::mpsc::channel();
463    thread::spawn(move || {
464        let (stream, _) = listener2.accept().unwrap();
465        accepted2_tx.send(stream).unwrap();
466    });
467
468    let mut stream2 = accepted2_rx
469        .recv_timeout(Duration::from_secs(2))
470        .map_err(|e| {
471            io::Error::new(
472                io::ErrorKind::TimedOut,
473                format!("shared bench reconnect accept failed: {e}"),
474            )
475        })?;
476    stream2.set_read_timeout(Some(Duration::from_secs(2)))?;
477    let frames = read_until_frames(
478        &mut stream2,
479        announce_count,
480        rns_core::constants::CONTEXT_PATH_RESPONSE,
481    )?;
482
483    tx.send(Event::Shutdown)
484        .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, format!("{e}")))?;
485    let _ = driver_handle.join();
486
487    Ok(frames.len())
488}
489
490#[cfg(test)]
491mod tests {
492    use super::*;
493    use crate::hdlc;
494    use rns_core::packet::RawPacket;
495    use rns_core::types::IdentityHash;
496    use rns_crypto::identity::Identity;
497    use rns_crypto::OsRng;
498    use std::io::Read;
499    use std::sync::atomic::AtomicU64;
500    use std::sync::mpsc;
501    use std::sync::Arc;
502
503    use crate::interface::local::LocalServerConfig;
504
505    struct NoopCallbacks;
506    impl Callbacks for NoopCallbacks {
507        fn on_announce(&mut self, _: crate::destination::AnnouncedIdentity) {}
508        fn on_path_updated(&mut self, _: rns_core::types::DestHash, _: u8) {}
509        fn on_local_delivery(
510            &mut self,
511            _: rns_core::types::DestHash,
512            _: Vec<u8>,
513            _: rns_core::types::PacketHash,
514        ) {
515        }
516    }
517
518    fn find_free_port() -> u16 {
519        std::net::TcpListener::bind("127.0.0.1:0")
520            .unwrap()
521            .local_addr()
522            .unwrap()
523            .port()
524    }
525
526    #[test]
527    fn connect_shared_to_tcp_server() {
528        let port = find_free_port();
529        let next_id = Arc::new(AtomicU64::new(50000));
530        let (server_tx, server_rx) = crate::event::channel();
531
532        // Start a local server
533        let server_config = LocalServerConfig {
534            instance_name: "test-shared-connect".into(),
535            port,
536            interface_id: rns_core::transport::types::InterfaceId(99),
537        };
538
539        crate::interface::local::start_server(server_config, server_tx, next_id).unwrap();
540        thread::sleep(Duration::from_millis(50));
541
542        // Connect as shared client
543        let config = SharedClientConfig {
544            instance_name: "test-shared-connect".into(),
545            port,
546            rpc_port: 0,
547        };
548
549        let node = RnsNode::connect_shared(config, Box::new(NoopCallbacks)).unwrap();
550
551        // Server should see InterfaceUp for the client
552        let event = server_rx.recv_timeout(Duration::from_secs(2)).unwrap();
553        assert!(matches!(event, crate::event::Event::InterfaceUp(_, _, _)));
554
555        node.shutdown();
556    }
557
558    #[test]
559    fn shared_client_register_destination() {
560        let port = find_free_port();
561        let next_id = Arc::new(AtomicU64::new(51000));
562        let (server_tx, _server_rx) = crate::event::channel();
563
564        let server_config = LocalServerConfig {
565            instance_name: "test-shared-reg".into(),
566            port,
567            interface_id: rns_core::transport::types::InterfaceId(98),
568        };
569
570        crate::interface::local::start_server(server_config, server_tx, next_id).unwrap();
571        thread::sleep(Duration::from_millis(50));
572
573        let config = SharedClientConfig {
574            instance_name: "test-shared-reg".into(),
575            port,
576            rpc_port: 0,
577        };
578
579        let node = RnsNode::connect_shared(config, Box::new(NoopCallbacks)).unwrap();
580
581        // Register a destination
582        let dest_hash = [0xAA; 16];
583        node.register_destination(dest_hash, rns_core::constants::DESTINATION_SINGLE)
584            .unwrap();
585
586        // Give time for event processing
587        thread::sleep(Duration::from_millis(100));
588
589        node.shutdown();
590    }
591
592    #[test]
593    fn shared_client_send_packet() {
594        let port = find_free_port();
595        let next_id = Arc::new(AtomicU64::new(52000));
596        let (server_tx, server_rx) = crate::event::channel();
597
598        let server_config = LocalServerConfig {
599            instance_name: "test-shared-send".into(),
600            port,
601            interface_id: rns_core::transport::types::InterfaceId(97),
602        };
603
604        crate::interface::local::start_server(server_config, server_tx, next_id).unwrap();
605        thread::sleep(Duration::from_millis(50));
606
607        let config = SharedClientConfig {
608            instance_name: "test-shared-send".into(),
609            port,
610            rpc_port: 0,
611        };
612
613        let node = RnsNode::connect_shared(config, Box::new(NoopCallbacks)).unwrap();
614
615        // Build a minimal packet and send it
616        let raw = vec![0x00, 0x00, 0xAA, 0xBB, 0xCC, 0xDD]; // minimal raw packet
617        node.send_raw(raw, rns_core::constants::DESTINATION_PLAIN, None)
618            .unwrap();
619
620        // Server should receive a Frame event from the client
621        // (the packet will be HDLC-framed over the local connection)
622        for _ in 0..10 {
623            match server_rx.recv_timeout(Duration::from_secs(1)) {
624                Ok(crate::event::Event::Frame { .. }) => {
625                    break;
626                }
627                Ok(_) => continue,
628                Err(_) => break,
629            }
630        }
631        // The packet may or may not arrive as a Frame depending on transport
632        // routing, so we don't assert on it — the important thing is no crash.
633
634        node.shutdown();
635    }
636
637    #[test]
638    fn shared_client_replays_single_announces_after_reconnect() {
639        let port = find_free_port();
640        let addr = format!("127.0.0.1:{}", port);
641        let instance_name = format!("test-shared-replay-{}", port);
642
643        let listener1 = std::net::TcpListener::bind(&addr).unwrap();
644        let (accepted1_tx, accepted1_rx) = mpsc::channel();
645        thread::spawn(move || {
646            let (stream, _) = listener1.accept().unwrap();
647            accepted1_tx.send(stream).unwrap();
648        });
649
650        let node = RnsNode::connect_shared(
651            SharedClientConfig {
652                instance_name,
653                port,
654                rpc_port: 0,
655            },
656            Box::new(NoopCallbacks),
657        )
658        .unwrap();
659
660        let identity = Identity::new(&mut OsRng);
661        let dest = crate::destination::Destination::single_in(
662            "shared-replay",
663            &["echo"],
664            IdentityHash(*identity.hash()),
665        );
666        node.register_destination(dest.hash.0, dest.dest_type.to_wire_constant())
667            .unwrap();
668        node.announce(&dest, &identity, Some(b"hello")).unwrap();
669
670        let mut stream1 = accepted1_rx.recv_timeout(Duration::from_secs(2)).unwrap();
671        stream1
672            .set_read_timeout(Some(Duration::from_secs(2)))
673            .unwrap();
674
675        let mut decoder = hdlc::Decoder::new();
676        let mut buf = [0u8; 4096];
677        let n = stream1.read(&mut buf).unwrap();
678        let frames = decoder.feed(&buf[..n]);
679        assert!(!frames.is_empty(), "expected initial announce frame");
680        let packet1 = RawPacket::unpack(&frames[0]).unwrap();
681        assert_eq!(packet1.destination_hash, dest.hash.0);
682        assert_eq!(packet1.context, rns_core::constants::CONTEXT_NONE);
683
684        drop(stream1);
685
686        let listener2 = std::net::TcpListener::bind(&addr).unwrap();
687        let (accepted2_tx, accepted2_rx) = mpsc::channel();
688        thread::spawn(move || {
689            let (stream, _) = listener2.accept().unwrap();
690            accepted2_tx.send(stream).unwrap();
691        });
692
693        let mut stream2 = accepted2_rx.recv_timeout(Duration::from_secs(15)).unwrap();
694        stream2
695            .set_read_timeout(Some(Duration::from_secs(15)))
696            .unwrap();
697
698        let mut decoder = hdlc::Decoder::new();
699        let n = stream2.read(&mut buf).unwrap();
700        let frames = decoder.feed(&buf[..n]);
701        assert!(!frames.is_empty(), "expected replayed announce frame");
702        let packet2 = RawPacket::unpack(&frames[0]).unwrap();
703        assert_eq!(packet2.destination_hash, dest.hash.0);
704        assert_eq!(packet2.context, rns_core::constants::CONTEXT_PATH_RESPONSE);
705
706        node.shutdown();
707    }
708
709    #[test]
710    fn connect_shared_fails_no_server() {
711        let port = find_free_port();
712
713        let config = SharedClientConfig {
714            instance_name: "nonexistent-instance-12345".into(),
715            port,
716            rpc_port: 0,
717        };
718
719        // Should fail because no server is running
720        let result = RnsNode::connect_shared(config, Box::new(NoopCallbacks));
721        assert!(result.is_err());
722    }
723
724    #[test]
725    fn shared_config_defaults() {
726        let config = SharedClientConfig::default();
727        assert_eq!(config.instance_name, "default");
728        assert_eq!(config.port, 37428);
729        assert_eq!(config.rpc_port, 37429);
730    }
731}
rns_net/shared_client.rs

rns_net/
shared_client.rs
