1//! Live presence and Live Share: who else has the notebook open, the page they are on and
2//! their caret, straight from one Snowbound to another, or through a relay (`crates/relay`)
3//! where they aren't on one network; and a notebook one machine holds opened on another
4//! (`share`). Peers find each other with mDNS (`_snowbound._tcp`) or in the relay's room, and
5//! meet through a secret both hold, a room's or a code typed on both, which SPAKE2 turns into
6//! the keys every frame after the opening is sealed with. Each connection has a thread reading
7//! and one writing. A connection whose frames arrive out of order is dropped and met again
8//! from scratch.
9
10pub use ::relay::code;
11mod group;
12pub mod proxy;
13mod relay;
14pub mod share;
15mod transport;
16pub use transport::Trouble;
17pub mod wire;
18pub use wire::{Caret, Guid, Hello, Presence, Spot};
19
20use mdns_sd::{IfKind, ServiceDaemon, ServiceEvent, ServiceInfo};
21use minicbor::Encode;
22use std::{
23 collections::BTreeMap,
24 io::{self, Read, Write},
25 net::{IpAddr, Ipv4Addr, Shutdown, SocketAddr, TcpListener, TcpStream},
26 sync::{
27 Arc, Mutex,
28 atomic::{AtomicBool, AtomicU64, Ordering},
29 mpsc,
30 },
31 thread,
32 time::{Duration, Instant},
33};
34use wire::{Sealer, Side, kind};
35
36const SERVICE: &str = "_snowbound._tcp.local.";
37/// How long a connection may be quiet before a ping, and before the peer counts as gone.
38const PING: Duration = Duration::from_secs(15);
39const GONE: Duration = Duration::from_secs(45);
40const OPENING: Duration = Duration::from_secs(5);
41/// The most often presence goes to a peer.
42const PRESENCE_EVERY: Duration = Duration::from_millis(100);
43/// The longest wait before meeting again.
44const PATIENCE: Duration = Duration::from_secs(30);
45/// Wrong tries of a code met off any relay before it admits no one new, as a relay burns one.
46const TRIES: u32 = 5;
47
48mod model;
49pub use model::{Event, Peer, Reach, Relayed, Room};
50use model::{code_parts, hex};
51
52/// Presence on the network while it lives; dropping it leaves.
53pub struct Live {
54 shared: Arc<Shared>,
55 address: SocketAddr,
56}
57
58struct Shared {
59 me: Hello,
60 room: Room,
61 secret: Vec<u8>,
62 reach: Option<Reach>,
63 state: Mutex<State>,
64 #[allow(clippy::type_complexity)]
65 events: Box<dyn Fn(&Arc<Shared>, Event) + Send + Sync>,
66 stopped: AtomicBool,
67 connections: AtomicU64,
68}
69
70#[derive(Default)]
71struct State {
72 presence: Presence,
73 /// Counts changes to `presence`, so a writer sends only the newest.
74 generation: u64,
75 peers: BTreeMap<[u8; 16], Link>,
76 dialing: BTreeMap<SocketAddr, thread::Thread>,
77 /// The code others type, once known.
78 code: Option<String>,
79 /// The relay connection open now, to hang up on leaving.
80 relay: Option<Arc<relay::Socket>>,
81 relayed: Option<Relayed>,
82 /// Meetings that failed on the secret: wrong codes tried here, or this end's.
83 failed: u32,
84 /// The code admits no one new.
85 burned: bool,
86 /// The Live Share version of the last peer met that speaks another.
87 outdated: Option<u16>,
88 daemon: Option<ServiceDaemon>,
89}
90
91impl State {
92 /// The relay's group, in a notebook's room while the relay is reached.
93 fn group(&self) -> Option<&relay::Group> {
94 self.relay.as_ref()?.group.as_ref()
95 }
96}
97
98/// Presence sent at most every `PRESENCE_EVERY`, and only the newest.
99#[derive(Default)]
100struct Paced {
101 due: Option<Instant>,
102 last: Option<Instant>,
103 sent: Option<u64>,
104}
105
106impl Paced {
107 /// How long to wait for news: until presence is due, else `idle`.
108 fn wait(&self, idle: Duration) -> Duration {
109 self.due
110 .map_or(idle, |due| due.saturating_duration_since(Instant::now()))
111 }
112
113 /// Hears that presence changed.
114 fn changed(&mut self) {
115 let last = self.last;
116 self.due
117 .get_or_insert_with(|| last.map_or_else(Instant::now, |at| at + PRESENCE_EVERY));
118 }
119
120 /// The presence to send now, where it is due and new.
121 fn due(&mut self, state: &Mutex<State>) -> Option<Presence> {
122 self.due.filter(|due| *due <= Instant::now())?;
123 self.due = None;
124 let state = state.lock().unwrap();
125 if self.sent == Some(state.generation) {
126 return None;
127 }
128 (self.sent, self.last) = (Some(state.generation), Some(Instant::now()));
129 Some(state.presence.clone())
130 }
131}
132
133struct Link {
134 connection: u64,
135 peer: Peer,
136 line: Line,
137 pipe: Arc<dyn Pipe>,
138}
139
140/// What a connection's writer sends next.
141enum Out {
142 /// The newest presence.
143 Presence,
144 Frame(u16, Vec<u8>),
145 /// A `Bye`, after which it hangs up.
146 Bye(Vec<u8>),
147}
148
149/// The way to one connected peer: frames sent on it go after those sent before.
150#[derive(Clone)]
151pub struct Line(mpsc::Sender<Out>);
152
153impl Line {
154 pub fn send(&self, kind: u16, body: &impl Encode<()>) -> io::Result<()> {
155 let body = minicbor::to_vec(body).map_err(io::Error::other)?;
156 self.0
157 .send(Out::Frame(kind, body))
158 .map_err(|_| io::ErrorKind::NotConnected.into())
159 }
160
161 /// Says `reason` after the frames sent before, then hangs up.
162 pub fn hang_up(&self, reason: &str) {
163 let bye = minicbor::to_vec(wire::Bye {
164 reason: reason.into(),
165 })
166 .unwrap_or_default();
167 let _ = self.0.send(Out::Bye(bye));
168 }
169}
170
171/// A stream to one peer, read by one thread and written by another: a TCP connection, or
172/// one carried through a relay.
173trait Pipe: Send + Sync {
174 fn read(&self, buffer: &mut [u8]) -> io::Result<usize>;
175 fn write(&self, bytes: &[u8]) -> io::Result<usize>;
176 fn set_read_timeout(&self, timeout: Duration) -> io::Result<()>;
177 /// Hangs up, ending the thread reading.
178 fn shutdown(&self);
179 /// Whether it goes straight to the peer, which is better than through a relay.
180 fn direct(&self) -> bool;
181 /// Hears whether the peer at the other end knew the secret.
182 fn met(&self, _met: bool) {}
183 /// Hears that the stream failed, after which this end meets its peers again.
184 fn broken(&self) {}
185}
186
187impl Pipe for TcpStream {
188 fn read(&self, buffer: &mut [u8]) -> io::Result<usize> {
189 Read::read(&mut &*self, buffer)
190 }
191
192 fn write(&self, bytes: &[u8]) -> io::Result<usize> {
193 Write::write(&mut &*self, bytes)
194 }
195
196 fn set_read_timeout(&self, timeout: Duration) -> io::Result<()> {
197 TcpStream::set_read_timeout(self, Some(timeout))
198 }
199
200 fn shutdown(&self) {
201 let _ = TcpStream::shutdown(self, Shutdown::Both);
202 }
203
204 fn direct(&self) -> bool {
205 true
206 }
207}
208
209impl Read for &dyn Pipe {
210 fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
211 Pipe::read(*self, buffer)
212 }
213}
214
215impl Write for &dyn Pipe {
216 fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
217 Pipe::write(*self, bytes)
218 }
219
220 fn flush(&mut self) -> io::Result<()> {
221 Ok(())
222 }
223}
224
225impl Live {
226 /// Starts listening as `me` in `room`, advertised and looked for where `reach` says, or
227 /// not at all with `None`, leaving peers to `connect`, and in the room at `relay`
228 /// (`wss://live.example.net`) where given. `events` runs on a network thread.
229 pub fn start(
230 me: Hello,
231 room: &Room,
232 reach: Option<Reach>,
233 relay: Option<&str>,
234 events: impl Fn(Event) + Send + Sync + 'static,
235 ) -> io::Result<Live> {
236 Self::start_with(me, room, reach, relay, move |_, event| events(event))
237 }
238
239 fn start_with(
240 me: Hello,
241 room: &Room,
242 reach: Option<Reach>,
243 relay: Option<&str>,
244 events: impl Fn(&Arc<Shared>, Event) + Send + Sync + 'static,
245 ) -> io::Result<Live> {
246 let host = match reach {
247 Some(Reach::Network) => IpAddr::V4(Ipv4Addr::UNSPECIFIED),
248 Some(Reach::Loopback) | None => IpAddr::V4(Ipv4Addr::LOCALHOST),
249 };
250 let listener = TcpListener::bind((host, 0))?;
251 let mut address = listener.local_addr()?;
252 if address.ip().is_unspecified() {
253 address.set_ip(IpAddr::V4(Ipv4Addr::LOCALHOST));
254 }
255 let code = match room {
256 Room::Code { code, .. } => match code_parts(code) {
257 (Some(number), secret) => code::format(number, &secret),
258 // Off any relay, the end sharing a secret alone numbers it itself.
259 (None, secret) if relay.is_none() => {
260 let mut number = [0; 4];
261 getrandom::fill(&mut number)
262 .map_err(|_| io::Error::other("System random source failed"))?;
263 code::format(u32::from_le_bytes(number) % code::NAMEPLATES, &secret)
264 }
265 (None, _) => None,
266 },
267 Room::Notebook(_) => None,
268 };
269 let shared = Arc::new(Shared {
270 me,
271 room: room.clone(),
272 secret: room.secret(),
273 reach,
274 state: Mutex::new(State {
275 code,
276 ..State::default()
277 }),
278 events: Box::new(events),
279 stopped: AtomicBool::new(false),
280 connections: AtomicU64::new(0),
281 });
282 if let Some(relay) = relay {
283 relay::join(&shared, relay, address.port())?;
284 }
285 let accepting = Arc::clone(&shared);
286 thread::Builder::new()
287 .name("live accept".into())
288 .spawn(move || {
289 for stream in listener.incoming() {
290 if accepting.stopped.load(Ordering::Acquire) {
291 return;
292 }
293 if let (Ok(stream), Some(tag)) = (stream, accepting.tag()) {
294 let shared = Arc::clone(&accepting);
295 thread::spawn(move || shared.run(Arc::new(stream), Side::Responder, &tag));
296 }
297 }
298 })?;
299 shared.advertise(address.port());
300 Ok(Live { shared, address })
301 }
302
303 /// Where this end listens.
304 pub fn address(&self) -> SocketAddr {
305 self.address
306 }
307
308 /// The code others type to meet this end: the one it was given, or the one the relay
309 /// or this end numbered.
310 pub fn code(&self) -> Option<String> {
311 self.shared.state.lock().unwrap().code.clone()
312 }
313
314 /// How the relay last answered.
315 pub fn relayed(&self) -> Relayed {
316 let state = self.shared.state.lock().unwrap();
317 state.relayed.clone().unwrap_or(Relayed::Unknown)
318 }
319
320 /// Meetings that failed on the secret: wrong tries of this end's code, or this end's own
321 /// wrong code.
322 pub fn failed(&self) -> u32 {
323 self.shared.state.lock().unwrap().failed
324 }
325
326 /// The Live Share version of the last peer met that speaks another, which one of the two
327 /// must update to meet.
328 pub fn other_version(&self) -> Option<u16> {
329 self.shared.state.lock().unwrap().outdated
330 }
331
332 /// Whether this end's code had too many wrong tries and admits no one new.
333 pub fn burned(&self) -> bool {
334 self.shared.state.lock().unwrap().burned
335 }
336
337 /// Connects to a peer at `address` that discovery did not find.
338 pub fn connect(&self, address: SocketAddr) {
339 self.shared.dial(address);
340 }
341
342 /// Says where this end is now; peers hear only the newest of quick changes, at most every
343 /// tenth of a second.
344 pub fn set_presence(&self, presence: Presence) {
345 let mut state = self.shared.state.lock().unwrap();
346 if state.presence == presence {
347 return;
348 }
349 state.presence = presence;
350 state.generation += 1;
351 for link in state.peers.values() {
352 if self.shared.carries_presence(&*link.pipe) {
353 let _ = link.line.0.send(Out::Presence);
354 }
355 }
356 if let Some(group) = state.group() {
357 group.send(relay::Out::Presence);
358 }
359 }
360
361 /// The peers in the room now, by id.
362 pub fn peers(&self) -> Vec<Peer> {
363 self.shared.peers()
364 }
365
366 /// A way to send to this room's peers that doesn't keep it open.
367 pub fn sender(&self) -> Sender {
368 Sender(Arc::downgrade(&self.shared))
369 }
370
371 /// The line to `peer`, while it is connected.
372 pub fn line(&self, peer: &[u8; 16]) -> Option<Line> {
373 let state = self.shared.state.lock().unwrap();
374 state.peers.get(peer).map(|link| link.line.clone())
375 }
376
377 /// Says `reason` to every peer and leaves, waiting a moment for them to hear it.
378 pub fn leave(self, reason: &str) {
379 let bye = minicbor::to_vec(wire::Bye {
380 reason: reason.into(),
381 })
382 .unwrap_or_default();
383 for link in self.shared.state.lock().unwrap().peers.values() {
384 let _ = link.line.0.send(Out::Bye(bye.clone()));
385 }
386 let deadline = Instant::now() + Duration::from_secs(1);
387 while !self.shared.state.lock().unwrap().peers.is_empty() && Instant::now() < deadline {
388 thread::sleep(Duration::from_millis(20));
389 }
390 }
391}
392
393/// Sends to a room's peers while the room is open.
394#[derive(Clone)]
395pub struct Sender(std::sync::Weak<Shared>);
396
397impl Sender {
398 /// Sends message `kind` holding `body` to the peers `to` names, or to everyone: once to
399 /// the room's group through the relay, and to each peer met directly.
400 pub fn send(&self, kind: u16, body: &impl Encode<()>, to: Option<&[[u8; 16]]>) {
401 let (Some(shared), Ok(body)) = (self.0.upgrade(), minicbor::to_vec(body)) else {
402 return;
403 };
404 let state = shared.state.lock().unwrap();
405 let group = state.group();
406 let named = |id: &[u8; 16]| to.is_none_or(|to| to.contains(id));
407 let mut slots = Vec::new();
408 for (id, link) in state.peers.iter().filter(|(id, _)| named(id)) {
409 match group {
410 Some(group) if !link.pipe.direct() => slots.extend(group.slot(id)),
411 _ => {
412 let _ = link.line.0.send(Out::Frame(kind, body.clone()));
413 }
414 }
415 }
416 let Some(group) = group else {
417 return;
418 };
419 match to {
420 None => group.send(relay::Out::Frame(kind, body, None)),
421 Some(to) => {
422 let unlinked = to.iter().filter(|id| !state.peers.contains_key(*id));
423 slots.extend(unlinked.filter_map(|id| group.slot(id)));
424 if !slots.is_empty() {
425 group.send(relay::Out::Frame(kind, body, Some(slots)));
426 }
427 }
428 }
429 }
430}
431
432impl Drop for Live {
433 fn drop(&mut self) {
434 self.shared.stopped.store(true, Ordering::Release);
435 // Wakes the accepting thread to see it has stopped.
436 let _ = TcpStream::connect_timeout(&self.address, OPENING);
437 let mut state = self.shared.state.lock().unwrap();
438 for dial in state.dialing.values() {
439 dial.unpark();
440 }
441 if let Some(daemon) = state.daemon.take() {
442 let _ = daemon.shutdown();
443 }
444 for link in state.peers.values() {
445 link.pipe.shutdown();
446 }
447 if let Some(relay) = &state.relay {
448 relay.hang_up();
449 }
450 }
451}
452
453impl Shared {
454 fn gone(&self, hello: &Arc<Hello>) -> bool {
455 !self.state.lock().unwrap().peers.contains_key(&hello.peer)
456 }
457
458 /// Whether `hello` is how a peer is in the room now: its link, or its place in the group.
459 fn connected(&self, hello: &Arc<Hello>) -> bool {
460 let current = |peer: &Peer| Arc::ptr_eq(&peer.hello, hello);
461 let state = self.state.lock().unwrap();
462 !self.stopped.load(Ordering::Acquire)
463 && (state
464 .peers
465 .get(&hello.peer)
466 .is_some_and(|link| current(&link.peer))
467 || state.group().is_some_and(|group| {
468 (group.members.lock().unwrap().values())
469 .any(|member| member.peer.as_ref().is_some_and(current))
470 }))
471 }
472
473 /// The peers in the room: those met directly, and the rest as the relay's group or a
474 /// stream through the relay last heard of them.
475 fn peers(&self) -> Vec<Peer> {
476 let state = self.state.lock().unwrap();
477 let mut peers = BTreeMap::new();
478 if let Some(group) = state.group() {
479 for member in group.members.lock().unwrap().values() {
480 if let Some(peer) = &member.peer {
481 peers.insert(peer.hello.peer, peer.clone());
482 }
483 }
484 }
485 for (id, link) in &state.peers {
486 if link.pipe.direct() || !peers.contains_key(id) {
487 peers.insert(*id, link.peer.clone());
488 }
489 }
490 peers.into_values().collect()
491 }
492
493 /// Whether peer `id` is met directly, which is where it says everything.
494 fn direct(&self, id: &[u8; 16]) -> bool {
495 let state = self.state.lock().unwrap();
496 state.peers.get(id).is_some_and(|link| link.pipe.direct())
497 }
498
499 /// Whether presence goes on `pipe`: not on a stream through the relay in a notebook's
500 /// room, whose group carries it.
501 fn carries_presence(&self, pipe: &dyn Pipe) -> bool {
502 pipe.direct() || matches!(self.room, Room::Code { .. })
503 }
504
505 /// The room's tag as it stands: a code's, once numbered.
506 fn tag(&self) -> Option<String> {
507 match &self.room {
508 Room::Code { .. } => code_parts(self.state.lock().unwrap().code.as_deref()?)
509 .0
510 .map(|number| format!("code-{number}")),
511 room => room.tag(),
512 }
513 }
514
515 /// Advertises this end on `port` once its room has a tag, where its reach says, and
516 /// connects to the peers in its room that discovery finds with a higher id than its own,
517 /// which leave the connecting to it.
518 fn advertise(self: &Arc<Self>, port: u16) {
519 let (Some(reach), Some(tag)) = (self.reach, self.tag()) else {
520 return;
521 };
522 let mut state = self.state.lock().unwrap();
523 if state.daemon.is_some() || self.stopped.load(Ordering::Acquire) {
524 return;
525 }
526 match discover(self, reach, tag, port) {
527 Ok(daemon) => state.daemon = Some(daemon),
528 Err(error) => eprintln!("Live: no discovery: {error}"),
529 }
530 }
531
532 fn knows(&self, peer: &str) -> bool {
533 let state = self.state.lock().unwrap();
534 state.peers.keys().any(|id| hex(id) == peer)
535 }
536
537 /// Keeps one dial to `address`, meeting again after a transport failure.
538 fn dial(self: &Arc<Self>, address: SocketAddr) {
539 let mut state = self.state.lock().unwrap();
540 if self.stopped.load(Ordering::Acquire)
541 || state.burned
542 || state.dialing.contains_key(&address)
543 {
544 return;
545 }
546 let shared = Arc::downgrade(self);
547 let dialing = thread::spawn(move || {
548 let mut wait = Duration::from_secs(1);
549 loop {
550 let Some(live) = shared.upgrade() else {
551 return;
552 };
553 if live.stopped.load(Ordering::Acquire) || live.state.lock().unwrap().burned {
554 break;
555 }
556 let Some(tag) = live.tag() else {
557 break;
558 };
559 drop(live);
560 if let Ok(stream) = TcpStream::connect_timeout(&address, OPENING) {
561 let Some(live) = shared.upgrade() else {
562 return;
563 };
564 if !live.run(Arc::new(stream), Side::Initiator, &tag) {
565 break;
566 }
567 }
568 thread::park_timeout(wait);
569 wait = (wait * 2).min(PATIENCE);
570 }
571 if let Some(live) = shared.upgrade() {
572 live.state.lock().unwrap().dialing.remove(&address);
573 }
574 });
575 state.dialing.insert(address, dialing.thread().clone());
576 }
577
578 /// Counts a meeting that failed on the secret: the end sharing a code burns it after too
579 /// many, and an end that typed one gives up at once.
580 fn failed(self: &Arc<Self>) {
581 let mut state = self.state.lock().unwrap();
582 state.failed += 1;
583 match &self.room {
584 Room::Code { owner: true, .. } if state.failed >= TRIES => state.burned = true,
585 Room::Code { owner: false, .. } => self.stopped.store(true, Ordering::Release),
586 _ => {}
587 }
588 drop(state);
589 (self.events)(self, Event::Changed);
590 }
591
592 /// Meets the peer at the other end of `pipe` in room `tag`, then reads from it until it
593 /// goes: whether the next dial may try again.
594 fn run(self: Arc<Self>, pipe: Arc<dyn Pipe>, side: Side, tag: &str) -> bool {
595 if self.stopped.load(Ordering::Acquire) || self.state.lock().unwrap().burned {
596 pipe.shutdown();
597 return false;
598 }
599 let mut stream: &dyn Pipe = &*pipe;
600 let met = (|| {
601 stream.set_read_timeout(OPENING)?;
602 let (mut send, mut receive) = wire::open(&mut stream, side, tag, &self.secret)?;
603 send.send(&mut stream, kind::HELLO, &self.me)?;
604 let (first, body) = receive.receive(&mut stream)?;
605 if first != kind::HELLO {
606 return Err(io::Error::new(io::ErrorKind::InvalidData, "No hello"));
607 }
608 let hello: Hello = minicbor::decode(&body)
609 .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "A malformed hello"))?;
610 stream.set_read_timeout(GONE)?;
611 Ok((send, receive, hello))
612 })();
613 let other = (met.as_ref().err())
614 .and_then(|error| error.get_ref()?.downcast_ref::<wire::Version>())
615 .copied();
616 let (send, mut receive, hello) = match met {
617 Ok(met) => {
618 pipe.met(true);
619 met
620 }
621 Err(error) => {
622 eprintln!("Live: no meeting in {tag}: {error}");
623 if let Some(wire::Version(version)) = other {
624 // Another version guessed nothing: no keys were agreed.
625 pipe.met(true);
626 self.state.lock().unwrap().outdated = Some(version);
627 if matches!(self.room, Room::Code { owner: false, .. }) {
628 self.stopped.store(true, Ordering::Release);
629 }
630 (self.events)(&self, Event::Changed);
631 } else if error.kind() == io::ErrorKind::InvalidData {
632 pipe.met(false);
633 self.failed();
634 } else {
635 pipe.broken();
636 }
637 pipe.shutdown();
638 return other.is_none() && error.kind() != io::ErrorKind::InvalidData;
639 }
640 };
641 let peer = hello.peer;
642 let name = hello.name.clone();
643 let hello = Arc::new(hello);
644 let connection = self.connections.fetch_add(1, Ordering::Relaxed);
645 let (out, outgoing) = mpsc::channel();
646 let line = Line(out);
647 let retired = {
648 let mut state = self.state.lock().unwrap();
649 // A peer met both directly and through a relay keeps the direct connection, as
650 // both ends then agree.
651 let kept = match state.peers.get(&peer) {
652 _ if peer == self.me.peer => false,
653 Some(link) if link.pipe.direct() || !pipe.direct() => false,
654 Some(_) => true,
655 None => true,
656 };
657 if !kept {
658 drop(state);
659 pipe.shutdown();
660 return false;
661 }
662 state.peers.insert(
663 peer,
664 Link {
665 connection,
666 peer: Peer {
667 hello: Arc::clone(&hello),
668 presence: None,
669 },
670 line: line.clone(),
671 pipe: Arc::clone(&pipe),
672 },
673 )
674 };
675 if let Some(retired) = retired {
676 retired.pipe.shutdown();
677 }
678 if self.carries_presence(&*pipe) {
679 let _ = line.0.send(Out::Presence);
680 }
681 let writing = Arc::clone(&pipe);
682 let shared = Arc::clone(&self);
683 thread::spawn(move || shared.write(&*writing, send, outgoing));
684 (self.events)(&self, Event::Met(&hello, &line));
685 (self.events)(&self, Event::Changed);
686 let ended = loop {
687 let (message, body) = match receive.receive(&mut stream) {
688 Ok(frame) => frame,
689 Err(error) => break Some(error),
690 };
691 match message {
692 kind::HELLO | kind::PING => {}
693 kind::PRESENCE => {
694 let Ok(presence) = minicbor::decode::<Presence>(&body) else {
695 break None;
696 };
697 if let Some(link) = self.state.lock().unwrap().peers.get_mut(&peer)
698 && link.connection == connection
699 {
700 link.peer.presence = Some(presence);
701 }
702 (self.events)(&self, Event::Changed);
703 }
704 kind => {
705 (self.events)(
706 &self,
707 Event::Frame {
708 from: &hello,
709 kind,
710 body: &body,
711 },
712 );
713 // The peer hangs up after its bye.
714 if kind == kind::BYE {
715 break None;
716 }
717 }
718 }
719 };
720 if let Some(error) = ended {
721 eprintln!("Live: the connection to {name} broke ({error}); meeting again");
722 pipe.broken();
723 }
724 pipe.shutdown();
725 let mut state = self.state.lock().unwrap();
726 if state
727 .peers
728 .get(&peer)
729 .is_some_and(|link| link.connection == connection)
730 {
731 state.peers.remove(&peer);
732 drop(state);
733 (self.events)(&self, Event::Left(&hello));
734 (self.events)(&self, Event::Changed);
735 }
736 true
737 }
738
739 /// Sends the newest presence when woken, at most every `PRESENCE_EVERY`, frames as they
740 /// come, and a ping when quiet.
741 fn write(&self, pipe: &dyn Pipe, mut send: Sealer, outgoing: mpsc::Receiver<Out>) {
742 let mut stream = pipe;
743 let mut paced = Paced::default();
744 loop {
745 let result = match outgoing.recv_timeout(paced.wait(PING)) {
746 Ok(Out::Presence) => {
747 paced.changed();
748 Ok(())
749 }
750 Ok(Out::Frame(kind, body)) => send.send_encoded(&mut stream, kind, &body),
751 Ok(Out::Bye(body)) => {
752 let _ = send.send_encoded(&mut stream, kind::BYE, &body);
753 pipe.shutdown();
754 return;
755 }
756 Err(mpsc::RecvTimeoutError::Timeout) if paced.due.is_none() => {
757 send.send(&mut stream, kind::PING, &())
758 }
759 Err(mpsc::RecvTimeoutError::Timeout) => Ok(()),
760 Err(mpsc::RecvTimeoutError::Disconnected) => return,
761 };
762 let result = result.and_then(|()| match paced.due(&self.state) {
763 Some(presence) => send.send(&mut stream, kind::PRESENCE, &presence),
764 None => Ok(()),
765 });
766 if result.is_err() {
767 pipe.broken();
768 pipe.shutdown();
769 return;
770 }
771 }
772 }
773}
774
775/// Advertises `shared` as in room `tag` on `port` and connects to the peers in its room that
776/// discovery finds with a higher id than its own.
777fn discover(
778 shared: &Arc<Shared>,
779 reach: Reach,
780 tag: String,
781 port: u16,
782) -> mdns_sd::Result<ServiceDaemon> {
783 let daemon = ServiceDaemon::new()?;
784 let id = hex(&shared.me.peer);
785 let properties = [("v", "1"), ("room", tag.as_str()), ("peer", id.as_str())];
786 let host = format!("snowbound-{id}.local.");
787 let info = match reach {
788 Reach::Network => {
789 ServiceInfo::new(SERVICE, &id, &host, "", port, &properties[..])?.enable_addr_auto()
790 }
791 Reach::Loopback => {
792 daemon.disable_interface(IfKind::All)?;
793 daemon.enable_interface(IfKind::LoopbackV4)?;
794 ServiceInfo::new(
795 SERVICE,
796 &id,
797 &host,
798 IpAddr::V4(Ipv4Addr::LOCALHOST),
799 port,
800 &properties[..],
801 )?
802 }
803 };
804 daemon.register(info)?;
805 let found = daemon.browse(SERVICE)?;
806 let shared = Arc::downgrade(shared);
807 thread::Builder::new()
808 .name("live discovery".into())
809 .spawn(move || {
810 while let Ok(event) = found.recv() {
811 let ServiceEvent::ServiceResolved(service) = event else {
812 continue;
813 };
814 let Some(shared) = shared.upgrade() else {
815 return;
816 };
817 let (Some(room), Some(peer)) = (
818 service.get_property_val_str("room"),
819 service.get_property_val_str("peer"),
820 ) else {
821 continue;
822 };
823 if room != tag || peer <= id.as_str() || shared.knows(peer) {
824 continue;
825 }
826 let mut addresses: Vec<IpAddr> =
827 service.addresses.iter().map(|ip| ip.to_ip_addr()).collect();
828 addresses.sort_by_key(|ip| (!ip.is_ipv4(), !ip.is_loopback()));
829 if let Some(ip) = addresses.first() {
830 let address = SocketAddr::new(*ip, service.port);
831 shared.dial(address);
832 }
833 }
834 })
835 .map_err(|error| mdns_sd::Error::Msg(error.to_string()))?;
836 Ok(daemon)
837}
838
839/// `text` with `%XX` escapes decoded, as a URL's name and password.
840fn decode(text: &str) -> String {
841 let bytes = text.as_bytes();
842 let mut decoded = Vec::with_capacity(bytes.len());
843 let mut at = 0;
844 while at < bytes.len() {
845 match text
846 .get(at + 1..at + 3)
847 .filter(|_| bytes[at] == b'%')
848 .and_then(|hex| u8::from_str_radix(hex, 16).ok())
849 {
850 Some(byte) => {
851 decoded.push(byte);
852 at += 3;
853 }
854 None => {
855 decoded.push(bytes[at]);
856 at += 1;
857 }
858 }
859 }
860 String::from_utf8_lossy(&decoded).into_owned()
861}
862
863#[cfg(test)]
864mod tests;