Async futures, yielding and polling for values

[kforce]

Trying to create a single-threaded app with 2 async tasks

a 'simulation' task that will yield a stream of int's at various intervals.

a 'calc' task that will poll for new simulated values every loop (say every 100mS), and maintain an aged value result, where the last yielded simulation value is reduced by 1 for each time there is no new value.

The numbers/durations are not really important, its the async, yield, poll bit I'm stuck on. Tried futues-rs, async-std, futures-async-stream

I'm new here, but from javascript land & a seasoned user of evented runtimes, Promises API, async/await, Any code examples would be amazing!. 12hrs of googling & watching videos & no further forward

[alice]

Sounds like you should be spawning two Tokio tasks, and using tokio::sync::mpsc or tokio::sync::watch to send over the ints from one task to another. You use tokio::time::sleep or tokio::time::interval to go to sleep for some duration.

You shouldn't manually "poll" for new values. Just .await the mpsc channel and you will be woken up immediately when a message is sent.

You may find the Tokio tutorial useful.

[alice]

Here's an example from Tokio's docs on the mpsc channel:

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(100);

    tokio::spawn(async move {
        tx.send("hello").await.unwrap();
    });

    assert_eq!(Some("hello"), rx.recv().await);
    assert_eq!(None, rx.recv().await);
}

It shows off how to send and receive messages. The None return value is special and does not mean that there is no message, rather it means that all senders have gone out of scope, and there will never be more messages.

[kforce]

Thank you for the guidance alice, I havn't looked at tokio! really useful! I didn't think I can use await because I still need to process a calc even if there is no value from simulation, so i cannot just block calc

[alice]

You can use tokio::time::timeout if you want to receive with a timeout.

tokio::time::timeout(duration, rx.recv()).await;

Or you can use tokio::select!, which does similar things but is more powerful.