multicast.rs

use std::task;
use std::task::TaskBuilder;

enum MultiCastHackForSelect<T> {
    Msg(T),
    MsgChan(Chan<T>),
}

struct MultiCast<T> {
    priv ch: Chan<MultiCastHackForSelect<T>>
}
impl<T:Send+Clone> MultiCast<T> {
    fn new() -> MultiCast<T> {
        let (po, ch) = stream::<MultiCastHackForSelect<T>>();
        do named_task(~"mc").spawn {
            let mut chans: ~[Chan<T>] = ~[];
            loop {
                match po.try_recv() {
                    Some(Msg(msg)) => {
                        for c in chans.iter() {
                            c.send(msg.clone());
                        }
                    }
                    Some(MsgChan(ch)) => {
                        chans.push(ch);
                    }
                    None => break
                }
            }
        }
        MultiCast { ch: ch }
    }

    fn send(&self, msg: T) {
        self.ch.send(Msg(msg));
    }

    fn push(&self, ch: Chan<T>) {
        self.ch.send(MsgChan(ch));
    }
}

pub fn named_task(name: ~str) -> TaskBuilder {
    let mut ui_task = task::task();
    ui_task.name(name);
    ui_task
}

pub fn do_something(name: ~str, po: Port<int>) {
    do named_task(name.clone()).spawn {
        loop {
            match po.try_recv() {
                Some(i) => printfln!("%s got %i", name, i),
                None => {
                    printfln!("%s finished", name);
                    break;
                }
            }
        }
    }
}

fn main() {
    let (po0, ch0) = stream();
    let (po1, ch1) = stream();
    let (po2, ch2) = stream();

    do_something(~"po0", po0);
    do_something(~"po1", po1);
    do_something(~"po2", po2);

    let mc = MultiCast::new();
    mc.push(ch0);
    mc.push(ch1);
    for i in range(0, 5) {
        mc.send(i)
    }

    mc.push(ch2);
    for i in range(5, 10) {
        mc.send(i)
    }
}