beam/adapters/multicast.rs
1//! UDP multicast LAN discovery and sync adapter.
2//!
3//! [`Multicast`] uses UDP multicast to discover and sync with BEAM peers on
4//! the local network. It broadcasts `Put` and `Get` messages to a multicast
5//! group, enabling zero-config peer discovery on LANs.
6//!
7//! # Configuration
8//!
9//! - Multicast group: `233.255.255.255:7654`
10//! - Buffer size: 64 KB
11//! - Interfaces: all IPv4 interfaces
12//!
13//! # Behavior
14//!
15//! - `pre_start`: Joins the multicast group and starts a blocking receive
16//! loop in a `blocking_child_task`
17//! - `handle`: Broadcasts outgoing `Put` and `Get` messages to the group
18//! - Incoming messages are parsed and forwarded to the [`crate::router::Router`]
19//! - Marks itself as `subscribe_to_everything` (receives all messages)
20//!
21//! # Limitations
22//!
23//! The receive loop uses `blocking_child_task` — the `MulticastSocket::receive`
24//! call is synchronous and blocks. This is not optimal for async contexts
25//! but is required by the `multicast_socket` crate's API.
26
27use oko_multicast_socket::{MulticastOptions, MulticastSocket, all_ipv4_interfaces};
28use std::net::SocketAddrV4;
29
30use crate::Config;
31use crate::actor::{Actor, ActorContext};
32use crate::message::Message;
33use async_trait::async_trait;
34use log::{debug, error, info};
35use std::sync::Arc;
36use tokio::sync::RwLock;
37
38/// UDP multicast adapter for LAN peer discovery and sync.
39///
40/// Broadcasts Gun protocol messages to the multicast group `233.255.255.255:7654`
41/// and receives messages from other peers on the same LAN.
42pub struct Multicast {
43 socket: Arc<RwLock<MulticastSocket>>,
44 config: Config,
45}
46
47impl Multicast {
48 /// Creates a new multicast adapter bound to the default group.
49 ///
50 /// # Panics
51 ///
52 /// Panics if the multicast socket cannot be created (e.g. no network
53 /// interfaces available, or port 7654 is in use).
54 pub fn new(config: Config) -> Self {
55 let bind_address = SocketAddrV4::new([233, 255, 255, 255].into(), 7654);
56 let options = MulticastOptions {
57 buffer_size: 64 * 1024,
58 ..MulticastOptions::default()
59 };
60 let interfaces = all_ipv4_interfaces().expect("could not list multicast interfaces");
61 let socket = MulticastSocket::with_options(bind_address, interfaces, options)
62 .expect("could not create and bind multicast socket");
63 let socket = Arc::new(RwLock::new(socket));
64 Multicast { socket, config }
65 }
66
67 /// Parses an incoming multicast message and forwards it to the router.
68 ///
69 /// Only `Put` and `Get` messages are forwarded — other message types
70 /// (Hi, Flush, RtcSignal) are not meaningful over multicast.
71 fn handle_incoming_message(data: &str, ctx: &ActorContext, allow_public_space: bool) {
72 debug!("in {}", data);
73 match Message::try_from(data, ctx.addr.clone(), allow_public_space) {
74 Ok(msgs) => {
75 for msg in msgs.into_iter() {
76 match msg {
77 Message::Put(put) => {
78 let put = put.clone();
79 if let Err(e) = ctx.router.send(Message::Put(put)) {
80 error!("failed to send message to node: {:?}", e);
81 }
82 }
83 Message::Get(get) => {
84 let get = get.clone();
85 if let Err(e) = ctx.router.send(Message::Get(get)) {
86 error!("failed to send message to node: {:?}", e);
87 }
88 }
89 _ => {}
90 }
91 }
92 }
93 Err(e) => error!("message parsing failed: {}", e),
94 }
95 }
96}
97
98#[async_trait]
99impl Actor for Multicast {
100 async fn handle(&mut self, msg: Message, ctx: &ActorContext) {
101 debug!("out {}", msg.get_id());
102 if msg.is_from(&ctx.addr) {
103 return;
104 }
105 match msg {
106 Message::Put(mut put) => {
107 if let Err(e) = self
108 .socket
109 .read()
110 .await
111 .broadcast(put.to_string().as_bytes())
112 {
113 error!("multicast send error {}", e);
114 }
115 }
116 Message::Get(get) => {
117 if let Err(e) = self
118 .socket
119 .read()
120 .await
121 .broadcast(get.to_string().as_bytes())
122 {
123 error!("multicast send error {}", e);
124 }
125 }
126 _ => {
127 debug!("not sending");
128 }
129 }
130 }
131
132 /// Returns `true` — multicast subscribes to all messages.
133 fn subscribe_to_everything(&self) -> bool {
134 true
135 }
136
137 async fn pre_start(&mut self, ctx: &ActorContext) {
138 info!("Syncing over multicast\n");
139
140 let ctx_clone = ctx.clone();
141
142 let bind_address = SocketAddrV4::new([233, 255, 255, 255].into(), 7654);
143 let options = MulticastOptions {
144 buffer_size: 64 * 1024,
145 ..MulticastOptions::default()
146 };
147 let interfaces = all_ipv4_interfaces().expect("could not list multicast interfaces");
148 let socket = MulticastSocket::with_options(bind_address, interfaces, options)
149 .expect("could not create and bind multicast socket");
150
151 let allow_public_space = self.config.allow_public_space;
152 ctx.blocking_child_task(move || {
153 // blocking — not optimal!
154 loop {
155 if let Ok(message) = socket.receive() {
156 // TODO: if message.from == multicast_[interface], don't resend to [interface]
157 if let Ok(data) = std::str::from_utf8(&message.data) {
158 Self::handle_incoming_message(data, &ctx_clone, allow_public_space);
159 }
160 }
161 if *ctx_clone.is_stopped.read() {
162 break;
163 }
164 }
165 });
166 }
167
168 async fn stopping(&mut self, _ctx: &ActorContext) {
169 // The blocking child task checks is_stopped and will break on the
170 // next iteration. The multicast socket is dropped when the task
171 // completes. No additional cleanup needed.
172 info!("Multicast stopping");
173 }
174}