|
|
extern crate trpl; // required for mdbook test
|
|
|
|
|
|
use std::{pin::pin, thread, time::Duration};
|
|
|
|
|
|
use trpl::{ReceiverStream, Stream, StreamExt};
|
|
|
|
|
|
fn main() {
|
|
|
trpl::run(async {
|
|
|
let messages = get_messages().timeout(Duration::from_millis(200));
|
|
|
let intervals = get_intervals()
|
|
|
.map(|count| format!("Interval #{count}"))
|
|
|
.throttle(Duration::from_millis(500))
|
|
|
.timeout(Duration::from_secs(10));
|
|
|
let merged = messages.merge(intervals).take(20);
|
|
|
let mut stream = pin!(merged);
|
|
|
|
|
|
while let Some(result) = stream.next().await {
|
|
|
match result {
|
|
|
Ok(item) => println!("{item}"),
|
|
|
Err(reason) => eprintln!("Problem: {reason:?}"),
|
|
|
}
|
|
|
}
|
|
|
});
|
|
|
}
|
|
|
|
|
|
fn get_messages() -> impl Stream<Item = String> {
|
|
|
let (tx, rx) = trpl::channel();
|
|
|
|
|
|
trpl::spawn_task(async move {
|
|
|
let messages = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"];
|
|
|
|
|
|
for (index, message) in messages.into_iter().enumerate() {
|
|
|
let time_to_sleep = if index % 2 == 0 { 100 } else { 300 };
|
|
|
trpl::sleep(Duration::from_millis(time_to_sleep)).await;
|
|
|
|
|
|
if let Err(send_error) = tx.send(format!("Message: '{message}'")) {
|
|
|
eprintln!("Cannot send message '{message}': {send_error}");
|
|
|
break;
|
|
|
}
|
|
|
}
|
|
|
});
|
|
|
|
|
|
ReceiverStream::new(rx)
|
|
|
}
|
|
|
|
|
|
// ANCHOR: threads
|
|
|
fn get_intervals() -> impl Stream<Item = u32> {
|
|
|
let (tx, rx) = trpl::channel();
|
|
|
|
|
|
// 这里 *不是* `trpl::spawn`，是 `std::thread::spawn`！
|
|
|
thread::spawn(move || {
|
|
|
let mut count = 0;
|
|
|
loop {
|
|
|
// 同样，这里 *不是* `trpl::sleep`，是 `std::thread::sleep`！
|
|
|
thread::sleep(Duration::from_millis(1));
|
|
|
count += 1;
|
|
|
|
|
|
if let Err(send_error) = tx.send(count) {
|
|
|
eprintln!("Could not send interval {count}: {send_error}");
|
|
|
break;
|
|
|
};
|
|
|
}
|
|
|
});
|
|
|
|
|
|
ReceiverStream::new(rx)
|
|
|
}
|
|
|
// ANCHOR_END: threads
|
