Bem-vindo à aula sobre Streams em Rust! Nesta aula, vamos mergulhar no mundo da programação assíncrona, focando em como processar sequências de valores que chegam de forma assíncrona. Se você já está familiarizado com iteradores em Rust, os streams são o equivalente assíncrono: eles permitem que você trabalhe com uma sequência de itens que podem não estar todos disponíveis imediatamente, como dados vindos de uma rede, arquivos grandes ou eventos de usuário.

O trait Stream é a base para essa abstração, e junto com adaptadores e consumidores, você pode construir pipelines de processamento de dados de forma elegante e eficiente. Vamos começar entendendo o trait em si e depois ver como usá-lo na prática.

trait Stream

O trait Stream é definido na biblioteca padrão do Rust (a partir da versão 1.80, estabilizado) e é fundamental para a programação assíncrona. Ele é análogo ao trait Iterator, mas em vez de retornar itens de forma síncrona, ele retorna um Future que resolve para Option<Self::Item>. A definição básica é:

pub trait Stream {
    type Item;

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

O método poll_next é chamado quando o executor está pronto para avançar o stream. Ele retorna Poll::Ready(Some(item)) se um item está disponível, Poll::Ready(None) se o stream terminou, e Poll::Pending se não há nada pronto no momento. Para usar streams de forma ergonômica, normalmente usamos a trait StreamExt da crate futures, que fornece métodos como next(), map(), filter(), e muitos outros. O trait StreamExt é como o IteratorExt para iteradores.

Vamos ver um exemplo simples de implementação manual de um stream que gera números incrementais:

use std::pin::Pin;
use std::task::{Context, Poll};
use futures::Stream;

struct Counter {
    count: u32,
}

impl Stream for Counter {
    type Item = u32;

    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<u32>> {
        self.count += 1;
        if self.count < 10 {
            Poll::Ready(Some(self.count))
        } else {
            Poll::Ready(None)
        }
    }
}

No entanto, na prática, raramente implementamos Stream diretamente; usamos geradores como async_stream ou combinamos streams existentes.

Adaptadores

Adaptadores são métodos que transformam um stream em outro stream, permitindo composição. Eles são definidos na trait StreamExt. Alguns dos mais comuns são:

  • map: aplica uma função a cada item.
  • filter: mantém apenas itens que satisfazem um predicado.
  • take: limita o número de itens.
  • skip: pula os primeiros n itens.
  • chain: concatena dois streams.
  • zip: combina dois streams em pares.

Vamos ver um exemplo usando map e filter:

use futures::stream::{self, StreamExt};

#[tokio::main]
async fn main() {
    let stream = stream::iter(1..=10);
    let doubled = stream
        .map(|x| x * 2)
        .filter(|x| x % 3 == 0);

    let result: Vec<_> = doubled.collect().await;
    println!("{:?}", result); // [6, 12, 18]
}

Note que stream::iter cria um stream a partir de um iterador síncrono. Os adaptadores são lazy: eles não são executados até que o stream seja consumido.

Outro adaptador útil é then, que permite executar uma função assíncrona em cada item:

let stream = stream::iter(1..=3).then(|n| async move {
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    n * 10
});

Consumindo

Para obter os itens de um stream, precisamos consumi-lo. Os métodos consumidores também estão em StreamExt. Os mais comuns são:

  • next(): retorna o próximo item como um Future.
  • collect(): coleta todos os itens em uma coleção.
  • fold(): acumula um valor ao longo dos itens.
  • for_each(): executa uma função em cada item.
  • count(): conta o número de itens.

O método next() é frequentemente usado em loops while let:

use futures::stream::{self, StreamExt};

#[tokio::main]
async fn main() {
    let mut stream = stream::iter(1..=5);

    while let Some(item) = stream.next().await {
        println!("Got: {}", item);
    }
}

Já collect() pode coletar em qualquer tipo que implemente FromStream, como Vec, HashMap, etc.

let items: Vec<i32> = stream::iter(1..=5).map(|x| x * 2).collect().await;
assert_eq!(items, vec![2, 4, 6, 8, 10]);

Para fold, veja um exemplo de soma:

let sum = stream::iter(1..=10).fold(0, |acc, x| async move { acc + x }).await;
assert_eq!(sum, 55);

Casos de uso

Streams são extremamente úteis em aplicações assíncronas. Alguns casos comuns:

