Notify Me, Please: An Async-Powered Observer for Rust and Tokio

Introduction

Writing concurrent code in Rust can be a real challenge. You’re constantly thinking about shared state, race conditions, and how to safely pass data between threads without the whole program crashing. But what if you need to know when a value changes, and you need that notification to be handled asynchronously in a multi-threaded environment? That’s a puzzle many of us face, and the standard library doesn’t always have a straightforward answer. You might find yourself building custom channels or wrapping your values in layers of mutexes and atomics, which quickly becomes complicated.

This is where the observer pattern comes in, a classic solution for broadcasting changes to interested parties. But simply implementing the pattern isn’t enough for the modern Rust ecosystem, which relies on asynchronous runtimes like Tokio. We need a solution that is both thread-safe and async-compatible, allowing for non-blocking notifications without sacrificing performance. This article introduces a new crate that tackles this exact problem: an observable property implementation built on Tokio. It’s designed to bring the elegance of the observer pattern to your async Rust projects, providing a simple, robust, and safe way to react to property changes. Let’s dive in and see how it works.

Implementation in Rust

Updating the dependencies

We will start by adding to crates to our Cargo.toml file:

[dependencies]
tokio = { version = "1.47.1", features = ["time"] }
rand = "0.9.2"
observable-property-tokio = "0.1.3"

Line by line:

1. We start by importing the tokio runtime, with the time feature enabled, as Tokio will provide the foundation of asynchronous programming in this application. We will use the time feature to let thread sleep for a certain amount of time.
2. The rand crate provides random number generation. This will be used to simulate stock price volatility.
3. Finally we have observable-property-tokio crate which enables reactive programming through observable properties.

Taken together, these dependencies create the foundation of an event-driven asynchronous stock price simulation with can send updates to its subscribers.

Starting with the prerequisites

In your main.rs add the following lines:

use std::sync::Arc;
use tokio::time::{sleep, Duration};
use observable_property_tokio::{ObservableProperty, PropertyError};
use rand::{Rng, SeedableRng, rngs::StdRng};

Line by line:

1. We stand by import the Arc from the sync crate. This struct provides shared ownership of resources across multiple threads.
2. The imports from the tokio asynchronous runtime. The example application uses sleep() and Duration to simulate real-time price updates and delayed logging.
3. The import from the observable_property_tokio crate provides the reactive programming foundation of the example application.
4. Finally we have the imports from the rand crate. These random number generators power the price simulation algorithm.

The Stock struct

In our very simplified example a stock looks like this:

#[derive(Debug)]
struct Stock {
    symbol: String,
    price: ObservableProperty<f64>,
}

A Stock has two properties: the symbol like ‘MSFT’ and the current price. The current price is wrapped in an ObservableProperty. This enables the subscription of interested observers, which will be called when the price changes.

Testing time

Now we can start testing our stock simulation. Start by making sure your main() function looks like this:

#[tokio::main]
async fn main() -> Result<(), PropertyError> {
}

Some explanation may be needed here. To start with the #[tokio::main] attribute macro transform this function behind the scenes. It creates a runtime environment enable the execution of the asynchronous code.

The async keyword indicates that the function contains asynchronous operations. Throughout the function body, as we shall see, we will find ..await expressions that yield control back to the runtime while waiting for operations to complete. This prevents blocking behaviour.

Finally the return type Result<(),PropertyError> shows that:

1. The function returns a Result type, enabling error handling.
2. On success it returns (), which is a unit type, similar to void in other languages.
3. If something goes wrong, the function returns a PropertyError.

This signature enables elegant error handling which be seen in the body of the function by the use of the ? operator.

Next we can create our first stock:

    let stock = Stock::new("OBSP", 100.0);

    println!("--- Initializing Stock Ticker ---");
    println!("Starting price for {}: €{:.2}", stock.symbol, stock.price.get()?);
    println!("---------------------------------");

This piece of code simply creates a new stock and prints out its initial state.

Adding the dashboard subscriber

Time to add our first subscriber, which is could be a called the dashboard subscriber as it prints out any changes to the price:

    let symbol_dashboard = stock.symbol.clone();
    stock.price.subscribe(Arc::new(move |_old_price, new_price| {
        println!("\n[Dashboard] Price Updated for {}: €{:.2}", symbol_dashboard, new_price);
    }))?;

Line by line:

1. We clone the stock symbol, This is needed because the value needs to be captured and moved into the closure. Not cloning would move the original stock.symbol which would make it unavailable for other operations.
2. Next call the subscribe() method. The method is provided with a closure which will be called any time the price changes.
3. Also note the ? at the end. This means that any error while subscribing to the property will be propagated to the calling function i.e. main().

