Tokio

https://tokio.rs/tokio/tutorial
Admin
Tokio is an asynchronous runtime for Rust that provides the building blocks for writing fast,...
Tokens:3,383
Snippets:41
Trust Score:9.8
Update:6 months ago
Show doc for...
Tokens:
Raw
### Start Mini-Redis Server

Source: https://tokio.rs/tokio/tutorial/streams

Before running Tokio stream examples, ensure the Mini-Redis server is running.

```bash
$ mini-redis-server

```

--------------------------------

### Basic Stream Processing with Tokio

Source: https://tokio.rs/tokio/tutorial/streams

This example shows initial stream processing where messages are received from a channel.

```rust
got = Ok(Message { channel: "numbers", content: b"1" })
got = Ok(Message { channel: "numbers", content: b"two" })
got = Ok(Message { channel: "numbers", content: b"3" })
got = Ok(Message { channel: "numbers", content: b"four" })
got = Ok(Message { channel: "numbers", content: b"five" })
got = Ok(Message { channel: "numbers", content: b"6" })

```

--------------------------------

### Limit Stream Messages with `take`

Source: https://tokio.rs/tokio/tutorial/streams

Use the `take` adapter to limit a stream to at most a specified number of messages. This example stops after receiving three messages.

```rust
let messages = subscriber
    .into_stream()
    .take(3);

```

--------------------------------

### Transform Stream Messages with `map`

Source: https://tokio.rs/tokio/tutorial/streams

Use the `map` adapter to transform the content of each message. This example extracts the content after filtering and unwrapping the message.

```rust
let messages = subscriber
    .into_stream()
    .filter(|msg| match msg {
        Ok(msg) if msg.content.len() == 1 => true,
        _ => false,
    })
    .map(|msg| msg.unwrap().content)
    .take(3);

```

--------------------------------

### Unpinned Stream Error Example

Source: https://tokio.rs/tokio/tutorial/streams

Illustrates the compiler error when attempting to iterate over a stream that has not been pinned. This highlights the necessity of `tokio::pin!` for stream processing.

