Introduction
In a previous article I wrote about observable properties in Rust. However, some things were missing:
- The
set()method executed the observers synchronously. Maybe we ought to have a method which does this asynchronously. - Sometimes it is handy to have some kind of filter to make sure the observer really needs to be notified. For example if a property does not change at all, it might be a good idea to not call the observer function.
In this article we will implement these two functionalities. For this, make sure to have the source code of the previous article ready, as we will be adding code to that.
The asynchronous set()
We will start by implementing the asynchronous set() function which surprisingly is called set_async():
pub fn set_async(&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)
};
if observers_snapshot.is_empty() {
return Ok(());
}
const MAX_THREADS: usize = 4;
let observers_per_thread = (observers_snapshot.len() + MAX_THREADS - 1) / MAX_THREADS;
for batch in observers_snapshot.chunks(observers_per_thread) {
let batch_observers = batch.to_vec();
let old_val = old_value.clone();
let new_val = new_value.clone();
thread::spawn(move || {
for observer in batch_observers {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_val, &new_val);
})) {
eprintln!("Observer panic in batch thread: {:?}", e);
}
}
});
}
Ok(())
}Add this code to the implementation block of the ObservableProperty<T> struct.
Line by line:
- Like the synchronous
set()method this method has one parameter, the new value of the property. - Next, like in the synchronous
set()method, 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 thevaluefield 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. - We next check if we have any observers, if there are none we can return immediately.
- Next calculate the observers per thread, since we will not be having a separate thread for each observer. We could of course do that, but that would be very resource intensive.
- Next we divide the observers slice into chunk of maximum size
observers_per_threadand iterate over it. - In the iterator’s body we first convert the batch to a vector, and clone the old and new values.
- Now we can spawn a thread.
- In the thread we have a loop iterating over the batch. Each observer is called with a
panic::catch_unwindwrapper. In case the observer panics, it will return an error, if everything went Ok it will return the closure’s result. - If no errors occurred we return an
Ok(()).
Adding methods to the Person struct
Now we need to add two simple methods to the Person struct, to call the new set_async() method:
pub fn set_name_async(&self, name: String) -> Result<(), PropertyError> {
self.name.set_async(name)
}
pub fn set_age_async(&self, age: i32) -> Result<(), PropertyError> {
self.age.set_async(age)
}Testing the async method
To test this new method, as these lines at the end of your main() function:
println!("Demonstrating non-blocking observers with background threads");
let person = Person::new("Alice".to_string(), 25);
let observer_id = match person
.name
.subscribe(Arc::new(|old_value, new_value| {
println!("Slow observer started: {} -> {}", old_value, new_value);
std::thread::sleep(std::time::Duration::from_millis(50));
println!("Slow observer finished: {} -> {}", old_value, new_value);
})) {
Ok(id) => {
println!("Successfully subscribed slow observer with ID: {}", id);
Some(id)
}
Err(e) => {
eprintln!("Failed to subscribe slow observer: {}", e);
None
}
};
println!("Setting name with blocking call...");
let start = std::time::Instant::now();
match person.set_name("Bob".to_string()) {
Ok(()) => {
println!("Blocking call took: {:?}", start.elapsed());
}
Err(e) => {
eprintln!("Failed to set name (blocking): {}", e);
}
}
println!("\nSetting name with non-blocking call...");
let start = std::time::Instant::now();
match person.set_name_async("Charlie".to_string()) {
Ok(()) => {
println!("Non-blocking call took: {:?}", start.elapsed());
}
Err(e) => {
eprintln!("Failed to set name (non-blocking): {}", e);
}
}
std::thread::sleep(std::time::Duration::from_millis(100));
if let Some(id) = observer_id {
match person.name.unsubscribe(id) {
Ok(true) => println!("Successfully unsubscribed observer"),
Ok(false) => println!("Observer was already unsubscribed"),
Err(e) => eprintln!("Failed to unsubscribe observer: {}", e),
}
}
println!("Background thread example completed successfully!");Line by line:
- In this additional code we start by creating a new person.
- Next we subcribe a closure to the name property of this struct. Note that we now handle the error properly.
- Now we can test the blocking, i.e. the synchronous call. Again, we handle any errors which might occur. If no errors occurs we print the time it took.
- Next we do the same, but try the new async functionality.
- To make sure all threads have finished we sleep for 100 milliseconds.
- Now to clean up we unsubscribe the observer. Because in the previous code due to the error handling the
observer_idis anOption<>we need to test for the existence of an id. If there is one we can call theunsubscribe()method, and handle any errors there. - Finally display a termination message.
Filtering observers
Sometimes an observer only needs to be triggered when a certain condition is met, for example in a UI framework an observer which updates an UI element needs only to be triggered when a property has really changed: a label of a button for example must only be updated if it has a different value.
Of course we could build this into the observer itself, however, it would be more elegant to have a filter wrapper around the observer like this:
pub fn subscribe_filtered<F>(&self, observer: Observer<T>, filter: F) -> Result<ObserverId, PropertyError>
where
F: Fn(&T, &T) -> bool + Send + Sync + 'static,
{
let filter = Arc::new(filter);
let filtered_observer = Arc::new(move |old_val: &T, new_val: &T| {
if filter(old_val, new_val) {
observer(old_val, new_val);
}
});
self.subscribe(filtered_observer)
}Add this code the implementation of ObservableProperty<>.
Line by line:
- The method signature is similar to the original
subscribe()method, however we added a filter function which returns a bool. - We start by wrapping the filter function in an
Arc. - Next we create a closure and wrap it in an
Arc. The closure has two parameters, the old value and the value. In the closure body we start by executing the filter function. Only if this function returns true, the observer is triggered. - Finally we call the original
subscribe()method with the newly created observer.
Testing the filters
Now we can test the filters, add this to the end of the main() function:
println!("Demonstrating filtered observers");
let person = Person::new("Alice".to_string(), 25);
let filtered_observer_id = match person
.age
.subscribe_filtered(
Arc::new(|old_val, new_val| {
println!("Age increased from {} to {}", old_val, new_val);
}),
|old_val, new_val| new_val > old_val,
) {
Ok(id) => {
println!("Successfully subscribed filtered observer with ID: {}", id);
Some(id)
}
Err(e) => {
eprintln!("Failed to subscribe filtered observer: {}", e);
None
}
};
println!("Setting age to 30 (should trigger filtered observer)...");
if let Err(e) = person.set_age(30) {
eprintln!("Failed to set age to 30: {}", e);
}
println!("Setting age to 25 (should NOT trigger filtered observer)...");
if let Err(e) = person.set_age(25) {
eprintln!("Failed to set age to 25: {}", e);
}
println!("Setting age to 35 (should trigger filtered observer again)...");
if let Err(e) = person.set_age(35) {
eprintln!("Failed to set age to 35: {}", e);
}
if let Some(id) = filtered_observer_id {
match person.age.unsubscribe(id) {
Ok(true) => println!("Successfully unsubscribed filtered observer"),
Ok(false) => println!("Filtered observer was already unsubscribed"),
Err(e) => eprintln!("Failed to unsubscribe filtered observer: {}", e),
}
}
println!("Filtered observer example completed successfully!");Line by line:
- We start by creating a new person.
- Next we subscribe a filtered observer to the
ageproperty of the person. This observer will only be triggered if the new age is strictly larger than the old age. Note that because of the error handling, the returned id of this expression is of typeOption<>, and it will be none if thesubscribe_filtered()method returns an error. - Next we test the new observer three times, the first and third time should trigger the observer, the second time should not trigger the observer as the new age is smaller than the old age.
- Now we can unsubscribe the observer. We first have to confirm the existence of an observer id. If there is one, we try and unsubscribe the observer, and handle both the
Ok()andErr()variants. - Print a status message.
Full code for the implementation of ObservableProperty<T>
The full implementation of ObservableProperty<T> looks like this:
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 set_async(&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)
};
if observers_snapshot.is_empty() {
return Ok(());
}
const MAX_THREADS: usize = 4;
let observers_per_thread = (observers_snapshot.len() + MAX_THREADS - 1) / MAX_THREADS;
for batch in observers_snapshot.chunks(observers_per_thread) {
let batch_observers = batch.to_vec();
let old_val = old_value.clone();
let new_val = new_value.clone();
thread::spawn(move || {
for observer in batch_observers {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_val, &new_val);
})) {
eprintln!("Observer panic in batch thread: {:?}", 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)
}
pub fn subscribe_filtered<F>(
&self,
observer: Observer<T>,
filter: F,
) -> Result<ObserverId, PropertyError>
where
F: Fn(&T, &T) -> bool + Send + Sync + 'static,
{
let filter = Arc::new(filter);
let filtered_observer = Arc::new(move |old_val: &T, new_val: &T| {
if filter(old_val, new_val) {
observer(old_val, new_val);
}
});
self.subscribe(filtered_observer)
}
}The full implementation of the Person struct
Here is the full implementation of the Person struct:
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()
}
pub fn set_name_async(&self, name: String) -> Result<(), PropertyError> {
self.name.set_async(name)
}
pub fn set_age_async(&self, age: i32) -> Result<(), PropertyError> {
self.age.set_async(age)
}
}The main() function
And finally the main() function looks like this:
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);
}
match person.name.unsubscribe(name_id1) {
Ok(true) => println!("Name observer unsubscribed successfully"),
Ok(false) => println!("Name observer was already unsubscribed"),
Err(e) => eprintln!("Failed to unsubscribe name observer: {}", e),
}
match person.age.unsubscribe(age_id1) {
Ok(true) => println!("Age observer unsubscribed successfully"),
Ok(false) => println!("Age observer was already unsubscribed"),
Err(e) => eprintln!("Failed to unsubscribe age observer: {}", e),
}
println!("Demonstrating thread-safe observable properties");
let person = Arc::new(Person::new("Alice".to_string(), 30));
let observer_id = match person
.age
.subscribe(Arc::new(|old, new| {
println!(
"Observer: Age changed {} -> {} [Thread: {:?}]",
old,
new,
thread::current().id()
);
})) {
Ok(id) => {
println!("Successfully subscribed multi-thread observer with ID: {}", id);
id
}
Err(e) => {
eprintln!("Failed to subscribe multi-thread observer: {}", e);
println!("Continuing without observer...");
0 // Use 0 as placeholder since we'll skip cleanup
}
};
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));
}
});
if let Err(e) = handle1.join() {
eprintln!("Thread 1 failed to join: {:?}", e);
} else {
println!("Thread 1 completed successfully");
}
if let Err(e) = handle2.join() {
eprintln!("Thread 2 failed to join: {:?}", e);
} else {
println!("Thread 2 completed successfully");
}
match person.age.unsubscribe(observer_id) {
Ok(true) => println!("Multi-thread observer unsubscribed successfully"),
Ok(false) => println!("Multi-thread observer was already unsubscribed"),
Err(e) => eprintln!("Failed to unsubscribe multi-thread observer: {}", e),
}
println!("Multi-threading example completed successfully!");
println!("Demonstrating non-blocking observers with background threads");
let person = Person::new("Alice".to_string(), 25);
let observer_id = match person.name.subscribe(Arc::new(|old_value, new_value| {
println!("Slow observer started: {} -> {}", old_value, new_value);
std::thread::sleep(std::time::Duration::from_millis(50));
println!("Slow observer finished: {} -> {}", old_value, new_value);
})) {
Ok(id) => {
println!("Successfully subscribed slow observer with ID: {}", id);
Some(id)
}
Err(e) => {
eprintln!("Failed to subscribe slow observer: {}", e);
None
}
};
println!("Setting name with blocking call...");
let start = std::time::Instant::now();
match person.set_name("Bob".to_string()) {
Ok(()) => {
println!("Blocking call took: {:?}", start.elapsed());
}
Err(e) => {
eprintln!("Failed to set name (blocking): {}", e);
}
}
println!("\nSetting name with non-blocking call...");
let start = std::time::Instant::now();
match person.set_name_async("Charlie".to_string()) {
Ok(()) => {
println!("Non-blocking call took: {:?}", start.elapsed());
}
Err(e) => {
eprintln!("Failed to set name (non-blocking): {}", e);
}
}
std::thread::sleep(std::time::Duration::from_millis(100));
if let Some(id) = observer_id {
match person.name.unsubscribe(id) {
Ok(true) => println!("Successfully unsubscribed observer"),
Ok(false) => println!("Observer was already unsubscribed"),
Err(e) => eprintln!("Failed to unsubscribe observer: {}", e),
}
}
println!("Background thread example completed successfully!");
println!("Demonstrating filtered observers");
let person = Person::new("Alice".to_string(), 25);
let filtered_observer_id = match person
.age
.subscribe_filtered(
Arc::new(|old_val, new_val| {
println!("Age increased from {} to {}", old_val, new_val);
}),
|old_val, new_val| new_val > old_val,
) {
Ok(id) => {
println!("Successfully subscribed filtered observer with ID: {}", id);
Some(id)
}
Err(e) => {
eprintln!("Failed to subscribe filtered observer: {}", e);
None
}
};
println!("Setting age to 30 (should trigger filtered observer)...");
if let Err(e) = person.set_age(30) {
eprintln!("Failed to set age to 30: {}", e);
}
println!("Setting age to 25 (should NOT trigger filtered observer)...");
if let Err(e) = person.set_age(25) {
eprintln!("Failed to set age to 25: {}", e);
}
println!("Setting age to 35 (should trigger filtered observer again)...");
if let Err(e) = person.set_age(35) {
eprintln!("Failed to set age to 35: {}", e);
}
if let Some(id) = filtered_observer_id {
match person.age.unsubscribe(id) {
Ok(true) => println!("Successfully unsubscribed filtered observer"),
Ok(false) => println!("Filtered observer was already unsubscribed"),
Err(e) => eprintln!("Failed to unsubscribe filtered observer: {}", e),
}
}
println!("Filtered observer example completed successfully!");
}Conclusion
In conclusion, this code effectively demonstrates the power and flexibility of the observable properties pattern in Rust. By encapsulating data with a subscription mechanism, we can create a robust system where changes to an object’s state automatically trigger actions in other parts of our application.
This example highlighted several key benefits of this approach: it’s type-safe, ensuring that property changes are handled correctly; it’s thread-safe, allowing for concurrent access and modification of properties from multiple threads without data races; and it’s versatile, supporting advanced features like non-blocking asynchronous updates and filtered observers that only react to specific conditions. Ultimately, this pattern is a powerful tool for building responsive and maintainable applications, whether for a user interface, an event-driven system, or any scenario where you need to react to changes in data.
