Introduction
Rust’s ownership and borrowing system is a cornerstone of its safety guarantees, but it can present a challenge when you need to react to changes in your data. Imagine building a UI, a game, or any application where a change in one piece of data needs to automatically update another. This is where the concept of ‘observable properties’ comes in. In this article, we’ll explore how you can create your own observable properties in Rust, allowing you to build reactive and maintainable applications without compromising on Rust’s core principles of safety and performance. We’ll dive into the patterns and techniques that make this possible, giving you the tools to write more expressive and responsive code.
In this article we will build a simple Person struct with some observable properties, show how they can be used and implement some tests.
Implementation in Rust
The prerequisites
We will start by importing some prerequisites:
use std::collections::HashMap;
use std::panic;
use std::sync::{Arc, RwLock};
use std::fmt;
use std::thread;
use std::time::Duration;Line by line:
- Our observable properties will store the observers in a
HashMapwhere they key is the observer id, and the value is the observer function itself. - Next we bring in two concurrency tools, the first one is
Arc, which stands for ‘Atomically Reference Counted`. This is a smart pointer and it enables multiple owners of the same resource. It allows for a resource to be safely shared between threads. The second one isRwLockwhich allows multiple readers or one writer at a time, but never both. - In the final import we bring the
panicmodule. This handles Rust’s panic mechanism. A panic is triggered when your program encounters an unrecoverable error. - The
std::threadandstd::time::Durationare needed for demonstration purpose in themain()function later on.
Defining the errors
We will be working in a multi-threaded environment, so some errors can occurs, for example a write lock could not be acquired. That is why we start by defining some errors:
#[derive(Debug, Clone)]
pub enum PropertyError {
ReadLockError { context: String },
WriteLockError { context: String },
ObserverNotFound { id: usize },
PoisonedLock,
ObserverError { reason: String },
}Next we implement the Display trait for the PropertyError. This makes the errors a bit more userfriendly when printed out:
impl fmt::Display for PropertyError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
PropertyError::ReadLockError { context } => {
write!(f, "Failed to acquire read lock: {}", context)
}
PropertyError::WriteLockError { context } => {
write!(f, "Failed to acquire write lock: {}", context)
}
PropertyError::ObserverNotFound { id } => {
write!(f, "Observer with ID {} not found", id)
}
PropertyError::PoisonedLock => {
write!(f, "Property is in a poisoned state due to a panic in another thread")
}
PropertyError::ObserverError { reason } => {
write!(f, "Observer execution failed: {}", reason)
}
}
}
}What happens here is that we match the error and print out an appropiate error message.
To integrate the errors into Rust’s error eco system, we must also implement the Error trait:
impl std::error::Error for PropertyError {}The Observer type
An observer in our implementation is a function wrapped in an Arc which makes it possible to share references to the same observer function. If the same observer is registered with several properties or when the same observer needs to be called from different threads, the Arc ensure the observer remains valid and accessible until all references are dropped.
The function itself starts with dyn Fn(&T,&T) which specifies that observers are dynamically dispatched function objects. The function takes two parameters, the first is the old value of the property, the second parameter represents the new value of the property. Dynamic dispatch means the specific implementation of the function is determined at runtime rather than at compiletime. Doing this allows various types of functions , closure or complex function objects, as long of course they match the expected signature.
The Send + Sync ensure observers can be safely used in multithreaded environments. To be precies the Send trait indicates ownership of the observer can be transferred between threads. We need this when observers are registered from one thread, but called from a different thread.
The Sync trait ensures the observers can be safely accessed concurrently from multiple threads. This is important as an observer might be triggered simultaneously by different property changes. These trait bounds work in together with the Arc wrapper, to deliver a completely thread safe observer system.
We also define an ObserverId type as usize. This is a simple yet effective way to uniquely identify each registered observer.
The code looks like this:
type Observer<T> = Arc<dyn Fn(&T, &T) + Send + Sync>;
type ObserverId = usize;The ObservableProperty and the InnerProperty structs
Now we can define the ObservableProperty struct:
pub struct ObservableProperty<T> {
inner: Arc<RwLock<InnerProperty<T>>>,
}The ObservableProperty represents the public interface for an observable property. It servers as a thread safe wrapper around a shared mutable state using the Arc<RwLock<...>> pattern. The struct is generic so that it can work with any type meeting cerating bounds. This makes it a flexible foundation for building reactive systems where changes to resources can be automatically be observed, and if needed acted upon.
The InnerProperty looks like this:
struct InnerProperty<T> {
value: T,
observers: HashMap<ObserverId, Observer<T>>,
next_id: ObserverId,
}This is the type where the actual data and state management happens. Line by line:
- The
valuefields holds the actual data being observed. - The
observersHashMap maintains a collection of observers, i.e. callback functions which will be notified once the value changes. - Finally the
next_idfield serves a counter to generate unique identifiers for each observer registration.
The ObservableProperty implementation
We will start by setting some type constraints:
impl<T: Clone + Send + Sync + 'static> ObservableProperty<T> {
}What does this mean?
- The
Clonetrait values can be copied when needed. - Likewise
SendandSyncguarantee the type can be transferred between threads and accessed concurrently. - Finally the
'staticlifetime bound ensures any references to the type live for the entire program duration.
Why these constraints? They are crucial for the concurrent nature of the observable pattern, i.e. multiple threads might need to access or even modify the property simultaneously.
The constructor
Next we will start simple by creating the constructor:
pub fn new(initial_value: T) -> Self {
Self {
inner: Arc::new(RwLock::new(InnerProperty {
value: initial_value,
observers: HashMap::new(),
next_id: 0,
})),
}
}Line by line:
- The constructor receives the initial value of the property as its parameter.
- Next it initialize the
ObservableProperty. - Create a new
InnerPropertywith the initial value, and empty hashmap and the next id set to 0 - Wrap this in an
Arcand newRwLock. - Finally return the newly created
ObservableProperty.
The get() method
Next we will the implement the get() method.
pub fn get(&self) -> Result<T, PropertyError> {
self.inner
.read()
.map(|prop| prop.value.clone())
.map_err(|_| PropertyError::PoisonedLock)
}Line by line:
- The method returns a
Result<T,PropertyError>instead of a pureT, since it is possible a read lock can not be acquired, in which case an error should be returned. - We use method chaining to handle lock acquisitiion and value extraction.
- First we try and acquire a read lock. If this is not possible. We use an
RwLockwhich will only block if another thread currently holds a write lock, but allows multiple readers. - If the lock acquisition was successfull we transform this into the actual value by cloning it. Why do we need to clone it? Because returning a direct reference would be violating Rust’s borrowing rules, as the reference would become invalid as soon as the lock is released.
- The call to
map_err()is used to handle an error. In Rust, a lock becomes poisoned when the thread holding it panics. It becomes poisoned to prevent other threads from accessing possibly corrupted data. In this method, rather than propagating the lowlevel, we return aPropertyError::PoisonedLockerror. The|_|indicates we discard the specific details of the poisoning error.
The set() method
Now we also need a method to set the value:
pub fn set(&self, new_value: T) -> Result<(), PropertyError> {
let (old_value, observers_snapshot) = {
let mut prop = self.inner.write()
.map_err(|_| PropertyError::WriteLockError {
context: "setting property value".to_string()
})?;
let old_value = prop.value.clone();
prop.value = new_value.clone();
let observers_snapshot: Vec<Observer<T>> = prop.observers.values().cloned().collect();
(old_value, observers_snapshot)
};
for observer in observers_snapshot {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_value, &new_value);
})) {
eprintln!("Observer panic: {:?}", e);
}
}
Ok(())
}Line by line:
- The method has one parameter, the new value of the property.
- Next we create a scope where we do several things: the first is two acquire a write lock, and handle any errors. Next we retrieve and clone the old value. More importantly we also set the
valuefield to the new value, or at least to a clone of the new value. We also make a clone of the observers, which creates an independent snapshot of the current observers hashmap. Finally we return the old value and the observers snapshot. Why do we need to do this in a separate scope? This has to do with the fact that once we start notifying the observers the write lock must be released to prevent dead locks. - Now we iterate over the observers snapshot. Each observer is called with
panic::catch_unwindwrapper. In case the observer panics, it will return an error, if everything went Ok it will return the closure’s result. - When all observers have been called, and no errors or panics occurred during this process we can now return
Ok())indicating no errors.
Subscribing and unsubscribing
For a property to have observers, we must be able to subscribe an observer to the property:
pub fn subscribe(&self, observer: Observer<T>) -> Result<ObserverId, PropertyError> {
let mut prop = self.inner.write()
.map_err(|_| PropertyError::WriteLockError {
context: "subscribing observer".to_string()
})?;
let id = prop.next_id;
prop.next_id += 1;
prop.observers.insert(id, observer);
Ok(id)
}Line by line:
- The method receives an observer parameter, which is of type
Observer<T>. - The return type is the id of the newly subscribe observer, or in case something went wrong a
PropertyError. - Next we calculate the id of the new observer and insert it into the hashmap.
- Finally we return the newly created id.
Why do we need this id? Well, this makes it easier to unsubscribe observers:
pub fn unsubscribe(&self, id: ObserverId) -> Result<bool, PropertyError> {
let mut prop = self.inner.write()
.map_err(|_| PropertyError::WriteLockError {
context: "unsubscribing observer".to_string()
})?;
let was_present = prop.observers.remove(&id).is_some();
Ok(was_present)
}This more or less follows the same pattern as the subscribe() method:
- The method receives the id of the observer.
- The return type is a
Result<bool,PropertyError>. The bool indicates success, i.e. the method was able to successfully remove an existing observer. In case the method for example fails to acquire a read lock, aPropertyErroris returned. - The method starts by acquiring a write lock on the property. If this fails the method returns an error.
- Now we can remove the observer. Note that the
remove()method returns anOption<>. In case of a successfull this method returnsSome<Observer<T>>, in case the observer with the specified id was not found, aNoneis returned. Theis_some()method returns true if theSomevariant is returned, and false otherwise. - Finally we return an
Ok()variant containing the result of theremove()call.
For completeness, here is the code ObservableProperty implementation:
impl<T: Clone + Send + Sync + 'static> ObservableProperty<T> {
pub fn new(initial_value: T) -> Self {
Self {
inner: Arc::new(RwLock::new(InnerProperty {
value: initial_value,
observers: HashMap::new(),
next_id: 0,
})),
}
}
pub fn get(&self) -> Result<T, PropertyError> {
self.inner
.read()
.map(|prop| prop.value.clone())
.map_err(|_| PropertyError::PoisonedLock)
}
pub fn set(&self, new_value: T) -> Result<(), PropertyError> {
let (old_value, observers_snapshot) = {
let mut prop = self.inner.write()
.map_err(|_| PropertyError::WriteLockError {
context: "setting property value".to_string()
})?;
let old_value = prop.value.clone();
prop.value = new_value.clone();
let observers_snapshot: Vec<Observer<T>> = prop.observers.values().cloned().collect();
(old_value, observers_snapshot)
};
for observer in observers_snapshot {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_value, &new_value);
})) {
eprintln!("Observer panic: {:?}", e);
}
}
Ok(())
}
pub fn subscribe(&self, observer: Observer<T>) -> Result<ObserverId, PropertyError> {
let mut prop = self.inner.write()
.map_err(|_| PropertyError::WriteLockError {
context: "subscribing observer".to_string()
})?;
let id = prop.next_id;
prop.next_id += 1;
prop.observers.insert(id, observer);
Ok(id)
}
pub fn unsubscribe(&self, id: ObserverId) -> Result<bool, PropertyError> {
let mut prop = self.inner.write()
.map_err(|_| PropertyError::WriteLockError {
context: "unsubscribing observer".to_string()
})?;
let was_present = prop.observers.remove(&id).is_some();
Ok(was_present)
}
}All we need to do now is implement the Clone trait for the ObservableProperty<T>:
impl<T: Clone> Clone for ObservableProperty<T> {
fn clone(&self) -> Self {
Self {
inner: Arc::clone(&self.inner),
}
}
}The Person struct and implementation
Now we apply what we just made by building a simple person struct:
pub struct Person {
pub name: ObservableProperty<String>,
pub age: ObservableProperty<i32>,
}Where we would normally use a String or an i32 we now wrap these in ObservableProperty<> wrappers.
The constructor now looks like this:
pub fn new(name: String, age: i32) -> Self {
Person {
name: ObservableProperty::new(name),
age: ObservableProperty::new(age),
}
}And we need some convenience methods:
pub fn set_name(&self, name: String) -> Result<(), PropertyError> {
self.name.set(name)
}
pub fn get_name(&self) -> Result<String, PropertyError> {
self.name.get()
}
pub fn set_age(&self, age: i32) -> Result<(), PropertyError> {
self.age.set(age)
}
pub fn get_age(&self) -> Result<i32, PropertyError> {
self.age.get()
}These are wrappers for the set() and get() methods of the ObservableProperty<> struct.
The whole implementation of the Person struct looks like this:
impl Person {
pub fn new(name: String, age: i32) -> Self {
Person {
name: ObservableProperty::new(name),
age: ObservableProperty::new(age),
}
}
pub fn set_name(&self, name: String) -> Result<(), PropertyError> {
self.name.set(name)
}
pub fn get_name(&self) -> Result<String, PropertyError> {
self.name.get()
}
pub fn set_age(&self, age: i32) -> Result<(), PropertyError> {
self.age.set(age)
}
pub fn get_age(&self) -> Result<i32, PropertyError> {
self.age.get()
}
}Testing time
Now we can test our little setup by creating a small test program where we can show that this pattern works both in a synchronous and asynchronous environment.
fn main() {
println!("=== Observable Properties Example ===");
println!("Type-safe Person binding properties example");
let person = Person::new("Alice".to_string(), 30);
let name_id1 = person
.name
.subscribe(Arc::new(|old_value, new_value| {
println!(
"Name Observer 1: Name changed from '{}' to '{}'",
old_value, new_value
);
}))
.unwrap_or_else(|e| {
eprintln!("Failed to subscribe to name: {}", e);
0
});
let age_id1 = person
.age
.subscribe(Arc::new(|old_value, new_value| {
println!(
"Age Observer: Age changed from {} to {}",
old_value, new_value
);
}))
.unwrap_or_else(|e| {
eprintln!("Failed to subscribe to age: {}", e);
0
});
println!(
"Initial state - Name: {}, Age: {}",
person.get_name().unwrap_or_else(|e| {
eprintln!("Error getting name: {}", e);
"<error>".to_string()
}),
person.get_age().unwrap_or_else(|e| {
eprintln!("Error getting age: {}", e);
-1
})
);
println!("Setting name to 'Bob'...");
if let Err(e) = person.set_name("Bob".to_string()) {
eprintln!("Error setting name: {}", e);
}
println!("Setting age to 35...");
if let Err(e) = person.set_age(35) {
eprintln!("Error setting age: {}", e);
}
let _ = person.name.unsubscribe(name_id1);
let _ = person.age.unsubscribe(age_id1);
println!("Demonstrating thread-safe observable properties");
let person = Arc::new(Person::new("Alice".to_string(), 30));
let observer_id = person.age.subscribe(Arc::new(|old, new| {
println!("Observer: Age changed {} -> {} [Thread: {:?}]",
old, new, thread::current().id());
})).unwrap();
println!("Observer subscribed with ID: {}", observer_id);
let person1 = person.clone();
let person2 = person.clone();
let handle1 = thread::spawn(move || {
for i in 31..34 {
println!("Thread 1: Setting age to {}", i);
if let Err(e) = person1.set_age(i) {
eprintln!("Thread 1 error: {}", e);
}
thread::sleep(Duration::from_millis(100));
}
});
let handle2 = thread::spawn(move || {
for _ in 0..3 {
match person2.get_age() {
Ok(age) => println!("Thread 2: Current age is {}", age),
Err(e) => eprintln!("Thread 2 error: {}", e),
}
thread::sleep(Duration::from_millis(150));
}
});
handle1.join().unwrap();
handle2.join().unwrap();
let _ = person.age.unsubscribe(observer_id);
println!("Multi-threading example completed successfully!");
}A short overview:
- We start by creating a
Personstruct, of a person named ‘Alice’ aged 30. - Next we subscribe an observer which simply prints out the old value and the value of the person’s name.
- We do the same for the age of the person.
- Now we print the initial state of the person.
- Time to test the observers by setting both name and age to different values.
- Finally we can remove the subscribers from the two properties.
- Next we can show how to modify and read properties from different threads.
- We start by creating a
Personbut now it is wrapped in anArcso it can safely be referenced from different threads. - Now we create two clones of this person.
- The first thread we create sets the age to different values.
- The next thread reads the age and prints it out.
- We can now wait for the two threads to complete.
- Unsubscribe the observer, and print a success message.
Conclusion
In this article, we’ve built a robust and thread-safe observable properties system in Rust, demonstrating how to create reactive data structures that automatically notify observers of changes. By leveraging core Rust concepts like Arc for shared ownership and RwLock for concurrent access, we were able to design a system that is both safe and performant.
Our implementation showcased how to define custom error types to provide clear, actionable feedback to the user and how to use traits like Clone, Send and Sync to ensure our code is compatible with Rust’s concurrency model. We wrapped our core logic in a simple Person struct, illustrating how this pattern can be applied to build more expressive and maintainable applications.
The final examples, both synchronous and multi-threaded, proved the effectiveness and safety of our approach. By catching panics in observers and carefully managing lock lifecycles, we built a system that is resilient to errors and prevents deadlocks. This pattern of observable properties is a powerful tool for building reactive user interfaces, game logic, or any application where data changes need to trigger a cascade of updates, all while upholding the safety guarantees that make Rust so appealing.
In a next article I will implement several enhancements even further by doing the following:
- Improving the error handling.
- Having a
set_async()method or something similar so the observer functions can run asynchronously. - A subscribe filter function, which means that you also provide a filter which when true allows the observer to run.
- Add unit tests.
[…] a previous article I wrote about observable properties in Rust. However, some things were […]