How to select! on a list of Receiver?

[xuxiaocheng0201]

There is a Vec<tokio::sync::watch::Receiver<T>> with unknown length. And I want to receive any change from them, just like select!.
I don't known how to write the code. Is there any well solution?
My current idea is to spawn a list of task, and call join_next in a JoinSet. But spawning such many tasks seem unnecessary.

pub struct MultiWatcher<'a, T> {
    watches: &'a mut [Receiver<T>],
}
impl<'a, T> MultiWatcher<'a, T> {
    pub fn new(watches: &'a mut [Receiver<T>]) -> Self {
        Self { watches }
    }
    /// Return the index of the changed `watches`, or `None` if all channels are closed.
    pub async fn wait_next_change(&mut self) -> Option<usize> {
        todo!()
    }
}

[Darksonn]

Usually the answer to this question is "do not create multiple channels". You can clone the sender to obtain many senders that send messages to the same receiver. By doing that, you can usually rewrite your code to only have a single receiver.

[xuxiaocheng0201]

Thank you for your prompt response. But I do need to create multiple channls, which come from different independent tasks. The upstream api is designed like: pub async fn piece_task(id: usize, progress: tokio::sync::watch::Sender<usize>). To run multiple tasks concurrently and transmit the overall progress downstream, I have to use something like this. Anyway, thank you very much.

[Darksonn]

You can combine:

• https://docs.rs/tokio-stream/latest/tokio_stream/wrappers/struct.ReceiverStream.html
• https://docs.rs/futures/latest/futures/stream/struct.SelectAll.html

[xuxiaocheng0201]

Thank you. But the tokio_stream::wrappers::ReceiverStream requires tokio::sync::mpsc::Receiver<T> while my type is tokio::sync::watch::Receiver<T>.

[xuxiaocheng0201]

This is my final solution. Thank you very much for telling me the struct SelectAll.

pub struct MultiWatcher<T> {
    watches: Vec<Receiver<T>>,
}

impl<T> MultiWatcher<T> {
    pub fn new(watches: Vec<Receiver<T>>) -> Self {
        Self { watches, }
    }

    pub async fn wait_next_change(&mut self) -> Option<usize> {
        let mut futures = self.watches.iter_mut().map(|a| Box::pin(a.changed())).collect::<Vec<_>>();
        while !futures.is_empty() {
            let (res, index, rest) = futures::future::select_all(futures).await;
            if let Ok(()) = res { return Some(index) }
            futures = rest;
        }
        None
    }
}