```rust
error[E0277]: `from_generator::GenFuture<[static generator@Subscriber::into_stream::{closure#0} for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6> {ResumeTy, &'r mut Subscriber, Subscriber, impl Future, (), std::result::Result<Option<Message>, Box<(dyn std::error::Error + Send + Sync + 't0)>>, Box<(dyn std::error::Error + Send + Sync + 't1)>, &'t2 mut async_stream::yielder::Sender<std::result::Result<Message, Box<(dyn std::error::Error + Send + Sync + 't3)>>>, async_stream::yielder::Sender<std::result::Result<Message, Box<(dyn std::error::Error + Send + Sync + 't4)>>>, std::result::Result<Message, Box<(dyn std::error::Error + Send + Sync + 't5)>>, impl Future, Option<Message>, Message}]>` cannot be unpinned
  --> streams/src/main.rs:29:36
   |
29 |     while let Some(msg) = messages.next().await {
   |                                    ^^^^ within `tokio_stream::filter::_::__Origin<'_, impl Stream, [closure@streams/src/main.rs:22:17: 25:10]>`, the trait `Unpin` is not implemented for `from_generator::GenFuture<[static generator@Subscriber::into_stream::{closure#0} for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6> {ResumeTy, &'r mut Subscriber, Subscriber, impl Future, (), std::result::Result<Option<Message>, Box<(dyn std::error::Error + Send + Sync + 't0)>>, Box<(dyn std::error::Error + Send + Sync + 't1)>, &'t2 mut async_stream::yielder::Sender<std::result::Result<Message, Box<(dyn std::error::Error + Send + Sync + 't3)>>>, async_stream::yielder::Sender<std::result::Result<Message, Box<(dyn std::error::Error + Send + Sync + 't4)>>>, std::result::Result<Message, Box<(dyn std::error::Error + Send + Sync + 't5)>>, impl Future, Option<Message>, Message}]>`
   |
   = note: required because it appears within the type `impl Future`
   = note: required because it appears within the type `async_stream::async_stream::AsyncStream<std::result::Result<Message, Box<(dyn std::error::Error + Send + Sync + 'static)>>, impl Future>`
   = note: required because it appears within the type `impl Stream`
   = note: required because it appears within the type `tokio_stream::filter::_::__Origin<'_, impl Stream, [closure@streams/src/main.rs:22:17: 25:10]>`
   = note: required because of the requirements on the impl of `Unpin` for `tokio_stream::filter::Filter<impl Stream, [closure@streams/src/main.rs:22:17: 25:10]>`
   = note: required because it appears within the type `tokio_stream::map::_::__Origin<'_, tokio_stream::filter::Filter<impl Stream, [closure@streams/src/main.rs:22:17: 25:10]>, [closure@streams/src/main.rs:26:14: 26:40]>`
   = note: required because of the requirements on the impl of `Unpin` for `tokio_stream::map::Map<tokio_stream::filter::Filter<impl Stream, [closure@streams/src/main.rs:22:17: 25:10]>, [closure@streams/src/main.rs:26:14: 26:40]>`
```

--------------------------------

### Filter Stream Messages with `filter`

Source: https://tokio.rs/tokio/tutorial/streams

Use the `filter` adapter to process only messages that satisfy a given condition. This example filters for messages with single-digit content length.

```rust
let messages = subscriber
    .into_stream()
    .filter(|msg| match msg {
        Ok(msg) if msg.content.len() == 1 => true,
        _ => false,
    })
    .take(3);

```

--------------------------------

### Configure tokio-stream dependency

Source: https://tokio.rs/tokio/tutorial/streams

Add this entry to your Cargo.toml to include the tokio-stream crate.

```toml
tokio-stream = "0.1"
```

--------------------------------

### Implement a stream using async-stream

Source: https://tokio.rs/tokio/tutorial/streams

Using the stream! macro to define a stream with async/await syntax.

```rust
use async_stream::stream;
use std::time::{Duration, Instant};

stream! {
    let mut when = Instant::now();
    for _ in 0..3 {
        let delay = Delay { when };
        delay.await;
        yield ();
        when += Duration::from_millis(10);
    }
}
```

--------------------------------

### Main function for Mini-Redis Broadcast

Source: https://tokio.rs/tokio/tutorial/streams

Spawns a task to publish messages and then subscribes to the channel on the main task. This demonstrates concurrent message handling.

```rust
use tokio_stream::StreamExt;
use mini_redis::client;

#[tokio::main]
async fn main() -> mini_redis::Result<()> {
    tokio::spawn(async {
        publish().await
    });

    subscribe().await?;

    println!("DONE");

    Ok(())
}
```

--------------------------------

### Implement a custom Interval stream

Source: https://tokio.rs/tokio/tutorial/streams

A manual implementation of a stream that yields a unit value three times at 10ms intervals.

```rust
use tokio_stream::Stream;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

struct Interval {
    rem: usize,
    delay: Delay,
}

impl Interval {
    fn new() -> Self {
        Self {
            rem: 3,
            delay: Delay { when: Instant::now() }
        }
    }
}

impl Stream for Interval {
    type Item = ();

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<Option<()>>
    {
        if self.rem == 0 {
            // No more delays
            return Poll::Ready(None);
        }

        match Pin::new(&mut self.delay).poll(cx) {
            Poll::Ready(_) => {
                let when = self.delay.when + Duration::from_millis(10);
                self.delay = Delay { when };
                self.rem -= 1;
                Poll::Ready(Some(()))
            }
            Poll::Pending => Poll::Pending,
        }
    }
}
```

--------------------------------

### Iterate over a stream

Source: https://tokio.rs/tokio/tutorial/streams

Use a while let loop with StreamExt::next() to process stream values asynchronously.

```rust
use tokio_stream::StreamExt;

#[tokio::main]
async fn main() {
    let mut stream = tokio_stream::iter(&[1, 2, 3]);

    while let Some(v) = stream.next().await {
        println!("GOT = {:?}", v);
    }
}
```

--------------------------------

### Subscribe to Mini-Redis Channel

Source: https://tokio.rs/tokio/tutorial/streams

Subscribes to the 'numbers' channel and prints received messages. The stream of messages must be pinned using `tokio::pin!` before iteration.

```rust
use tokio_stream::StreamExt;
use mini_redis::client;

async fn subscribe() -> mini_redis::Result<()> {
    let client = client::connect("127.0.0.1:6379").await?;
    let subscriber = client.subscribe(vec!["numbers".to_string()]).await?;
    let messages = subscriber.into_stream();

    tokio::pin!(messages);

    while let Some(msg) = messages.next().await {
        println!("got = {:?}", msg);
    }

    Ok(())
}
```

--------------------------------

### Publish Messages to Mini-Redis

Source: https://tokio.rs/tokio/tutorial/streams

Publishes a series of string messages to the 'numbers' channel using the Mini-Redis client. Ensure a Mini-Redis server is running on the default port.

```rust
use tokio_stream::StreamExt;
use mini_redis::client;

async fn publish() -> mini_redis::Result<()> {
    let mut client = client::connect("127.0.0.1:6379").await?;

    // Publish some data
    client.publish("numbers", "1".into()).await?;
    client.publish("numbers", "two".into()).await?;
    client.publish("numbers", "3".into()).await?;
    client.publish("numbers", "four".into()).await?;
    client.publish("numbers", "five".into()).await?;
    client.publish("numbers", "6".into()).await?;
    Ok(())
}
```

--------------------------------

### Define the Stream trait

Source: https://tokio.rs/tokio/tutorial/streams

The Stream trait definition, which mirrors the Future trait but allows repeated polling for multiple values.

```rust
use std::pin::Pin;
use std::task::{Context, Poll};

pub trait Stream {
    type Item;

    fn poll_next(
        self: Pin<&mut Self>, 
        cx: &mut Context<'_>
    ) -> Poll<Option<Self::Item>>;

    fn size_hint(&self) -> (usize, Option<usize>) {
        (0, None)
    }
}
```
Tokens:
Raw