Adding a filtered subscriber

Sometimes it is not necessary to call the closure anytime a property changes, for example if the new value does not differ from the old value, or in this case only if the price drops below a certain threshold namely € 95.00:

    let symbol_alert = stock.symbol.clone();
    stock.price.subscribe_filtered(
        Arc::new(move |_old_price, new_price| {
            println!("\n*** [ALERT!] Price for {} has dropped to €{:.2} ***", symbol_alert, new_price);
        }),
        |_old_price, new_price| *new_price < 95.0,
    )?;

Adding an asynchronous subscriber

Next we will set up an asynchronous call back that trigger whenever the stock price changes, but this time we will include a deliberate delay:

    stock.price.subscribe_async(|old_price, new_price| async move {
        sleep(Duration::from_millis(500)).await;
        println!("[Log] Price change from €{:.2} to €{:.2} completed.", old_price, new_price);
    })?;

Unlike the regular subscribe() method which we saw earlier, subscribe_async() accepts an asynchronous closure, indicated by the async keyword. This allows for the use the await keyword insisde the body of the closure.

This closure, or callback if you will, receives both the previous price and the new price as its parameters. When it is triggered it starts by introducing a 500-millisecond delay using the sleep() function before executing the logging statement.

Running the stock ticker simulation

Now we can run the simulation:

    let price_arc = Arc::new(stock.price);
    let price_for_task = Arc::clone(&price_arc);
    let symbol_task = stock.symbol.clone();


    let data_feed_task = tokio::spawn(async move {
        let price = price_for_task;

        let mut rng = StdRng::seed_from_u64(rand::random());
        for _ in 0..10 {
            let current_price = price.get().unwrap_or_else(|e| {
                eprintln!("Error getting current price: {}", e);
                100.0 // Default value in case of error
            });

            let volatility = 0.05;

            let change_percent = rng.random_range(-volatility..volatility);
            let change_amount = current_price * change_percent;
            let noise = rng.random_range(-1.0..1.0);
            let new_price = (current_price + change_amount + noise).max(0.0);


            if rng.random_bool(0.5) {
                println!("\n[Data Feed] Updating price for {} synchronously...", symbol_task);
                if let Err(e) = price.set(new_price) {
                    eprintln!("Error setting value: {}", e);
                }
            } else {
                println!("\n[Data Feed] Updating price for {} asynchronously...", symbol_task);
                if let Err(e) = price.set_async(new_price).await {
                    eprintln!("Error setting value asynchronously: {}", e);
                }
            }

            sleep(Duration::from_millis(1000)).await;
        }
    });


    data_feed_task.await.expect("Data feed task panicked.");

    println!("\n--- Ticker Simulation Complete ---");
    Ok(())

Let’s break this down: we start by wrapping the stock price property in an Arc, which allows for thread-safe sharing. Next we create of this reference and the stock symbol. We need to do this because these values will be move into an asynchronous task.

Next we create the asynchronous task which runs concurrently with the main execution flow. The move keyword moves the ownership of the captured variables into the closure.

In the closure we loop ten times. In the loop we:

1. Get the current price using the ObservableProperty‘s get() method. Note that we unwrap_or_else to perform any error handling.
2. Next we calculate a new price, using a set volatility and some noise.
3. Finally we use the random_bool() method to decide whether to set the new price synchronously or asynchronously. The random_bool() method receives a probability of it returning true. We are passing 0.5 here so there is a 50% chance it returns true and a 50% chance it returns false.
4. After which the thread sleeps for a second before entering the next iteration.

Finally we wait for the task to complete using the await keyword. If something went wrong inside the thread, the call to expect() will report on that.

Lastly we print out a status message and return an Ok(()) status.

Conclusion

Rust’s approach to concurrency, while powerful, often presents developers with a complex landscape of shared state and synchronization challenges. The observer pattern offers a clean way to manage state changes, but adapting it for asynchronous, multi-threaded Rust applications requires a robust, non-blocking solution. The observable-property-tokio crate addresses this by providing a simple, safe, and thread-safe implementation built on the Tokio runtime. It allows you to create observable properties that notify subscribers of changes, whether those subscribers are regular closures, filtered callbacks, or even other asynchronous tasks. As demonstrated with the stock ticker simulation, this crate simplifies the development of event-driven architectures in Rust, enabling you to build responsive applications where different components can react to data changes without the overhead of manual synchronization primitives. By abstracting away the complexities of thread communication, it helps developers focus on the core logic, making concurrent and asynchronous programming in Rust more accessible and maintainable.

If you want to more, check out the crates homepage, or if you are interested in the source code, or want to contribute, please visit its github repository.

Any feedback or compliments is highly appreciated.