How to get a notification, or callback, when data arrives in a socket?

1

In Boost Asio it's normal to register a callback for when data arrives in a socket using async_read which receives a buffer and a callback. When the buffer is full the callback is executed.

How do I do the same thing using Tokio for a long-lived connection?

2

As far as I can tell from the documentation the equivalent code with Tokio is:

use tokio::io::AsyncReadExt;

tokio::spawn(async move {
    let mut buf = [0; 256];
    reader.read_exact(&mut buf).await?;
    // the rest of the code here
});

You can look at the AsyncReadExt docs for other useful methods on an AsyncRead.

1 Like
3

Won't read_exact block on a long lived connection even after reading all data?

4

.await doesn't block threads. It only suspends execution of code until the data arrives. It behaves exactly like turning rest of that async block into a callback (behind the scenes it is a callback from a Waker given by read_exact to runtime that executes that async block).

1 Like
5

So this would always be called when data arrives in the socket or only once?

6

spawn runs code once. read_exact waits until it got exactly as many bytes as required, even if that took multiple syscalls.

7

I know I've accepted a solution but just so I don't need to open a new thread, is there a way to always be notified whenever data arrives in the socket, assuming the connection will never be closed?

8 
loop {
   let bytes_read = socket.read(&mut buf).await?;
}
9

How would I make this live for the same amount of time as a struct? Can I turn it into a property?

10

Async code can be stored as an object — a Future, but it doesn't do anything while it's sitting in a struct. It needs to be spawned to run.

You can tie it to an object using channels:

let (sender, receiver) = tokio::sync::mpsc::channel(1);
// add your own error handling
spawn(async move {
   loop {
      let bytes_read = socket.read(&mut buf).await?;
      // it'll fail when receiver is dropped
      sender.send(buf[0..bytes_read].to_vec())?; 
   }
});

// store receiver in a struct, and later you can do:
somestruct.receiver.recv().await;

but that's a bit overcomplicated, because usually you'd store socket in a struct, and just read from it in async fn.

Callback-based design is rather different than async/future-based design, so I suspect you'll need to adjust your design to fit async code, instead of trying to make async functions work like callbacks.

11

This topic was automatically closed 90 days after the last reply. We invite you to open a new topic if you have further questions or comments.