foo.rs

pub struct MpscReceiverSet {
    receivers: Vec<MpscReceiver>,
}

impl MpscReceiverSet {
    pub fn new() -> Result<MpscReceiverSet,MpscError> {
        Ok(MpscReceiverSet {
            receivers: vec![],
        })
    }

    pub fn add(&mut self, receiver: MpscReceiver) -> Result<i64,MpscError> {
        self.receivers.push(receiver.consume());
        Ok(self.receivers.len() as i64)
    }

    pub fn select(&mut self) -> Result<Vec<MpscSelectionResult>,MpscError> {
        let select = mpsc::Select::new();
        let mut handles: Vec<mpsc::Handle<MpscChannelMessage>> = vec![];
        
        for r in &self.receivers {
            unsafe {
                let mut h = select.handle((*r.receiver.as_unsafe_cell().get()).as_ref().unwrap());
                h.add();
                handles.push(h);
            }
        }

        let id = select.wait();
        for (index,h) in handles.iter().enumerate() {
            if h.id() == id {
                let receiver = &self.receivers[index];
                match receiver.recv() {
                    Ok((data, channels, shmems)) =>
                        return Ok(vec![MpscSelectionResult::DataReceived(id as i64, data, channels, shmems)]),
                    Err(err) => {
                        match err {
                            MpscError::ChannelClosedError => return Ok(vec![MpscSelectionResult::ChannelClosed(id as i64)]),
                            _ => return Err(err),
                        }
                    }
                }
            }
        }
        return Err(MpscError::UnknownError);
    }
}