wait-on-multiple-receivers-aggregator.rs

// in cargo.toml
// rand = "0.7"
// parking_lot = "0.11"
// crossbeam-channel = "0.5"

use std::{thread::{JoinHandle, sleep, spawn}, time::Duration};

use crossbeam_channel::{Sender, unbounded};
use rand::{Rng, thread_rng};

const MESSAGES: u8 = 128;
const MESSAGE_GEN: usize = 256;
const DELAY_MAX: u64 = 5;

enum MainEvent {
    EventA(EventA),
    EventB(EventB),
    EventC(EventC),
}

#[derive(Debug)]
enum EventA {
    MsgOne,
    MsgTwo,
    MsgThree,
}

#[derive(Debug)]
enum EventB {
    MsgOne,
    MsgTwo,
    MsgThree,
}

#[derive(Debug)]
enum EventC {
    MsgOne,
    MsgTwo,
    MsgThree,
}

fn main() {
    let (tx, rx) = unbounded();
    let mut handles = vec![
        process_a(tx.clone()),
        process_b(tx.clone()),
        process_c(tx)
    ];

    handles.push(
        spawn(move || {
            loop {
                match rx.recv() {
                    Ok(MainEvent::EventA(event)) => println!("A: {:?}", event),
                    Ok(MainEvent::EventB(event)) => println!("B: {:?}", event),
                    Ok(MainEvent::EventC(event)) => println!("C: {:?}", event),
                    Err(..) => break
                }
            }
        })
    );

    for handle in handles {
        handle.join();
    }
}

fn generate_a(tx: Sender<EventA>) {
    spawn(move || {
        let mut rng = thread_rng();
        let mut counter = 0u8;

        loop {
            counter += 1;
            if counter > MESSAGES {
                println!("A: Done");
                break;
            }

            sleep(Duration::from_millis(rng.gen_range(0, DELAY_MAX)));

            tx.send(
                match rng.gen_range(0, MESSAGE_GEN) % 3 {
                    0 => EventA::MsgOne,
                    1 => EventA::MsgTwo,
                    2 => EventA::MsgThree,
                    _ => unreachable!()
                }
            ).unwrap()
        }
    });
}

fn generate_b(tx: Sender<EventB>) {
    spawn(move || {
        let mut rng = thread_rng();
        let mut counter = 0u8;

        loop {
            counter += 1;
            if counter > MESSAGES {
                println!("B: Done");
                break;
            }

            sleep(Duration::from_millis(rng.gen_range(0, DELAY_MAX)));

            tx.send(
                match rng.gen_range(0, MESSAGE_GEN) % 3 {
                    0 => EventB::MsgOne,
                    1 => EventB::MsgTwo,
                    2 => EventB::MsgThree,
                    _ => unreachable!()
                }
            ).unwrap()
        }
    });
}

fn generate_c(tx: Sender<EventC>) {
    spawn(move || {
        let mut rng = thread_rng();
        let mut counter = 0u8;

        loop {
            counter += 1;
            if counter > MESSAGES {
                println!("C: Done");
                break;
            }

            sleep(Duration::from_millis(rng.gen_range(0, DELAY_MAX)));

            tx.send(
                match rng.gen_range(0, MESSAGE_GEN) % 3 {
                    0 => EventC::MsgOne,
                    1 => EventC::MsgTwo,
                    2 => EventC::MsgThree,
                    _ => unreachable!()
                }
            ).unwrap()
        }
    });
}

fn process_a(main_tx: Sender<MainEvent>) -> JoinHandle<()> {
    let (tx, rx) = unbounded();

    generate_a(tx);

    spawn(move || {
        loop {
            match rx.recv() {
                Ok(event) => { main_tx.send(MainEvent::EventA(event)); }
                Err(..) => break
            }
        }
    })
}

//noinspection DuplicatedCode
fn process_b(main_tx: Sender<MainEvent>) -> JoinHandle<()> {
    let (tx, rx) = unbounded();

    generate_b(tx);

    spawn(move || {
        loop {
            match rx.recv() {
                Ok(event) => { main_tx.send(MainEvent::EventB(event)); }
                Err(..) => break
            }
        }
    })
}

//noinspection DuplicatedCode
fn process_c(main_tx: Sender<MainEvent>) -> JoinHandle<()> {
    let (tx, rx) = unbounded();

    generate_c(tx);

    spawn(move || {
        loop {
            match rx.recv() {
                Ok(event) => { main_tx.send(MainEvent::EventC(event)); }
                Err(..) => break
            }
        }
    })
}

this example works with std::sync::mpsc::channel as well