스트림: 순서대로의 Future
스트림: 순서대로의 Future (Streams: Futures in Sequence)
이번 장 앞부분의 "메시지 전달" 절에서 async 채널의 수신자(receiver)를 어떻게 사용했는지 떠올려 보세요. async recv 메서드는 시간에 걸쳐 일련의 항목을 만들어 냅니다. 이것은 스트림(stream) 이라고 알려진 훨씬 더 일반적인 패턴의 한 사례이에요. 많은 개념이 자연스럽게 스트림으로 표현됩니다. 큐에서 항목이 사용 가능해지거나, 전체 데이터 집합이 컴퓨터 메모리에 담기에는 너무 클 때 파일시스템에서 데이터 덩어리를 점진적으로 끌어오거나, 네트워크를 통해 시간에 걸쳐 데이터가 도착하는 경우 등이요. 스트림은 future이므로 다른 어떤 종류의 future와도 함께 쓸 수 있고 흥미로운 방식으로 결합할 수 있어요. 예를 들어 이벤트를 배치 처리해 너무 많은 네트워크 호출을 유발하지 않게 하거나, 오래 실행되는 연산의 시퀀스에 타임아웃을 설정하거나, 불필요한 작업을 피하도록 사용자 인터페이스 이벤트를 조절(throttle)할 수 있습니다.
일련의 항목을 본 것은 13장에서 Iterator 트레이트를 다룰 때였어요. "Iterator 트레이트와 next 메서드" 절에서요. 하지만 iterator와 async 채널 수신자 사이에는 두 가지 차이가 있습니다. 첫 번째 차이는 시간이에요. iterator는 동기적이고 채널 수신자는 비동기적이죠. 두 번째 차이는 API입니다. Iterator를 직접 다룰 때는 동기 next 메서드를 호출해요. 특히 trpl::Receiver 스트림에서는 대신 비동기 recv 메서드를 호출했죠. 그 외에는 이 두 API가 매우 비슷하게 느껴지는데, 그 유사성은 우연이 아니에요. 스트림은 일종의 비동기적 형태의 순회(iteration)입니다. trpl::Receiver가 구체적으로 메시지 수신을 기다리는 반면, 일반적인 목적의 스트림 API는 훨씬 더 넓어요. Iterator가 다음 항목을 제공하는 것처럼 하지만 비동기적으로 해 주는 거죠.
Rust에서 iterator와 스트림의 유사성 덕분에 우리는 실제로 어떤 iterator에서든 스트림을 만들 수 있어요. iterator와 마찬가지로, 스트림의 next 메서드를 호출하고 그 출력을 await해서 스트림을 다룰 수 있습니다. Listing 17-21처럼요. 이건 아직 컴파일되지 않아요.
extern crate trpl; // required for mdbook test
fn main() {
trpl::block_on(async {
let values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
let iter = values.iter().map(|n| n * 2);
let mut stream = trpl::stream_from_iter(iter);
while let Some(value) = stream.next().await {
println!("The value was: {value}");
}
});
}
숫자 배열로 시작해 그것을 iterator로 변환하고 map을 호출해 모든 값을 두 배로 만들었어요. 그런 다음 trpl::stream_from_iter 함수로 iterator를 스트림으로 변환합니다. 다음으로 while let 루프로 스트림의 항목들이 도착할 때마다 순회합니다.
안타깝게도 코드를 실행하려 하면 컴파일되지 않고, 대신 next 메서드가 없다고 알려줘요:
error[E0599]: no method named `next` found for struct `tokio_stream::iter::Iter` in the current scope
--> src/main.rs:10:40
|
10 | while let Some(value) = stream.next().await {
| ^^^^
|
= help: items from traits can only be used if the trait is in scope
help: the following traits which provide `next` are implemented but not in scope; perhaps you want to import one of them
|
1 + use crate::trpl::StreamExt;
|
1 + use futures_util::stream::stream::StreamExt;
|
1 + use std::iter::Iterator;
|
1 + use std::str::pattern::Searcher;
|
help: there is a method `try_next` with a similar name
|
10 | while let Some(value) = stream.try_next().await {
| ~~~~~~~~
이 출력이 설명하듯이 컴파일러 오류의 이유는 next 메서드를 사용하려면 올바른 트레이트가 스코프 안에 있어야 하기 때문이에요. 지금까지의 논의를 고려하면 그 트레이트가 Stream일 거라고 합리적으로 기대할 수 있는데, 실제로는 StreamExt입니다. ext(ension) 의 줄임말인 Ext는 한 트레이트를 다른 트레이트로 확장하는 Rust 커뮤니티의 흔한 패턴이에요.
Stream 트레이트는 효과적으로 Iterator와 Future 트레이트를 결합한 저수준 인터페이스를 정의합니다. StreamExt는 Stream 위에 next 메서드와 Iterator 트레이트가 제공하는 것과 비슷한 다른 유틸리티 메서드를 포함한 더 고수준의 API 집합을 제공해요. Stream과 StreamExt는 아직 Rust 표준 라이브러리 일부가 아니지만, 대부분의 생태계 크레이트가 비슷한 정의를 사용합니다.
컴파일러 오류의 해결책은 Listing 17-22처럼 trpl::StreamExt에 대한 use 문을 추가하는 거예요.
extern crate trpl; // required for mdbook test
use trpl::StreamExt;
fn main() {
trpl::block_on(async {
let values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
// --snip--
let iter = values.iter().map(|n| n * 2);
let mut stream = trpl::stream_from_iter(iter);
while let Some(value) = stream.next().await {
println!("The value was: {value}");
}
});
}
이 모든 조각을 조합하면 코드가 우리가 원하는 대로 동작해요! 게다가 이제 StreamExt가 스코프 안에 있으니, iterator에서처럼 그 유틸리티 메서드들을 모두 사용할 수 있습니다.