async_observable_os_threads/
async_observable_os_threads.rs

1//! This `Observable` emits values and completes. It returns an empty `Subscription`,
2//! making it unable to be unsubscribed from. Some operators like `take`, `switch_map`,
3//! `merge_map`, `concat_map`, and `exhaust_map` require unsubscribe functionality to
4//! work correctly.
5//!
6//! This asynchronous Observable utilizes an OS thread, preventing it from blocking
7//! the current thread.
8//!
9//! To run this example, execute `cargo run --example async_observable_os_threads`.
10
11use std::time::Duration;
12
13use rxr::{
14    subscribe::{Subscriber, Subscription, SubscriptionHandle, UnsubscribeLogic},
15    Observable, ObservableExt, Observer, Subscribeable,
16};
17
18fn main() {
19    // Create a custom observable that emits values in a separate thread.
20    let observable = Observable::new(|mut o| {
21        // Launch a new thread for the Observable's processing and store its handle.
22        let join_handle = std::thread::spawn(move || {
23            for i in 0..=15 {
24                // Emit the value to the subscriber.
25                o.next(i);
26                // Important. Put an await point after each emit or after some emits.
27                // This allows the `take()` operator to function properly.
28                // Not required in this example.
29                std::thread::sleep(Duration::from_millis(1));
30            }
31            // Signal completion to the subscriber.
32            o.complete();
33        });
34
35        // Return the subscription.
36        Subscription::new(
37            // In this example, we omit the unsubscribe functionality. Without it, we
38            // can't unsubscribe, which prevents the `take()` operator, as well as
39            // higher-order operators like `switch_map`, `merge_map`, `concat_map`,
40            // and `exhaust_map`, from functioning as expected.
41            UnsubscribeLogic::Nil,
42            // Store the `JoinHandle` to enable waiting functionality using the
43            // `Subscription` for this Observable thread to complete.
44            SubscriptionHandle::JoinThread(join_handle),
45        )
46    });
47
48    // Create the `Subscriber` with a mandatory `next` function, and optional
49    // `complete` function. No need for `error` function in this simple example.
50    let mut observer = Subscriber::on_next(|v| println!("Emitted {}", v));
51    observer.on_complete(|| println!("Completed"));
52
53    // This observable uses OS threads so it will not block the current thread.
54    // Observables are cold so if you comment out the statement bellow nothing
55    // will be emitted.
56    let subscription = observable
57        .filter(|&v| v <= 10)
58        .map(|v| format!("Mapped {}", v))
59        .subscribe(observer);
60
61    // Do something else here.
62    println!("Print something while Observable is emitting.");
63
64    // Because the subscription creates a new thread, we can utilize the `Subscription`
65    // to wait for its completion. This ensures that the main thread won't terminate
66    // prematurely and stop all child threads.
67    if subscription.join().is_err() {
68        // Handle error
69    }
70
71    println!("Custom Observable finished emmiting")
72}
async_observable_os_threads/async_observable_os_threads.rs

async_observable_os_threads/
async_observable_os_threads.rs
