scan_operator/
scan_operator.rs

1//! Demonstrating the `scan()` operator
2//!
3//! This example illustrates the functionality of the `scan()` operator, which
4//! applies an accumulator function to each element emitted by an observable,
5//! producing a new accumulated result on each emission.
6//!
7//! The `scan()` operator is useful for maintaining state across emissions and
8//! performing cumulative transformations on the observable stream. It takes a
9//! closure that defines the accumulation logic and an optional initial accumulator
10//! value. The result is a new observable emitting the accumulated values.
11//!
12//! To run this example, execute `cargo run --example scan_operator`.
13
14use std::sync::{Arc, Mutex};
15
16use rxr::{
17    subscribe::{Subscriber, Subscription, SubscriptionHandle, UnsubscribeLogic},
18    Observable, ObservableExt, Observer, Subscribeable,
19};
20use tokio::{sync::mpsc::channel, task, time};
21
22const UNSUBSCRIBE_SIGNAL: bool = true;
23
24fn get_response_observable() -> Observable<&'static str> {
25    Observable::new(|mut o| {
26        task::spawn(async move {
27            o.next("response");
28            time::sleep(time::Duration::from_millis(1)).await;
29            o.complete();
30        });
31        Subscription::new(UnsubscribeLogic::Nil, SubscriptionHandle::Nil)
32    })
33}
34
35#[tokio::main]
36async fn main() {
37    let observable = Observable::new(|mut o| {
38        let done = Arc::new(Mutex::new(false));
39        let done_c = Arc::clone(&done);
40        let (tx, mut rx) = channel(10);
41
42        task::spawn(async move {
43            if let Some(UNSUBSCRIBE_SIGNAL) = rx.recv().await {
44                *done_c.lock().unwrap() = UNSUBSCRIBE_SIGNAL;
45            }
46        });
47
48        let join_handle = task::spawn(async move {
49            for i in 0..100 {
50                if *done.lock().unwrap() == UNSUBSCRIBE_SIGNAL {
51                    break;
52                }
53                o.next(i);
54                time::sleep(time::Duration::from_millis(1)).await;
55            }
56            o.complete();
57        });
58
59        Subscription::new(
60            UnsubscribeLogic::Future(Box::pin(async move {
61                if tx.send(UNSUBSCRIBE_SIGNAL).await.is_err() {
62                    println!("Receiver dropped.");
63                }
64            })),
65            SubscriptionHandle::JoinTask(join_handle),
66        )
67    });
68
69    let mut observer = Subscriber::on_next(|v| println!("Emitted: {}", v));
70    observer.on_complete(|| println!("Completed"));
71
72    // Accumulate response strings into a single string.
73    // The types of `total` and `n` may differ as long as `total` implements the `From<n>` trait.
74    // In this example, `total` is of type `String`, and `n` is of type `&str`.
75    let subscription = observable
76        .take(6)
77        .delay(100)
78        .merge_map(|_| get_response_observable())
79        .scan(|total, n| format!("{} {}", total, n), None)
80        .subscribe(observer);
81
82    // Do something else here.
83    println!("Do something while Observable is emitting.");
84
85    // Wait for observable to finish before exiting the program.
86    if subscription.join_concurrent().await.is_err() {
87        // Handle error
88    }
89
90    println!("`main` function done")
91}
scan_operator/scan_operator.rs

scan_operator/
scan_operator.rs
