🎓 A1Learner
🔍

Rust Sinks: The Sending End

A Sink is the sending end of a pipe. A Sink takes items in, one send at a time.

send puts one item in

A futures channel gives you both ends at once. Its sender is a Sink. Its receiver is a Stream.

let (mut tx, mut rx) = mpsc::channel::<i32>(2);
tx.send(7).await.unwrap();
println!("got = {:?}", rx.next().await);

feed queues, flush pushes

feed waits for a slot and queues the item. It does not promise delivery yet. flush pushes out everything queued so far.

let (mut tx, mut rx) = mpsc::channel::<i32>(4);
tx.feed(1).await.unwrap();
tx.feed(2).await.unwrap();
tx.flush().await.unwrap();
println!("batch = {:?} {:?}", rx.next().await, rx.next().await);

forward pumps a whole stream in

forward connects a stream to a sink. It sends every item. It closes the sink when the stream ends. Plain items need an Ok wrap first, because forward works with Result streams.

let pump = stream::iter(vec![1, 2, 3]).map(Ok::<i32, mpsc::SendError>);
pump.forward(tx).await.unwrap();

A closed sink says no

Drop the receiver and the pipe has no far end. send now fails. The is_ok check shows the refusal without a panic.

let (mut tx, rx) = mpsc::channel::<i32>(2);
drop(rx);
println!("send works = {}", tx.send(1).await.is_ok());

Example

// Cargo.toml: futures = "0.3"
use futures::channel::mpsc;
use futures::executor::block_on;
use futures::stream::{self, StreamExt};
use futures::SinkExt;

async fn run() {
    let (mut tx, mut rx) = mpsc::channel::<String>(8);

    tx.send("direct".to_string()).await.unwrap();

    let words = stream::iter(vec!["one".to_string(), "two".to_string()])
        .map(Ok::<String, mpsc::SendError>);
    words.forward(&mut tx).await.unwrap();
    drop(tx);

    while let Some(word) = rx.next().await {
        println!("received: {word}");
    }
    println!("sink closed, stream drained");
}

fn main() {
    block_on(run());
}

Key points

  • Sink: the sending mirror of Stream — a stream yields items, a sink takes them
  • Use send(item) for one item: it queues the item and flushes it in one step
  • Use feed(item) to queue without flushing, then flush() to push the batch out
  • Use forward(sink) to pump a whole stream into a sink until the stream ends
  • A send to a closed sink fails with an error: the item has nowhere to go
  • tokio's mpsc Sender is not a futures Sink.
  • The Sink trait lives in the futures crate.

In this example, one sender feeds a receiver in two ways. First, send puts "direct" in by hand. Then forward pumps two words in from a stream. Each word wears an Ok wrap. forward works with Result streams, so plain items need one. drop ends the sending side for good. The receiver prints all three items in order. Its loop ends when the channel reports None.

Note

Its own channel sender implements it.

Try it Yourself

Edit the code, then press Run.

Running is simulated · Rust Playground

Exercise

What does forward do?