Optional future for optional connections?

I want to have a loop with a tokio::select! in it, and I want to be able to timeout a connection that hasn't been used in a while. Something along the line of:

async fn sometask() {
  loop {
    tokio::select!(
      frm = frame.next() => {
        // received a frame from the network.
      }
      msg = rx.recv() => {
        // received a message from the application.
        // if connection is closed, then establish it.
        // send message over network connection.
      }
      timeout => {
        // close connection if it hasn't been used for `x` seconds
      }
      killswitch => {
        break;
      }
    )
  }
}

I would prefer not to make two different loops (depending on whether the connection is established or not). Is there some kind of Future that can act sort of like an Option, but it will always return a Pending state when it is None, and forward the poll to the inner value if there is one? (The idea being that both the frame and the timeout could be wrapped in such a Future within the tokio::select!, allowing the same logic for all other arms regardless of whether the connection is established or not).

You can make your own:

use futures::{Stream, StreamExt};

async fn next_if_some<S>(s: &mut Option<S>) -> Option<S::Item>
where
    S: Stream + Unpin
{
    match s.as_mut() {
        Some(stream) => stream.next().await,
        None => std::future::pending().await,
    }
}
2 Likes