🎬 Fearless Concurrency Threads, Arc>, channels — safe parallelism enforced by the compiler.

📝 Text version (for readers / accessibility)

• std::thread::spawn creates OS threads — closures must be Send + 'static

• Arc> provides shared mutable state across threads safely

• Channels (mpsc) enable message passing — multiple producers, single consumer

• Send and Sync marker traits enforce thread safety at compile time

• Data races are impossible — the type system prevents them before your code runs

349: Broadcast Channel

Difficulty: 3 Level: Advanced One sender, many receivers — every subscriber gets a copy of every message. Pub/sub for async tasks.

The Problem This Solves

Sometimes you need to send the same event to multiple independent consumers. A configuration change event should reach every worker. A system shutdown signal should notify all background tasks. A price update should be delivered to every connected WebSocket client. A plain `mpsc` channel has one receiver — the first one to call `recv()` gets the message; others miss it. A broadcast channel (also called a fan-out channel) delivers each message to all active subscribers independently. Every subscriber has its own queue; the sender clones the message and pushes it to each queue. New subscribers join and receive messages from that point forward. `tokio::sync::broadcast` is the production-ready version with FIFO ordering, lagged receiver detection, and configurable capacity. This example builds the same concept with `std::sync::mpsc`.

The Intuition

Like Node.js `EventEmitter`:

emitter.on('update', handler1);
emitter.on('update', handler2);
emitter.emit('update', payload);  // both handlers run

Or JavaScript's `BroadcastChannel` for cross-tab communication. In Go, you'd fan out by ranging over a slice of channels. The broadcast channel abstracts this: one `send`, all subscribers receive.

How It Works in Rust

struct BroadcastSender<T: Clone> {
 subscribers: Arc<Mutex<Vec<mpsc::SyncSender<T>>>>,
}

impl<T: Clone + Send + 'static> BroadcastSender<T> {
 fn subscribe(&self, buf: usize) -> BroadcastReceiver<T> {
     let (tx, rx) = mpsc::sync_channel(buf);
     self.subscribers.lock().unwrap().push(tx);
     BroadcastReceiver { rx }
 }

 fn send(&self, msg: T) {
     let subs = self.subscribers.lock().unwrap();
     for sub in subs.iter() {
         // Clone message for each subscriber
         let _ = sub.try_send(msg.clone());
     }
 }
}

`T: Clone` is required — the sender clones the message once per subscriber. For expensive types, wrap in `Arc<T>` to make clones cheap (arc clone = reference count increment). In async Rust:

let (tx, _rx) = tokio::sync::broadcast::channel::<String>(16);
let rx1 = tx.subscribe();
let rx2 = tx.subscribe();
tx.send("hello".into())?;
// rx1 and rx2 both receive "hello"

What This Unlocks

Event bus — broadcast config changes, feature flag updates, or schema refreshes to all workers.
WebSocket fanout — each connected client subscribes; the server broadcasts price ticks or chat messages.
Graceful shutdown — broadcast a shutdown signal to all background tasks simultaneously.

Key Differences

Concept	OCaml	Rust
Broadcast	`Event.send` to list of channels	`BroadcastSender` with `Vec<SyncSender>`
Message delivery	`List.iter (Event.sync send)`	`try_send` to each subscriber's channel
Clone requirement	Values shared by GC reference	`T: Clone` — explicitly cloned per subscriber
Production crate	`Lwt_react` / custom	`tokio::sync::broadcast`

use std::sync::{Arc, Mutex};
use std::thread;
use std::sync::mpsc;

// Broadcast channel: one sender, many receivers each get a copy
struct BroadcastSender<T: Clone> {
    subscribers: Arc<Mutex<Vec<mpsc::SyncSender<T>>>>,
}

struct BroadcastReceiver<T> {
    rx: mpsc::Receiver<T>,
}

impl<T: Clone + Send + 'static> BroadcastSender<T> {
    fn new() -> Self {
        Self { subscribers: Arc::new(Mutex::new(Vec::new())) }
    }

    fn subscribe(&self, buf: usize) -> BroadcastReceiver<T> {
        let (tx, rx) = mpsc::sync_channel(buf);
        self.subscribers.lock().unwrap().push(tx);
        BroadcastReceiver { rx }
    }

    fn send(&self, msg: T) {
        let subs = self.subscribers.lock().unwrap();
        let mut alive = Vec::new();
        for sub in subs.iter() {
            // Clone for each subscriber
            if sub.try_send(msg.clone()).is_ok() {
                alive.push(sub.clone());
            }
        }
        // Note: we don't prune dead subs in this simple version
    }
}

impl<T> BroadcastReceiver<T> {
    fn recv(&self) -> Option<T> {
        self.rx.recv().ok()
    }
    fn try_recv(&self) -> Option<T> {
        self.rx.try_recv().ok()
    }
}

fn main() {
    let sender: BroadcastSender<String> = BroadcastSender::new();

    let r1 = sender.subscribe(16);
    let r2 = sender.subscribe(16);
    let r3 = sender.subscribe(16);

    let h1 = thread::spawn(move || {
        let msg = r1.recv().unwrap();
        println!("Receiver 1 got: {msg}");
        msg
    });
    let h2 = thread::spawn(move || {
        let msg = r2.recv().unwrap();
        println!("Receiver 2 got: {msg}");
        msg
    });
    let h3 = thread::spawn(move || {
        let msg = r3.recv().unwrap();
        println!("Receiver 3 got: {msg}");
        msg
    });

    thread::sleep(std::time::Duration::from_millis(5));
    sender.send("hello everyone".to_string());

    let (m1, m2, m3) = (h1.join().unwrap(), h2.join().unwrap(), h3.join().unwrap());
    assert_eq!(m1, m2);
    assert_eq!(m2, m3);
    println!("All received the same message ✓");
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn all_subscribers_get_message() {
        let s: BroadcastSender<i32> = BroadcastSender::new();
        let r1 = s.subscribe(4);
        let r2 = s.subscribe(4);
        s.send(42);
        assert_eq!(r1.recv(), Some(42));
        assert_eq!(r2.recv(), Some(42));
    }
    #[test]
    fn multiple_messages() {
        let s: BroadcastSender<i32> = BroadcastSender::new();
        let r = s.subscribe(8);
        for i in 0..5 { s.send(i); }
        let msgs: Vec<_> = (0..5).map(|_| r.recv().unwrap()).collect();
        assert_eq!(msgs, vec![0,1,2,3,4]);
    }
}

(* OCaml: broadcast via multiple channels *)

let broadcast subscribers message =
  List.iter (fun ch ->
    Event.sync (Event.send ch message)
  ) subscribers

let () =
  let ch1 = Event.new_channel () in
  let ch2 = Event.new_channel () in
  let ch3 = Event.new_channel () in

  let listener label ch =
    Thread.create (fun () ->
      let msg = Event.sync (Event.receive ch) in
      Printf.printf "%s received: %s\n" label msg
    ) ()
  in

  let t1 = listener "A" ch1 in
  let t2 = listener "B" ch2 in
  let t3 = listener "C" ch3 in

  broadcast [ch1; ch2; ch3] "hello everyone";

  Thread.join t1;
  Thread.join t2;
  Thread.join t3