OnceCell and mpsc

[Ricardicus]

Hello!

I want some help.

I need to understand why this does not work, and how I could possible fix it..

Basically, I want one blocking function relying on mpsc that I can export and use.
So I put an mpsc in a static variable, placed in a OnceCell, and one function does the
tx and another one does the blocking rx. For debugging I did a timeout on the rx.
When messages arrive, it seems like the rx side gets stuck. Why is that?

Here comes some code:

use std::sync::Arc;
use std::io::{self, Write};

use tokio::sync::{mpsc, Mutex, Notify, OnceCell, Semaphore};
use tokio::time::{timeout, Duration};

#[derive(Clone)]
pub struct WindowPipe<T> {
    pub tx: Arc<Mutex<mpsc::Sender<T>>>,
    pub rx: Arc<Mutex<mpsc::Receiver<T>>>,
}

impl<T> WindowPipe<T> {
    pub fn new() -> Self {
        let (tx, mut rx) = mpsc::channel(50);
        Self {
            tx: Arc::new(Mutex::new(tx)),
            rx: Arc::new(Mutex::new(rx)),
        }
    }

    pub fn clone(&self) -> Self {
        Self {
            tx: self.tx.clone(),
            rx: self.rx.clone(),
        }
    }

    pub async fn read(&self) -> Result<T, ()> {
        match self.rx.lock().await.recv().await {
            Some(msg) => Ok(msg),
            None => Err(()),
        }
    }

    pub async fn send(&self, cmd: T) {
        let _ = self.tx.lock().await.send(cmd).await;
    }
}

static PIPE: OnceCell<WindowPipe<String>> = OnceCell::const_new();

async fn initialize_global_values() {
    // Directly initialize the GLOBAL_VALUE using `init`
    let _ = PIPE.set(WindowPipe::<String>::new());
}

pub async fn read_terminal_input() -> Result<String, ()> {
    let timeout_duration = Duration::from_millis(200);
    match timeout(timeout_duration, PIPE.get().unwrap().read()).await {
        Ok(Ok(cmd)) => Ok(cmd),
        Ok(Err(_)) => Err(()),
        Err(_) => Err(()),
    }
}

async fn send_terminal_input(message: String) {
    let _ = PIPE.get().unwrap().send(message).await;
}

#[tokio::main]
async fn main() {
    // First task to initialize the global value
    let initializer = tokio::spawn(async {
        initialize_global_values().await;
    });

    // Wait for the initializer to complete
    initializer.await.unwrap();

    tokio::spawn(async move {
        loop {
            let mut input = String::new();

            print!("Please enter something: ");
            io::stdout().flush().unwrap(); // Make sure the prompt is printed before waiting for input

            io::stdin()
                .read_line(&mut input)
                .expect("Failed to read line");
            send_terminal_input(input).await;
        }
    });
    let l = tokio::spawn(async move {
        loop {
            println!("reading..");
            let r = read_terminal_input().await;
            if r.is_ok() {
                println!("you entered: {}", r.unwrap());
            }
        }
    });

    l.await;
}

The problem I experience is this..

The reading part gets stuck when there is a message. As an example, here I press "ENTER" after a while, and also add some messages that are supposed to be printed:

$ ./target/debug/test_tokio_once_cell_mpsc                          
Please enter something: reading..
reading..
reading..
reading..
reading..
reading..

Please enter something: hello
Please enter something: hello

The idea is to have these functions that I can call from outside of the module that are blocking until I get some input but I get stuck here.

Help would be much appreciated!

Thank you!

[Ricardicus]

If this worked, I think it could make my program more neat. I can get around it by changing my code and passing around the tx, and keeping the rx where it was created, but that means I have to pass this tx in a series of calls. And I just wonder why this approach doesn't work/isn't supported. I also wonder if there are alternative ways I can work to get to the same goal, maybe use different building blocks from tokio.