Async-std: Optionally merging TCPStreams

[matthewrobertbell]

Hello,

I have been writing code that is based on this example: https://github.com/async-rs/async-std/blob/master/examples/tcp-ipv4-and-6-echo.rs

I want to have something like this:

    //Runtime config bools
    let four = true;
    let six = false;

    let mut incoming = if four {
        let ipv4_listener =
            TcpListener::bind(format!("127.0.0.1:{}", local_peer_info.port)).await?;
        println!("Listening on {}", ipv4_listener.local_addr()?);
        let ipv4_incoming = ipv4_listener.incoming();
        if six {
            let ipv6_listener =
                TcpListener::bind(format!("[::1]:{}", local_peer_info.port)).await?;
            println!("Listening on {}", ipv6_listener.local_addr()?);
            let ipv6_incoming = ipv6_listener.incoming();
            Some(ipv4_incoming.merge(ipv6_incoming))
        } else {
            Some(ipv4_incoming)
        }
    } else if six {
        let ipv6_listener = TcpListener::bind(format!("[::1]:{}", local_peer_info.port)).await?;
        println!("Listening on {}", ipv6_listener.local_addr()?);
        Some(ipv6_listener.incoming())
    } else {
        None
    };

    if let Some(incoming) = incoming {
        while let Some(stream) = incoming.next().await {

The logic here is that the application can listen on IPV4, IPV6, both, or neither, and be useful with any of those combinations. I am running into the listener having a different type to merged listeners. I note that both implement Stream, but I couldn't figure out a workaround.

The error is:

514 | /             if six {
515 | |                 let ipv6_listener = TcpListener::bind(format!("[::1]:{}", local_peer_info.port)).await?;
516 | |                 println!("Listening on {}", ipv6_listener.local_addr()?);
517 | |                 let ipv6_incoming = ipv6_listener.incoming();
518 | |                 Some(ipv4_incoming.merge(ipv6_incoming))
    | |                 ---------------------------------------- expected because of this
519 | |             } else {
520 | |                 Some(ipv4_incoming)
    | |                 ^^^^^^^^^^^^^^^^^^^ expected struct `async_std::stream::stream::merge::Merge`, found struct `async_std::net::tcp::listener::Incoming`
521 | |             }
    | |_____________- if and else have incompatible types

Is this a problem with my code, or something that could be changed in async-std?

Thanks!

[yoshuawuyts]

@matthewrobertbell Rust wants types in all arms to always be the same. One type is Merge, the other is TcpStream, which aren't the same.

In this case what you could do is merge with a stream::Empty in the other branch to make both types the same. Or use type erasure using boxes, but that seems less efficient.

Hope this is helpful!

[matthewrobertbell]

@yoshuawuyts thank you for your response. I had already tried using empty, but I gave it another shot, and got the same error:

527 | / if six { 528 | | let ipv6_listener = 529 | | TcpListener::bind(format!("[::1]:{}", local_peer_info.port)).await?; 530 | | println!("Listening on {}", ipv6_listener.local_addr()?); 531 | | let ipv6_incoming = ipv6_listener.incoming(); 532 | | Some(ipv4_incoming.merge(ipv6_incoming)) | | ---------------------------------------- expected because of this 533 | | } else { 534 | | Some(ipv4_incoming.merge(stream::empty())) | | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ expected struct `async_std::net::tcp::listener::Incoming`, found struct `async_std::stream::empty::Empty` 535 | | } | |_________- if and else have incompatible types

I'm not sure why the returned types are different, when in all arms they are now merged streams?

[k-nasa]

@matthewrobertbell

I think that is because the generic type is different even for the same Merge type.

// ipv4_incoming.merge(ipv6_incoming)
async_std::stream::stream::merge::Merge<async_std::net::tcp::listener::Incoming<'_>, async_std::net::tcp::listener::Incoming<'_>>

// ipv4_incoming.merge(stream::empty())
async_std::stream::stream::merge::Merge<async_std::net::tcp::listener::Incoming<'_>, async_std::stream::empty::Empty<std::result::Result<async_std::net::tcp::stream::TcpStream, std::io::Error>>>>

Hmm. It's not a good way, but what about using trays and objects ❓

[yoshuawuyts]

Or alternatively: invert the application logic, and define a new function with the application logic:

async fn run(tpc_stream: impl BufRead) -> io::Result<()> {
    // core logic goes here; works on both variants
}

_note: I'm unsure if it should be impl BufRead or dyn BufRead; might need to play around with it_

[ghost]

@matthewrobertbell You can do the following (I've just tried it and it works):

//Runtime config bools
let four = true;
let six = true;

let ipv4_listener;
let ipv6_listener;

let mut incoming: Option<Box<dyn Stream<Item = io::Result<TcpStream>> + Unpin>> = if four {
    ipv4_listener =
        TcpListener::bind(format!("127.0.0.1:{}", local_peer_info.port)).await?;
    println!("Listening on {}", ipv4_listener.local_addr()?);
    let ipv4_incoming = ipv4_listener.incoming();
    if six {
        ipv6_listener =
            TcpListener::bind(format!("[::1]:{}", local_peer_info.port)).await?;
        println!("Listening on {}", ipv6_listener.local_addr()?);
        let ipv6_incoming = ipv6_listener.incoming();
        Some(Box::new(ipv4_incoming.merge(ipv6_incoming)))
    } else {
        Some(Box::new(ipv4_incoming))
    }
} else if six {
    ipv6_listener = TcpListener::bind(format!("[::1]:{}", local_peer_info.port)).await?;
    println!("Listening on {}", ipv6_listener.local_addr()?);
    Some(Box::new(ipv6_listener.incoming()))
} else {
    None
};

[matthewrobertbell]

Thanks to both answers, I have now solved this. I defined a new function with this signature:

async fn process_incoming_stream(
    mut incoming: impl stream::Stream<Item = io::Result<TcpStream>> + Unpin,
    internal_channel_sender: Sender<FromPeerInternalMessage>,
    local_peer_info: PeerInfo,
) -> io::Result<()> {

and this calling code:

    //Runtime config bools
    let four = true;
    let six = false;

    if four {
        let ipv4_listener =
            TcpListener::bind(format!("127.0.0.1:{}", local_peer_info.port)).await?;
        println!("Listening on {}", ipv4_listener.local_addr()?);
        let ipv4_incoming = ipv4_listener.incoming();
        if six {
            let ipv6_listener =
                TcpListener::bind(format!("[::1]:{}", local_peer_info.port)).await?;
            println!("Listening on {}", ipv6_listener.local_addr()?);
            let ipv6_incoming = ipv6_listener.incoming();
            process_incoming_stream(
                ipv4_incoming.merge(ipv6_incoming),
                internal_channel_sender,
                local_peer_info,
            )
            .await?;
        } else {
            process_incoming_stream(ipv4_incoming, internal_channel_sender, local_peer_info)
                .await?;
        }
    } else if six {
        let ipv6_listener = TcpListener::bind(format!("[::1]:{}", local_peer_info.port)).await?;
        println!("Listening on {}", ipv6_listener.local_addr()?);
        let ipv6_incoming = ipv6_listener.incoming();
        process_incoming_stream(ipv6_incoming, internal_channel_sender, local_peer_info).await?;
    }

Cheers!

[k-nasa]

It ’s solved, so close it.