  • Processamento de dados em tempo real: como ler de um socket ou de um canal e processar mensagens conforme chegam.
  • Paginação de APIs: consumir endpoints paginados de forma incremental, sem carregar tudo na memória.
  • Operações de I/O: ler arquivos grandes linha por linha, ou processar saída de subprocessos.
  • Eventos de interface: em GUIs assíncronas, lidar com eventos do usuário como cliques ou teclas.

Exemplo prático: consumir um canal assíncrono (tokio::sync::mpsc) como um stream.

use tokio::sync::mpsc;
use futures::stream::StreamExt;

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel::<u32>(10);

    tokio::spawn(async move {
        for i in 0..5 {
            tx.send(i).await.unwrap();
        }
    });

    // rx é um Receiver que implementa Stream
    let mut rx = rx;
    while let Some(item) = rx.next().await {
        println!("Received: {}", item);
    }
}

Outro caso é usar stream::unfold para gerar uma sequência infinita, como números de Fibonacci:

use futures::stream::{self, StreamExt};

let fib = stream::unfold((0u32, 1u32), |(a, b)| async move {
    Some((a, (b, a + b)))
});
let first_five: Vec<_> = fib.take(5).collect().await;
assert_eq!(first_five, vec![0, 1, 1, 2, 3]);

Boas práticas e observações finais

Ao trabalhar com streams, lembre-se de:

  • Use futures::stream::StreamExt para ter acesso aos métodos de conveniência.
  • Evite implementar Stream manualmente a menos que necessário; prefira combinadores.
  • Streams são lazy, portanto certifique-se de consumi-los para que o trabalho seja feito.
  • Para streams infinitos, sempre use take ou take_while para evitar loops infinitos.
  • Considere usar async-stream crate para gerar streams com sintaxe mais simples.

Referências

Exercícios

  1. Escreva uma função que retorne um stream dos primeiros 10 números pares (começando de 2). Use stream::iter e adaptadores.

    2. ✓ Resposta: 
    use futures::stream::{self, StreamExt};

fn even_stream() -> impl Stream {
    stream::iter(2..).filter(|x| x % 2 == 0).take(10)
}
  2. Dado um stream de números, use fold para calcular o produto de todos os números de 1 a 5.

    3. ✓ Resposta: 
    use futures::stream::{self, StreamExt};

#[tokio::main]
async fn main() {
    let product = stream::iter(1..=5).fold(1, |acc, x| async move { acc * x }).await;
    println!("{}", product); // 120
}
  3. Implemente um stream personalizado que gere a sequência de Fibonacci indefinidamente usando stream::unfold.

    4. ✓ Resposta: 
    use futures::stream::{self, Stream};

fn fibonacci() -> impl Stream {
    stream::unfold((0u128, 1u128), |(a, b)| async move {
        Some((a, (b, a + b)))
    })
}
  4. Use o crate futures para criar um stream a partir de um canal tokio::sync::mpsc e consuma 3 itens.

    5. ✓ Resposta: 
    use tokio::sync::mpsc;
use futures::stream::StreamExt;

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel::(10);
    tokio::spawn(async move {
        for i in 0..5 {
            tx.send(i).await.unwrap();
        }
    });

    let mut rx = rx;
    for _ in 0..3 {
        if let Some(item) = rx.next().await {
            println!("Got {}", item);
        }
    }
}
  5. Crie um stream que emita números de 1 a 10, mas apenas os que são múltiplos de 3, e colete em um Vec.

    6. ✓ Resposta: 
    use futures::stream::{self, StreamExt};

#[tokio::main]
async fn main() {
    let multiples: Vec<_> = stream::iter(1..=10)
        .filter(|x| x % 3 == 0)
        .collect()
        .await;
    assert_eq!(multiples, vec![3, 6, 9]);
}