CODEWITHSUNDEEP
asynchronousrustcallbackfuture
goertzenator
goertzenator

Reputation: 2039

How to use Rust futures in callbacks?

Is there any way to use futures in callbacks? For example...

// Send message on multiple channels while removing ones that are closed.

use smol::channel::Sender;

...

// (expecting bool, found opaque type)
vec_of_sender.retain( |sender| async {
  sender.send(msg.clone()).await.is_ok()
});

My work-around is to loop twice: On the first pass I delete closed senders (non-async) and on the second I do the actual send (async using for sender in ...). But it seems like I should be able to do it all in a single retain() call.

Upvotes: 2

Views: 2052

Answers (3)

Lambda Fairy
Lambda Fairy

Reputation: 14676

To avoid creating a new Vec, you can also use swap_remove:

let mut i = 0usize;
while i < vec_of_sender.len() {
    if vec_of_sender[i].send(msg.clone()).await.is_ok() {
        i += 1;
    } else {
        vec_of_sender.swap_remove(i);
    }
}

Note that this will change the order of the vector.

Upvotes: 0

Ibraheem Ahmed
Ibraheem Ahmed

Reputation: 13518

The closure passed to retain must return a bool, but every async function returns impl Future. Instead, you can use Stream, which is the asynchronous version of Iterator. You can convert the vector into a Stream:

let stream = stream::iter(vec_of_sender);

And then use the filter method, which accepts an asynchronous closure and returns a new Stream:

let vec_of_sender = stream.filter(|sender| async {
    sender.send(msg.clone()).await.is_ok()
}).collect::<Vec<Sender>>();

Upvotes: 1

Peter Hall
Peter Hall

Reputation: 58735

You can't use retain in this way. The closure that retain accepts must implement FnMut(&T) -> bool, but every async function returns an implementation of Future.

You can turn an async function into a synchronous one by blocking on it. For example, if you were using tokio, you could do this:

use tokio::runtime::Runtime;

let rt = Runtime::new().unwrap();

vec_of_sender.retain(|sender| {
    rt.block_on(async { sender.send().await.is_ok() })
});

However, there is overhead to adding an async runtime, and I have a feeling that you are trying to solve the wrong problem.

Upvotes: 1

Related Questions