1use 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
36pub struct SharedClientConfig {
38 pub instance_name: String,
40 pub port: u16,
42 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 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 local_hops_delta: 0,
78 prefer_shorter_path: false,
79 max_paths_per_destination: 1,
80 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
81 packet_hashlist_allocation: rns_core::transport::types::PacketHashlistAllocation::Eager,
82 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
83 max_path_destinations: rns_core::transport::types::DEFAULT_MAX_PATH_DESTINATIONS,
84 max_tunnel_destinations_total: usize::MAX,
85 destination_timeout_secs: rns_core::constants::DESTINATION_TIMEOUT,
86 announce_table_ttl_secs: rns_core::constants::ANNOUNCE_TABLE_TTL,
87 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
88 announce_sig_cache_enabled: true,
89 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
90 announce_sig_cache_ttl_secs: rns_core::constants::ANNOUNCE_SIG_CACHE_TTL,
91 announce_queue_max_entries: 256,
92 announce_queue_max_interfaces: 1024,
93 };
94
95 let (tx, rx) = event::channel();
96 let tick_interval_ms = Arc::new(AtomicU64::new(1000));
97 let mut driver = Driver::new(transport_config, rx, tx.clone(), callbacks);
98 driver.set_tick_interval_handle(Arc::clone(&tick_interval_ms));
99
100 let local_config = LocalClientConfig {
102 name: "Local shared instance".into(),
103 instance_name: config.instance_name.clone(),
104 port: config.port,
105 interface_id: rns_core::transport::types::InterfaceId(1),
106 reconnect_wait,
107 };
108
109 let id = local_config.interface_id;
110 let info = rns_core::transport::types::InterfaceInfo {
111 id,
112 name: "LocalInterface".into(),
113 mode: rns_core::constants::MODE_FULL,
114 gravity: 0,
115 recursive_prs: false,
116 announces_from_internal: true,
117 announces_to_internal: None,
118 out_capable: true,
119 in_capable: true,
120 bitrate: Some(crate::interface::local::BITRATE),
121 airtime_profile: None,
122 announce_rate_target: None,
123 announce_rate_grace: 0,
124 announce_rate_penalty: 0.0,
125 announce_cap: rns_core::constants::ANNOUNCE_CAP,
126 is_local_client: true,
127 wants_tunnel: false,
128 tunnel_id: None,
129 mtu: crate::interface::local::HW_MTU,
130 ia_freq: 0.0,
131 ip_freq: 0.0,
132 op_freq: 0.0,
133 op_samples: 0,
134 started: time::now(),
135 ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
136 };
137
138 let writer = crate::interface::local::start_client(local_config, tx.clone())?;
139 let (writer, async_writer_metrics) = crate::interface::wrap_async_writer(
140 writer,
141 id,
142 &info.name,
143 tx.clone(),
144 driver.interface_writer_queue_capacity,
145 );
146
147 driver.engine.register_interface(info.clone());
148 driver.interfaces.insert(
149 id,
150 InterfaceEntry {
151 id,
152 info,
153 writer,
154 async_writer_metrics: Some(async_writer_metrics),
155 enabled: true,
156 online: false,
157 dynamic: false,
158 ifac: None,
159 stats: InterfaceStats {
160 started: time::now(),
161 ..Default::default()
162 },
163 interface_type: "LocalClientInterface".to_string(),
164 send_retry_at: None,
165 send_retry_backoff: Duration::ZERO,
166 },
167 );
168
169 let timer_tx = tx.clone();
171 let timer_interval = Arc::clone(&tick_interval_ms);
172 thread::Builder::new()
173 .name("rns-timer-client".into())
174 .spawn(move || loop {
175 let ms = timer_interval.load(Ordering::Relaxed);
176 thread::sleep(Duration::from_millis(ms));
177 if timer_tx.send(event::Event::Tick).is_err() {
178 break;
179 }
180 })?;
181
182 let driver_handle = thread::Builder::new()
184 .name("rns-driver-client".into())
185 .spawn(move || {
186 driver.run();
187 })?;
188
189 Ok(RnsNode::from_parts(
190 tx,
191 driver_handle,
192 None,
193 tick_interval_ms,
194 ))
195 }
196
197 pub fn connect_shared_from_config(
201 config_path: Option<&Path>,
202 callbacks: Box<dyn Callbacks>,
203 ) -> io::Result<Self> {
204 let config_dir = storage::resolve_config_dir(config_path);
205
206 let config_file = config_dir.join("config");
208 let rns_config = if config_file.exists() {
209 crate::config::parse_file(&config_file)
210 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?
211 } else {
212 crate::config::parse("")
213 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?
214 };
215
216 let shared_config = SharedClientConfig {
217 instance_name: rns_config.reticulum.instance_name.clone(),
218 port: rns_config.reticulum.shared_instance_port,
219 rpc_port: rns_config.reticulum.instance_control_port,
220 };
221
222 Self::connect_shared(shared_config, callbacks)
223 }
224}
225
226#[doc(hidden)]
227pub fn bench_shared_client_replay_once(
228 announce_count: usize,
229 reconnect_wait: Duration,
230) -> io::Result<usize> {
231 struct BenchNoopCallbacks;
232 impl Callbacks for BenchNoopCallbacks {
233 fn on_announce(&mut self, _: crate::destination::AnnouncedIdentity) {}
234 fn on_path_updated(&mut self, _: rns_core::types::DestHash, _: u8) {}
235 fn on_local_delivery(
236 &mut self,
237 _: rns_core::types::DestHash,
238 _: Vec<u8>,
239 _: rns_core::types::PacketHash,
240 ) {
241 }
242 }
243
244 fn build_shared_announce_raw(
245 dest_hash: &[u8; 16],
246 name_hash: &[u8; 10],
247 identity_prv_key: &[u8; 64],
248 app_data: Option<&[u8]>,
249 path_response: bool,
250 ) -> Vec<u8> {
251 let identity = Identity::from_private_key(identity_prv_key);
252 let mut random_hash = [0u8; 10];
253 random_hash[..5].copy_from_slice(&[0xA5; 5]);
254 random_hash[5..10].copy_from_slice(&[0, 0, 0, 0, 1]);
255
256 let (announce_data, _) = rns_core::announce::AnnounceData::pack(
257 &identity,
258 dest_hash,
259 name_hash,
260 &random_hash,
261 None,
262 app_data,
263 )
264 .unwrap();
265
266 let flags = rns_core::packet::PacketFlags {
267 header_type: rns_core::constants::HEADER_1,
268 context_flag: rns_core::constants::FLAG_UNSET,
269 transport_type: rns_core::constants::TRANSPORT_BROADCAST,
270 destination_type: rns_core::constants::DESTINATION_SINGLE,
271 packet_type: rns_core::constants::PACKET_TYPE_ANNOUNCE,
272 };
273 let context = if path_response {
274 rns_core::constants::CONTEXT_PATH_RESPONSE
275 } else {
276 rns_core::constants::CONTEXT_NONE
277 };
278
279 rns_core::packet::RawPacket::pack(flags, 0, dest_hash, None, context, &announce_data)
280 .unwrap()
281 .raw
282 }
283
284 fn read_until_frames(
285 stream: &mut std::net::TcpStream,
286 expected: usize,
287 expected_context: u8,
288 ) -> io::Result<Vec<Vec<u8>>> {
289 let mut decoder = hdlc::Decoder::new();
290 let mut buf = [0u8; 4096];
291 let mut frames = Vec::new();
292 let deadline = std::time::Instant::now() + Duration::from_secs(2);
293 while frames.len() < expected {
294 let n = match stream.read(&mut buf) {
295 Ok(n) => n,
296 Err(e)
297 if e.kind() == io::ErrorKind::WouldBlock
298 || e.kind() == io::ErrorKind::TimedOut =>
299 {
300 if std::time::Instant::now() >= deadline {
301 return Err(io::Error::new(
302 io::ErrorKind::TimedOut,
303 format!(
304 "timed out waiting for {} frames, got {}",
305 expected,
306 frames.len()
307 ),
308 ));
309 }
310 continue;
311 }
312 Err(e) => return Err(e),
313 };
314 for frame in decoder.feed(&buf[..n]) {
315 let Ok(packet) = RawPacket::unpack(&frame) else {
316 log::debug!("skipping malformed shared-instance packet");
317 continue;
318 };
319 if packet.context == expected_context {
320 frames.push(frame);
321 }
322 }
323 }
324 Ok(frames)
325 }
326
327 let listener1 = std::net::TcpListener::bind("127.0.0.1:0")?;
328 let port = listener1.local_addr()?.port();
329 let instance_name = format!("bench-shared-replay-{port}");
330
331 let (accepted1_tx, accepted1_rx) = std::sync::mpsc::channel();
332 thread::spawn(move || {
333 let (stream, _) = listener1.accept().unwrap();
334 accepted1_tx.send(stream).unwrap();
335 });
336
337 let transport_config = TransportConfig {
338 transport_enabled: false,
339 identity_hash: None,
340 local_hops_delta: 0,
341 prefer_shorter_path: false,
342 max_paths_per_destination: 1,
343 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
344 packet_hashlist_allocation: rns_core::transport::types::PacketHashlistAllocation::Eager,
345 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
346 max_path_destinations: rns_core::transport::types::DEFAULT_MAX_PATH_DESTINATIONS,
347 max_tunnel_destinations_total: usize::MAX,
348 destination_timeout_secs: rns_core::constants::DESTINATION_TIMEOUT,
349 announce_table_ttl_secs: rns_core::constants::ANNOUNCE_TABLE_TTL,
350 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
351 announce_sig_cache_enabled: true,
352 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
353 announce_sig_cache_ttl_secs: rns_core::constants::ANNOUNCE_SIG_CACHE_TTL,
354 announce_queue_max_entries: 256,
355 announce_queue_max_interfaces: 1024,
356 };
357
358 let (tx, rx) = event::channel();
359 let tick_interval_ms = Arc::new(AtomicU64::new(1000));
360 let mut driver = Driver::new(
361 transport_config,
362 rx,
363 tx.clone(),
364 Box::new(BenchNoopCallbacks),
365 );
366 driver.set_tick_interval_handle(Arc::clone(&tick_interval_ms));
367
368 let local_config = LocalClientConfig {
369 name: "Shared replay bench".into(),
370 instance_name: instance_name.clone(),
371 port,
372 interface_id: rns_core::transport::types::InterfaceId(1),
373 reconnect_wait,
374 };
375
376 let id = local_config.interface_id;
377 let info = rns_core::transport::types::InterfaceInfo {
378 id,
379 name: "LocalInterface".into(),
380 mode: rns_core::constants::MODE_FULL,
381 gravity: 0,
382 recursive_prs: false,
383 announces_from_internal: true,
384 announces_to_internal: None,
385 out_capable: true,
386 in_capable: true,
387 bitrate: Some(crate::interface::local::BITRATE),
388 airtime_profile: None,
389 announce_rate_target: None,
390 announce_rate_grace: 0,
391 announce_rate_penalty: 0.0,
392 announce_cap: rns_core::constants::ANNOUNCE_CAP,
393 is_local_client: true,
394 wants_tunnel: false,
395 tunnel_id: None,
396 mtu: crate::interface::local::HW_MTU,
397 ia_freq: 0.0,
398 ip_freq: 0.0,
399 op_freq: 0.0,
400 op_samples: 0,
401 started: time::now(),
402 ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
403 };
404
405 let writer = crate::interface::local::start_client(local_config, tx.clone())?;
406 let (writer, async_writer_metrics) = crate::interface::wrap_async_writer(
407 writer,
408 id,
409 &info.name,
410 tx.clone(),
411 driver.interface_writer_queue_capacity,
412 );
413 driver.engine.register_interface(info.clone());
414 driver.interfaces.insert(
415 id,
416 InterfaceEntry {
417 id,
418 info,
419 writer,
420 async_writer_metrics: Some(async_writer_metrics),
421 enabled: true,
422 online: false,
423 dynamic: false,
424 ifac: None,
425 stats: InterfaceStats {
426 started: time::now(),
427 ..Default::default()
428 },
429 interface_type: "LocalClientInterface".to_string(),
430 send_retry_at: None,
431 send_retry_backoff: Duration::ZERO,
432 },
433 );
434
435 let driver_handle = thread::Builder::new()
436 .name("rns-driver-bench-client".into())
437 .spawn(move || {
438 driver.run();
439 })?;
440
441 let mut stream1 = accepted1_rx
442 .recv_timeout(Duration::from_secs(2))
443 .map_err(|e| {
444 io::Error::new(
445 io::ErrorKind::TimedOut,
446 format!("shared bench initial accept failed: {e}"),
447 )
448 })?;
449 stream1.set_read_timeout(Some(Duration::from_secs(2)))?;
450
451 let mut records = Vec::new();
452 for i in 0..announce_count {
453 let mut prv_key = [0u8; 64];
454 for (j, byte) in prv_key.iter_mut().enumerate() {
455 *byte = (i as u8)
456 .wrapping_mul(23)
457 .wrapping_add(j as u8)
458 .wrapping_add(5);
459 }
460 let identity = Identity::from_private_key(&prv_key);
461 let aspect = format!("echo-{i}");
462 let name_hash = name_hash("shared-bench", &[&aspect]);
463 let dest_hash = destination_hash("shared-bench", &[&aspect], Some(identity.hash()));
464 let app_data = format!("hello-{i}").into_bytes();
465 records.push((dest_hash, name_hash, prv_key, app_data));
466 }
467
468 for (dest_hash, name_hash, prv_key, app_data) in &records {
469 let raw = build_shared_announce_raw(dest_hash, name_hash, prv_key, Some(app_data), false);
470 tx.send(Event::RegisterDestination {
471 dest_hash: *dest_hash,
472 dest_type: rns_core::constants::DESTINATION_SINGLE,
473 })
474 .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, format!("{e}")))?;
475 tx.send(Event::StoreSharedAnnounce {
476 dest_hash: *dest_hash,
477 name_hash: *name_hash,
478 identity_prv_key: *prv_key,
479 app_data: Some(app_data.clone()),
480 })
481 .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, format!("{e}")))?;
482 tx.send(Event::SendOutbound {
483 raw,
484 dest_type: rns_core::constants::DESTINATION_SINGLE,
485 attached_interface: None,
486 })
487 .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, format!("{e}")))?;
488 }
489
490 let _ = read_until_frames(
491 &mut stream1,
492 announce_count,
493 rns_core::constants::CONTEXT_NONE,
494 )
495 .map_err(|e| io::Error::new(e.kind(), format!("initial announce delivery failed: {e}")))?;
496 drop(stream1);
497
498 let listener2 = std::net::TcpListener::bind(format!("127.0.0.1:{port}"))?;
499 let (accepted2_tx, accepted2_rx) = std::sync::mpsc::channel();
500 thread::spawn(move || {
501 let (stream, _) = listener2.accept().unwrap();
502 accepted2_tx.send(stream).unwrap();
503 });
504
505 let mut stream2 = accepted2_rx
506 .recv_timeout(Duration::from_secs(2))
507 .map_err(|e| {
508 io::Error::new(
509 io::ErrorKind::TimedOut,
510 format!("shared bench reconnect accept failed: {e}"),
511 )
512 })?;
513 stream2.set_read_timeout(Some(Duration::from_secs(2)))?;
514 let frames = read_until_frames(
515 &mut stream2,
516 announce_count,
517 rns_core::constants::CONTEXT_PATH_RESPONSE,
518 )
519 .map_err(|e| io::Error::new(e.kind(), format!("replayed announce delivery failed: {e}")))?;
520
521 tx.send(Event::Shutdown)
522 .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, format!("{e}")))?;
523 let _ = driver_handle.join();
524
525 Ok(frames.len())
526}
527
528#[cfg(test)]
529mod tests {
530 use super::*;
531 use crate::hdlc;
532 use rns_core::packet::RawPacket;
533 use rns_core::types::IdentityHash;
534 use rns_crypto::identity::Identity;
535 use rns_crypto::OsRng;
536 use std::io::Read;
537 use std::sync::atomic::AtomicU64;
538 use std::sync::mpsc;
539 use std::sync::Arc;
540
541 use crate::interface::local::LocalServerConfig;
542
543 struct NoopCallbacks;
544 impl Callbacks for NoopCallbacks {
545 fn on_announce(&mut self, _: crate::destination::AnnouncedIdentity) {}
546 fn on_path_updated(&mut self, _: rns_core::types::DestHash, _: u8) {}
547 fn on_local_delivery(
548 &mut self,
549 _: rns_core::types::DestHash,
550 _: Vec<u8>,
551 _: rns_core::types::PacketHash,
552 ) {
553 }
554 }
555
556 fn find_free_port() -> u16 {
557 crate::test_support::port()
558 }
559
560 #[test]
561 fn connect_shared_to_tcp_server() {
562 let port = find_free_port();
563 let next_id = Arc::new(AtomicU64::new(50000));
564 let (server_tx, server_rx) = crate::event::channel();
565
566 let server_config = LocalServerConfig {
568 instance_name: "test-shared-connect".into(),
569 port,
570 interface_id: rns_core::transport::types::InterfaceId(99),
571 };
572
573 crate::interface::local::start_server(server_config, server_tx, next_id).unwrap();
574 thread::sleep(Duration::from_millis(50));
575
576 let config = SharedClientConfig {
578 instance_name: "test-shared-connect".into(),
579 port,
580 rpc_port: 0,
581 };
582
583 let node = RnsNode::connect_shared(config, Box::new(NoopCallbacks)).unwrap();
584
585 let event = server_rx.recv_timeout(Duration::from_secs(2)).unwrap();
587 assert!(matches!(event, crate::event::Event::InterfaceUp(_, _, _)));
588
589 node.shutdown();
590 }
591
592 #[test]
593 fn shared_client_register_destination() {
594 let port = find_free_port();
595 let next_id = Arc::new(AtomicU64::new(51000));
596 let (server_tx, _server_rx) = crate::event::channel();
597
598 let server_config = LocalServerConfig {
599 instance_name: "test-shared-reg".into(),
600 port,
601 interface_id: rns_core::transport::types::InterfaceId(98),
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-reg".into(),
609 port,
610 rpc_port: 0,
611 };
612
613 let node = RnsNode::connect_shared(config, Box::new(NoopCallbacks)).unwrap();
614
615 let dest_hash = [0xAA; 16];
617 node.register_destination(dest_hash, rns_core::constants::DESTINATION_SINGLE)
618 .unwrap();
619
620 thread::sleep(Duration::from_millis(100));
622
623 node.shutdown();
624 }
625
626 #[test]
627 fn shared_client_send_packet() {
628 let port = find_free_port();
629 let next_id = Arc::new(AtomicU64::new(52000));
630 let (server_tx, server_rx) = crate::event::channel();
631
632 let server_config = LocalServerConfig {
633 instance_name: "test-shared-send".into(),
634 port,
635 interface_id: rns_core::transport::types::InterfaceId(97),
636 };
637
638 crate::interface::local::start_server(server_config, server_tx, next_id).unwrap();
639 thread::sleep(Duration::from_millis(50));
640
641 let config = SharedClientConfig {
642 instance_name: "test-shared-send".into(),
643 port,
644 rpc_port: 0,
645 };
646
647 let node = RnsNode::connect_shared(config, Box::new(NoopCallbacks)).unwrap();
648
649 let raw = vec![0x00, 0x00, 0xAA, 0xBB, 0xCC, 0xDD]; node.send_raw(raw, rns_core::constants::DESTINATION_PLAIN, None)
652 .unwrap();
653
654 for _ in 0..10 {
657 match server_rx.recv_timeout(Duration::from_secs(1)) {
658 Ok(crate::event::Event::Frame { .. }) => {
659 break;
660 }
661 Ok(_) => continue,
662 Err(_) => break,
663 }
664 }
665 node.shutdown();
669 }
670
671 #[test]
672 fn shared_client_replays_single_announces_after_reconnect() {
673 let port = find_free_port();
674 let addr = format!("127.0.0.1:{}", port);
675 let instance_name = format!("test-shared-replay-{}", port);
676
677 let listener1 = std::net::TcpListener::bind(&addr).unwrap();
678 let (accepted1_tx, accepted1_rx) = mpsc::channel();
679 thread::spawn(move || {
680 let (stream, _) = listener1.accept().unwrap();
681 accepted1_tx.send(stream).unwrap();
682 });
683
684 let node = RnsNode::connect_shared(
685 SharedClientConfig {
686 instance_name,
687 port,
688 rpc_port: 0,
689 },
690 Box::new(NoopCallbacks),
691 )
692 .unwrap();
693
694 let identity = Identity::new(&mut OsRng);
695 let dest = crate::destination::Destination::single_in(
696 "shared-replay",
697 &["echo"],
698 IdentityHash(*identity.hash()),
699 );
700 node.register_destination(dest.hash.0, dest.dest_type.to_wire_constant())
701 .unwrap();
702 node.announce_queued(&dest, &identity, Some(b"hello"))
703 .unwrap();
704
705 let mut stream1 = accepted1_rx.recv_timeout(Duration::from_secs(2)).unwrap();
706 stream1
707 .set_read_timeout(Some(Duration::from_secs(2)))
708 .unwrap();
709
710 let mut decoder = hdlc::Decoder::new();
711 let mut buf = [0u8; 4096];
712 let n = stream1.read(&mut buf).unwrap();
713 let frames = decoder.feed(&buf[..n]);
714 assert!(!frames.is_empty(), "expected initial announce frame");
715 let packet1 = RawPacket::unpack(&frames[0]).unwrap();
716 assert_eq!(packet1.destination_hash, dest.hash.0);
717 assert_eq!(packet1.context, rns_core::constants::CONTEXT_NONE);
718
719 drop(stream1);
720
721 let listener2 = std::net::TcpListener::bind(&addr).unwrap();
722 let (accepted2_tx, accepted2_rx) = mpsc::channel();
723 thread::spawn(move || {
724 let (stream, _) = listener2.accept().unwrap();
725 accepted2_tx.send(stream).unwrap();
726 });
727
728 let mut stream2 = accepted2_rx.recv_timeout(Duration::from_secs(15)).unwrap();
729 stream2
730 .set_read_timeout(Some(Duration::from_secs(15)))
731 .unwrap();
732
733 let mut decoder = hdlc::Decoder::new();
734 let n = stream2.read(&mut buf).unwrap();
735 let frames = decoder.feed(&buf[..n]);
736 assert!(!frames.is_empty(), "expected replayed announce frame");
737 let packet2 = RawPacket::unpack(&frames[0]).unwrap();
738 assert_eq!(packet2.destination_hash, dest.hash.0);
739 assert_eq!(packet2.context, rns_core::constants::CONTEXT_PATH_RESPONSE);
740
741 node.shutdown();
742 }
743
744 #[test]
745 fn connect_shared_fails_no_server() {
746 let port = find_free_port();
747
748 let config = SharedClientConfig {
749 instance_name: "nonexistent-instance-12345".into(),
750 port,
751 rpc_port: 0,
752 };
753
754 let result = RnsNode::connect_shared(config, Box::new(NoopCallbacks));
756 assert!(result.is_err());
757 }
758
759 #[test]
760 fn shared_config_defaults() {
761 let config = SharedClientConfig::default();
762 assert_eq!(config.instance_name, "default");
763 assert_eq!(config.port, 37428);
764 assert_eq!(config.rpc_port, 37429);
765 }
766}