Bridging two mpsc channels

help
1

Today I wrote this function:

/// Bridge two channels.
///
/// This function will read all messages from `input`, convert them to the message type of `output`
/// and send it to the output channel.
///
/// This function will terminate once either the input channel is exhausted or the output channel
/// has been closed.
pub async fn bridge<T, U>(mut input: Receiver<T>, output: Sender<U>)
where
    T: Into<U>,
{
    while let Some(msg) = input.recv().await {
        if let Err(error) = output.send(msg.into()).await {
            error!("Target channel closed: {error}");
            break;
        }
    }
}

Usage:

tokio::spawn(bridge(self.callbacks, message_tx.clone()));

I feel like something like this should/could be a general purpose library function.
I looked for something like this in tokio and searched for other crates, but couldn't find anything like it.
Is there something I overlooked in terms of library functions to bridge two mpsc channels or is there even a better way to do this?

2

you can convert the sender into a futures::Sink, and the receiver into a futures::Stream, and then you can use StreamExt::forward(). (use StreamExt::map() to convert the items when necessary)

but tbh I prefer the manual loop.

3

Speaking of edge cases (by the way you're logging it, it wouldn't be a standard functionality), the last msg is lost on error. You can avoid it by reserving a permit first.

pub async fn bridge<T, U>(mut input: Receiver<T>, output: Sender<U>)
where
    T: Into<U>,
{
    let Ok(mut place) = output.reserve().await else { return; }
    while let Some(msg) = input.recv().await {
        place.send(msg.into());
        place = match output.reserve().await {
            Ok(new_permit) => new_permit,
            Err(_) => return
        };
    }
}

By the way, these functions appear to have different deadlock propensity, although both of them should be used with care.

4

That's not an issue in my use case. The Receiver is not mpmc, so once the target channel closes, all bets are off anyway, since I don't intend to return the Receiver from the function.