Struct ObservableProperty

pub struct ObservableProperty<T> { /* private fields */ }

A thread-safe observable property that notifies observers when its value changes

This type wraps a value of type T and allows multiple observers to be notified whenever the value is modified. All operations are thread-safe and can be called from multiple threads concurrently.

Type Requirements

The generic type T must implement:

• Clone: Required for returning values and passing them to observers
• Send: Required for transferring between threads
• Sync: Required for concurrent access from multiple threads
• 'static: Required for observer callbacks that may outlive the original scope

Examples

use observable_property::ObservableProperty;
use std::sync::Arc;

let property = ObservableProperty::new("initial".to_string());

let observer_id = property.subscribe(Arc::new(|old, new| {
    println!("Changed from '{}' to '{}'", old, new);
})).unwrap();

property.set("updated".to_string()).unwrap(); // Prints: Changed from 'initial' to 'updated'
property.unsubscribe(observer_id).unwrap();

Implementations

impl<T: Clone + Send + Sync + 'static> ObservableProperty<T>

pub fn new(initial_value: T) -> Self

Creates a new observable property with the given initial value

Arguments

• initial_value - The starting value for this property

Examples

use observable_property::ObservableProperty;

let property = ObservableProperty::new(42);
assert_eq!(property.get().unwrap(), 42);

Examples found in repository

examples/basic.rs (line 18)

    fn new(name: String, age: i32) -> Self {
        Self {
            name: ObservableProperty::new(name),
            age: ObservableProperty::new(age),
        }
    }

    fn get_name(&self) -> Result<String, observable_property::PropertyError> {
        self.name.get()
    }

    fn set_name(&self, new_name: String) -> Result<(), observable_property::PropertyError> {
        self.name.set(new_name)
    }

    fn get_age(&self) -> Result<i32, observable_property::PropertyError> {
        self.age.get()
    }

    fn set_age(&self, new_age: i32) -> Result<(), observable_property::PropertyError> {
        self.age.set(new_age)
    }

    fn celebrate_birthday(&self) -> Result<(), observable_property::PropertyError> {
        let current_age = self.age.get()?;
        self.age.set(current_age + 1)
    }
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Basic Example ===\n");

    // Create a person
    let person = Person::new("Alice".to_string(), 25);

    // Subscribe to name changes
    let name_observer_id = person.name.subscribe(Arc::new(|old_name, new_name| {
        println!("📝 Name changed: '{}' → '{}'", old_name, new_name);
    }))?;

    // Subscribe to age changes
    let age_observer_id = person.age.subscribe(Arc::new(|old_age, new_age| {
        println!("🎂 Age changed: {} → {}", old_age, new_age);
    }))?;

    // Subscribe to only significant age changes (milestones)
    let milestone_observer_id = person.age.subscribe_filtered(
        Arc::new(|_old_age, new_age| {
            println!("🎉 Milestone reached! {} is now a special age: {}", 
                    if *new_age >= 18 { "Adult" } else { "Child" }, new_age);
        }),
        |old_age, new_age| {
            // Notify on milestone ages
            let milestones = [18, 21, 30, 40, 50, 65];
            milestones.contains(new_age) || 
            (milestones.contains(old_age) && !milestones.contains(new_age))
        }
    )?;

    println!("Initial state:");
    println!("  Name: {}", person.get_name()?);
    println!("  Age: {}\n", person.get_age()?);

    // Demonstrate property changes
    println!("Making changes...\n");

    // Change name
    person.set_name("Alice Johnson".to_string())?;

    // Age up a few times
    person.celebrate_birthday()?;
    person.celebrate_birthday()?;
    person.celebrate_birthday()?;

    // Change to milestone age
    person.set_age(21)?;

    // Direct property access
    println!("\nDirect property access:");
    let simple_property = ObservableProperty::new(100);
    
    simple_property.subscribe(Arc::new(|old, new| {
        println!("💰 Value changed: {} → {}", old, new);
    }))?;

    simple_property.set(200)?;
    simple_property.set(150)?;

    // Cleanup observers
    person.name.unsubscribe(name_observer_id)?;
    person.age.unsubscribe(age_observer_id)?;
    person.age.unsubscribe(milestone_observer_id)?;

    println!("\n✅ Example completed successfully!");
    Ok(())
}

examples/multithreaded.rs (line 15)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Multithreaded Example ===\n");

    // Shared counter with multiple observers
    let counter = Arc::new(ObservableProperty::new(0i32));
    let notification_count = Arc::new(AtomicUsize::new(0));

    // Add observers that track notifications
    let count_clone = notification_count.clone();
    let observer1_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔴 Observer 1: {} → {}", old, new);
    }))?;

    let count_clone = notification_count.clone();
    let observer2_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔵 Observer 2: {} → {}", old, new);
    }))?;

    // Filtered observer for even numbers only
    let count_clone = notification_count.clone();
    let even_observer_id = counter.subscribe_filtered(
        Arc::new(move |old, new| {
            count_clone.fetch_add(1, Ordering::SeqCst);
            println!("🟢 Even Observer: {} → {} (new value is even!)", old, new);
        }),
        |_old, new| new % 2 == 0
    )?;

    println!("📊 Starting concurrent updates...\n");

    // Test 1: Multiple threads updating the counter
    {
        let num_threads = 4;
        let updates_per_thread = 5;
        let barrier = Arc::new(Barrier::new(num_threads));

        let handles: Vec<_> = (0..num_threads).map(|thread_id| {
            let counter_clone = counter.clone();
            let barrier_clone = barrier.clone();

            thread::spawn(move || {
                // Wait for all threads to be ready
                barrier_clone.wait();

                for i in 0..updates_per_thread {
                    let new_value = (thread_id * 100 + i) as i32;
                    
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Thread {} failed to set counter: {}", thread_id, e);
                    } else {
                        println!("🧵 Thread {} set counter to {}", thread_id, new_value);
                    }

                    // Small delay to see interleaving
                    thread::sleep(Duration::from_millis(10));
                }
            })
        }).collect();

        // Wait for all threads to complete
        for handle in handles {
            if let Err(e) = handle.join() {
                eprintln!("Thread panicked: {:?}", e);
            }
        }
    }

    thread::sleep(Duration::from_millis(100)); // Let notifications settle

    println!("\n📈 Current counter value: {}", counter.get()?);
    println!("🔔 Total notifications sent: {}\n", notification_count.load(Ordering::SeqCst));

    // Test 2: Performance comparison between sync and async notifications
    println!("⚡ Performance Test: Sync vs Async notifications\n");

    // Create a property with a slow observer
    let perf_counter = ObservableProperty::new(0i32);
    let slow_notifications = Arc::new(AtomicUsize::new(0));
    let slow_count_clone = slow_notifications.clone();

    perf_counter.subscribe(Arc::new(move |_old, _new| {
        // Simulate slow observer work
        thread::sleep(Duration::from_millis(20));
        slow_count_clone.fetch_add(1, Ordering::SeqCst);
    }))?;

    // Test synchronous updates (will block)
    let start = Instant::now();
    for i in 1..=3 {
        perf_counter.set(i)?;
    }
    let sync_duration = start.elapsed();
    println!("🐌 Sync updates (3 changes): {:?}", sync_duration);

    // Test asynchronous updates (won't block)
    let start = Instant::now();
    for i in 4..=6 {
        perf_counter.set_async(i)?;
    }
    let async_duration = start.elapsed();
    println!("🚀 Async updates (3 changes): {:?}", async_duration);

    println!("⚡ Speedup: {:.1}x faster", 
             sync_duration.as_secs_f64() / async_duration.as_secs_f64());

    // Wait for async observers to complete
    thread::sleep(Duration::from_millis(100));
    println!("🔔 Slow observer notifications: {}\n", slow_notifications.load(Ordering::SeqCst));

    // Test 3: Concurrent reads while updating
    println!("📚 Concurrent reads test...\n");
    {
        let read_counter = Arc::new(ObservableProperty::new(1000i32));
        let read_count = Arc::new(AtomicUsize::new(0));
        let num_readers = 8;
        let reads_per_reader = 50;

        // Start reader threads
        let reader_handles: Vec<_> = (0..num_readers).map(|reader_id| {
            let counter_clone = read_counter.clone();
            let count_clone = read_count.clone();

            thread::spawn(move || {
                for _ in 0..reads_per_reader {
                    match counter_clone.get() {
                        Ok(value) => {
                            count_clone.fetch_add(1, Ordering::SeqCst);
                            if value % 100 == 0 {
                                println!("📖 Reader {} saw value: {}", reader_id, value);
                            }
                        }
                        Err(e) => eprintln!("Reader {} error: {}", reader_id, e),
                    }
                    thread::sleep(Duration::from_millis(1));
                }
            })
        }).collect();

        // Start writer thread
        let writer_handle = {
            let counter_clone = read_counter.clone();
            thread::spawn(move || {
                for i in 1..=20 {
                    let new_value = 1000 + i * 50;
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Writer error: {}", e);
                    } else {
                        println!("✍️  Writer set value to: {}", new_value);
                    }
                    thread::sleep(Duration::from_millis(25));
                }
            })
        };

        // Wait for all threads
        for handle in reader_handles {
            if let Err(e) = handle.join() {
                eprintln!("Reader thread panicked: {:?}", e);
            }
        }

        if let Err(e) = writer_handle.join() {
            eprintln!("Writer thread panicked: {:?}", e);
        }

        println!("📊 Total successful reads: {}", read_count.load(Ordering::SeqCst));
    }

    // Cleanup
    counter.unsubscribe(observer1_id)?;
    counter.unsubscribe(observer2_id)?;
    counter.unsubscribe(even_observer_id)?;

    println!("\n✅ Multithreaded example completed successfully!");
    Ok(())
}

pub fn get(&self) -> Result<T, PropertyError>

Gets the current value of the property

This method acquires a read lock, which allows multiple concurrent readers but will block if a writer currently holds the lock.

Returns

Ok(T) containing a clone of the current value, or Err(PropertyError) if the lock is poisoned.

Examples

use observable_property::ObservableProperty;

let property = ObservableProperty::new("hello".to_string());
assert_eq!(property.get().unwrap(), "hello");

Examples found in repository

examples/basic.rs (line 24)

    fn get_name(&self) -> Result<String, observable_property::PropertyError> {
        self.name.get()
    }

    fn set_name(&self, new_name: String) -> Result<(), observable_property::PropertyError> {
        self.name.set(new_name)
    }

    fn get_age(&self) -> Result<i32, observable_property::PropertyError> {
        self.age.get()
    }

    fn set_age(&self, new_age: i32) -> Result<(), observable_property::PropertyError> {
        self.age.set(new_age)
    }

    fn celebrate_birthday(&self) -> Result<(), observable_property::PropertyError> {
        let current_age = self.age.get()?;
        self.age.set(current_age + 1)
    }

examples/multithreaded.rs (line 82)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Multithreaded Example ===\n");

    // Shared counter with multiple observers
    let counter = Arc::new(ObservableProperty::new(0i32));
    let notification_count = Arc::new(AtomicUsize::new(0));

    // Add observers that track notifications
    let count_clone = notification_count.clone();
    let observer1_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔴 Observer 1: {} → {}", old, new);
    }))?;

    let count_clone = notification_count.clone();
    let observer2_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔵 Observer 2: {} → {}", old, new);
    }))?;

    // Filtered observer for even numbers only
    let count_clone = notification_count.clone();
    let even_observer_id = counter.subscribe_filtered(
        Arc::new(move |old, new| {
            count_clone.fetch_add(1, Ordering::SeqCst);
            println!("🟢 Even Observer: {} → {} (new value is even!)", old, new);
        }),
        |_old, new| new % 2 == 0
    )?;

    println!("📊 Starting concurrent updates...\n");

    // Test 1: Multiple threads updating the counter
    {
        let num_threads = 4;
        let updates_per_thread = 5;
        let barrier = Arc::new(Barrier::new(num_threads));

        let handles: Vec<_> = (0..num_threads).map(|thread_id| {
            let counter_clone = counter.clone();
            let barrier_clone = barrier.clone();

            thread::spawn(move || {
                // Wait for all threads to be ready
                barrier_clone.wait();

                for i in 0..updates_per_thread {
                    let new_value = (thread_id * 100 + i) as i32;
                    
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Thread {} failed to set counter: {}", thread_id, e);
                    } else {
                        println!("🧵 Thread {} set counter to {}", thread_id, new_value);
                    }

                    // Small delay to see interleaving
                    thread::sleep(Duration::from_millis(10));
                }
            })
        }).collect();

        // Wait for all threads to complete
        for handle in handles {
            if let Err(e) = handle.join() {
                eprintln!("Thread panicked: {:?}", e);
            }
        }
    }

    thread::sleep(Duration::from_millis(100)); // Let notifications settle

    println!("\n📈 Current counter value: {}", counter.get()?);
    println!("🔔 Total notifications sent: {}\n", notification_count.load(Ordering::SeqCst));

    // Test 2: Performance comparison between sync and async notifications
    println!("⚡ Performance Test: Sync vs Async notifications\n");

    // Create a property with a slow observer
    let perf_counter = ObservableProperty::new(0i32);
    let slow_notifications = Arc::new(AtomicUsize::new(0));
    let slow_count_clone = slow_notifications.clone();

    perf_counter.subscribe(Arc::new(move |_old, _new| {
        // Simulate slow observer work
        thread::sleep(Duration::from_millis(20));
        slow_count_clone.fetch_add(1, Ordering::SeqCst);
    }))?;

    // Test synchronous updates (will block)
    let start = Instant::now();
    for i in 1..=3 {
        perf_counter.set(i)?;
    }
    let sync_duration = start.elapsed();
    println!("🐌 Sync updates (3 changes): {:?}", sync_duration);

    // Test asynchronous updates (won't block)
    let start = Instant::now();
    for i in 4..=6 {
        perf_counter.set_async(i)?;
    }
    let async_duration = start.elapsed();
    println!("🚀 Async updates (3 changes): {:?}", async_duration);

    println!("⚡ Speedup: {:.1}x faster", 
             sync_duration.as_secs_f64() / async_duration.as_secs_f64());

    // Wait for async observers to complete
    thread::sleep(Duration::from_millis(100));
    println!("🔔 Slow observer notifications: {}\n", slow_notifications.load(Ordering::SeqCst));

    // Test 3: Concurrent reads while updating
    println!("📚 Concurrent reads test...\n");
    {
        let read_counter = Arc::new(ObservableProperty::new(1000i32));
        let read_count = Arc::new(AtomicUsize::new(0));
        let num_readers = 8;
        let reads_per_reader = 50;

        // Start reader threads
        let reader_handles: Vec<_> = (0..num_readers).map(|reader_id| {
            let counter_clone = read_counter.clone();
            let count_clone = read_count.clone();

            thread::spawn(move || {
                for _ in 0..reads_per_reader {
                    match counter_clone.get() {
                        Ok(value) => {
                            count_clone.fetch_add(1, Ordering::SeqCst);
                            if value % 100 == 0 {
                                println!("📖 Reader {} saw value: {}", reader_id, value);
                            }
                        }
                        Err(e) => eprintln!("Reader {} error: {}", reader_id, e),
                    }
                    thread::sleep(Duration::from_millis(1));
                }
            })
        }).collect();

        // Start writer thread
        let writer_handle = {
            let counter_clone = read_counter.clone();
            thread::spawn(move || {
                for i in 1..=20 {
                    let new_value = 1000 + i * 50;
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Writer error: {}", e);
                    } else {
                        println!("✍️  Writer set value to: {}", new_value);
                    }
                    thread::sleep(Duration::from_millis(25));
                }
            })
        };

        // Wait for all threads
        for handle in reader_handles {
            if let Err(e) = handle.join() {
                eprintln!("Reader thread panicked: {:?}", e);
            }
        }

        if let Err(e) = writer_handle.join() {
            eprintln!("Writer thread panicked: {:?}", e);
        }

        println!("📊 Total successful reads: {}", read_count.load(Ordering::SeqCst));
    }

    // Cleanup
    counter.unsubscribe(observer1_id)?;
    counter.unsubscribe(observer2_id)?;
    counter.unsubscribe(even_observer_id)?;

    println!("\n✅ Multithreaded example completed successfully!");
    Ok(())
}

pub fn set(&self, new_value: T) -> Result<(), PropertyError>

Sets the property to a new value and notifies all observers

This method will:

1. Acquire a write lock (blocking other readers/writers)
2. Update the value and capture a snapshot of observers
3. Release the lock
4. Notify all observers sequentially with the old and new values

Observer notifications are wrapped in panic recovery to prevent one misbehaving observer from affecting others.

Arguments

• new_value - The new value to set

Returns

Ok(()) if successful, or Err(PropertyError) if the lock is poisoned.

Examples

use observable_property::ObservableProperty;
use std::sync::Arc;

let property = ObservableProperty::new(10);
 
property.subscribe(Arc::new(|old, new| {
    println!("Value changed from {} to {}", old, new);
})).unwrap();

property.set(20).unwrap(); // Triggers observer notification

Examples found in repository

examples/basic.rs (line 28)

    fn set_name(&self, new_name: String) -> Result<(), observable_property::PropertyError> {
        self.name.set(new_name)
    }

    fn get_age(&self) -> Result<i32, observable_property::PropertyError> {
        self.age.get()
    }

    fn set_age(&self, new_age: i32) -> Result<(), observable_property::PropertyError> {
        self.age.set(new_age)
    }

    fn celebrate_birthday(&self) -> Result<(), observable_property::PropertyError> {
        let current_age = self.age.get()?;
        self.age.set(current_age + 1)
    }
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Basic Example ===\n");

    // Create a person
    let person = Person::new("Alice".to_string(), 25);

    // Subscribe to name changes
    let name_observer_id = person.name.subscribe(Arc::new(|old_name, new_name| {
        println!("📝 Name changed: '{}' → '{}'", old_name, new_name);
    }))?;

    // Subscribe to age changes
    let age_observer_id = person.age.subscribe(Arc::new(|old_age, new_age| {
        println!("🎂 Age changed: {} → {}", old_age, new_age);
    }))?;

    // Subscribe to only significant age changes (milestones)
    let milestone_observer_id = person.age.subscribe_filtered(
        Arc::new(|_old_age, new_age| {
            println!("🎉 Milestone reached! {} is now a special age: {}", 
                    if *new_age >= 18 { "Adult" } else { "Child" }, new_age);
        }),
        |old_age, new_age| {
            // Notify on milestone ages
            let milestones = [18, 21, 30, 40, 50, 65];
            milestones.contains(new_age) || 
            (milestones.contains(old_age) && !milestones.contains(new_age))
        }
    )?;

    println!("Initial state:");
    println!("  Name: {}", person.get_name()?);
    println!("  Age: {}\n", person.get_age()?);

    // Demonstrate property changes
    println!("Making changes...\n");

    // Change name
    person.set_name("Alice Johnson".to_string())?;

    // Age up a few times
    person.celebrate_birthday()?;
    person.celebrate_birthday()?;
    person.celebrate_birthday()?;

    // Change to milestone age
    person.set_age(21)?;

    // Direct property access
    println!("\nDirect property access:");
    let simple_property = ObservableProperty::new(100);
    
    simple_property.subscribe(Arc::new(|old, new| {
        println!("💰 Value changed: {} → {}", old, new);
    }))?;

    simple_property.set(200)?;
    simple_property.set(150)?;

    // Cleanup observers
    person.name.unsubscribe(name_observer_id)?;
    person.age.unsubscribe(age_observer_id)?;
    person.age.unsubscribe(milestone_observer_id)?;

    println!("\n✅ Example completed successfully!");
    Ok(())
}

examples/multithreaded.rs (line 60)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Multithreaded Example ===\n");

    // Shared counter with multiple observers
    let counter = Arc::new(ObservableProperty::new(0i32));
    let notification_count = Arc::new(AtomicUsize::new(0));

    // Add observers that track notifications
    let count_clone = notification_count.clone();
    let observer1_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔴 Observer 1: {} → {}", old, new);
    }))?;

    let count_clone = notification_count.clone();
    let observer2_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔵 Observer 2: {} → {}", old, new);
    }))?;

    // Filtered observer for even numbers only
    let count_clone = notification_count.clone();
    let even_observer_id = counter.subscribe_filtered(
        Arc::new(move |old, new| {
            count_clone.fetch_add(1, Ordering::SeqCst);
            println!("🟢 Even Observer: {} → {} (new value is even!)", old, new);
        }),
        |_old, new| new % 2 == 0
    )?;

    println!("📊 Starting concurrent updates...\n");

    // Test 1: Multiple threads updating the counter
    {
        let num_threads = 4;
        let updates_per_thread = 5;
        let barrier = Arc::new(Barrier::new(num_threads));

        let handles: Vec<_> = (0..num_threads).map(|thread_id| {
            let counter_clone = counter.clone();
            let barrier_clone = barrier.clone();

            thread::spawn(move || {
                // Wait for all threads to be ready
                barrier_clone.wait();

                for i in 0..updates_per_thread {
                    let new_value = (thread_id * 100 + i) as i32;
                    
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Thread {} failed to set counter: {}", thread_id, e);
                    } else {
                        println!("🧵 Thread {} set counter to {}", thread_id, new_value);
                    }

                    // Small delay to see interleaving
                    thread::sleep(Duration::from_millis(10));
                }
            })
        }).collect();

        // Wait for all threads to complete
        for handle in handles {
            if let Err(e) = handle.join() {
                eprintln!("Thread panicked: {:?}", e);
            }
        }
    }

    thread::sleep(Duration::from_millis(100)); // Let notifications settle

    println!("\n📈 Current counter value: {}", counter.get()?);
    println!("🔔 Total notifications sent: {}\n", notification_count.load(Ordering::SeqCst));

    // Test 2: Performance comparison between sync and async notifications
    println!("⚡ Performance Test: Sync vs Async notifications\n");

    // Create a property with a slow observer
    let perf_counter = ObservableProperty::new(0i32);
    let slow_notifications = Arc::new(AtomicUsize::new(0));
    let slow_count_clone = slow_notifications.clone();

    perf_counter.subscribe(Arc::new(move |_old, _new| {
        // Simulate slow observer work
        thread::sleep(Duration::from_millis(20));
        slow_count_clone.fetch_add(1, Ordering::SeqCst);
    }))?;

    // Test synchronous updates (will block)
    let start = Instant::now();
    for i in 1..=3 {
        perf_counter.set(i)?;
    }
    let sync_duration = start.elapsed();
    println!("🐌 Sync updates (3 changes): {:?}", sync_duration);

    // Test asynchronous updates (won't block)
    let start = Instant::now();
    for i in 4..=6 {
        perf_counter.set_async(i)?;
    }
    let async_duration = start.elapsed();
    println!("🚀 Async updates (3 changes): {:?}", async_duration);

    println!("⚡ Speedup: {:.1}x faster", 
             sync_duration.as_secs_f64() / async_duration.as_secs_f64());

    // Wait for async observers to complete
    thread::sleep(Duration::from_millis(100));
    println!("🔔 Slow observer notifications: {}\n", slow_notifications.load(Ordering::SeqCst));

    // Test 3: Concurrent reads while updating
    println!("📚 Concurrent reads test...\n");
    {
        let read_counter = Arc::new(ObservableProperty::new(1000i32));
        let read_count = Arc::new(AtomicUsize::new(0));
        let num_readers = 8;
        let reads_per_reader = 50;

        // Start reader threads
        let reader_handles: Vec<_> = (0..num_readers).map(|reader_id| {
            let counter_clone = read_counter.clone();
            let count_clone = read_count.clone();

            thread::spawn(move || {
                for _ in 0..reads_per_reader {
                    match counter_clone.get() {
                        Ok(value) => {
                            count_clone.fetch_add(1, Ordering::SeqCst);
                            if value % 100 == 0 {
                                println!("📖 Reader {} saw value: {}", reader_id, value);
                            }
                        }
                        Err(e) => eprintln!("Reader {} error: {}", reader_id, e),
                    }
                    thread::sleep(Duration::from_millis(1));
                }
            })
        }).collect();

        // Start writer thread
        let writer_handle = {
            let counter_clone = read_counter.clone();
            thread::spawn(move || {
                for i in 1..=20 {
                    let new_value = 1000 + i * 50;
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Writer error: {}", e);
                    } else {
                        println!("✍️  Writer set value to: {}", new_value);
                    }
                    thread::sleep(Duration::from_millis(25));
                }
            })
        };

        // Wait for all threads
        for handle in reader_handles {
            if let Err(e) = handle.join() {
                eprintln!("Reader thread panicked: {:?}", e);
            }
        }

        if let Err(e) = writer_handle.join() {
            eprintln!("Writer thread panicked: {:?}", e);
        }

        println!("📊 Total successful reads: {}", read_count.load(Ordering::SeqCst));
    }

    // Cleanup
    counter.unsubscribe(observer1_id)?;
    counter.unsubscribe(observer2_id)?;
    counter.unsubscribe(even_observer_id)?;

    println!("\n✅ Multithreaded example completed successfully!");
    Ok(())
}

pub fn set_async(&self, new_value: T) -> Result<(), PropertyError>

Sets the property to a new value and notifies observers asynchronously

This method is similar to set() but spawns observers in background threads for non-blocking operation. This is useful when observers might perform time-consuming operations.

Observers are batched into groups and each batch runs in its own thread to limit resource usage while still providing parallelism.

Arguments

• new_value - The new value to set

Returns

Ok(()) if successful, or Err(PropertyError) if the lock is poisoned. Note that this only indicates the property was updated successfully; observer execution happens asynchronously.

Examples

use observable_property::ObservableProperty;
use std::sync::Arc;
use std::time::Duration;

let property = ObservableProperty::new(0);
 
property.subscribe(Arc::new(|old, new| {
    // This observer does slow work but won't block the caller
    std::thread::sleep(Duration::from_millis(100));
    println!("Slow observer: {} -> {}", old, new);
})).unwrap();

// This returns immediately even though observer is slow
property.set_async(42).unwrap();

Examples found in repository

examples/multithreaded.rs (line 110)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Multithreaded Example ===\n");

    // Shared counter with multiple observers
    let counter = Arc::new(ObservableProperty::new(0i32));
    let notification_count = Arc::new(AtomicUsize::new(0));

    // Add observers that track notifications
    let count_clone = notification_count.clone();
    let observer1_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔴 Observer 1: {} → {}", old, new);
    }))?;

    let count_clone = notification_count.clone();
    let observer2_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔵 Observer 2: {} → {}", old, new);
    }))?;

    // Filtered observer for even numbers only
    let count_clone = notification_count.clone();
    let even_observer_id = counter.subscribe_filtered(
        Arc::new(move |old, new| {
            count_clone.fetch_add(1, Ordering::SeqCst);
            println!("🟢 Even Observer: {} → {} (new value is even!)", old, new);
        }),
        |_old, new| new % 2 == 0
    )?;

    println!("📊 Starting concurrent updates...\n");

    // Test 1: Multiple threads updating the counter
    {
        let num_threads = 4;
        let updates_per_thread = 5;
        let barrier = Arc::new(Barrier::new(num_threads));

        let handles: Vec<_> = (0..num_threads).map(|thread_id| {
            let counter_clone = counter.clone();
            let barrier_clone = barrier.clone();

            thread::spawn(move || {
                // Wait for all threads to be ready
                barrier_clone.wait();

                for i in 0..updates_per_thread {
                    let new_value = (thread_id * 100 + i) as i32;
                    
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Thread {} failed to set counter: {}", thread_id, e);
                    } else {
                        println!("🧵 Thread {} set counter to {}", thread_id, new_value);
                    }

                    // Small delay to see interleaving
                    thread::sleep(Duration::from_millis(10));
                }
            })
        }).collect();

        // Wait for all threads to complete
        for handle in handles {
            if let Err(e) = handle.join() {
                eprintln!("Thread panicked: {:?}", e);
            }
        }
    }

    thread::sleep(Duration::from_millis(100)); // Let notifications settle

    println!("\n📈 Current counter value: {}", counter.get()?);
    println!("🔔 Total notifications sent: {}\n", notification_count.load(Ordering::SeqCst));

    // Test 2: Performance comparison between sync and async notifications
    println!("⚡ Performance Test: Sync vs Async notifications\n");

    // Create a property with a slow observer
    let perf_counter = ObservableProperty::new(0i32);
    let slow_notifications = Arc::new(AtomicUsize::new(0));
    let slow_count_clone = slow_notifications.clone();

    perf_counter.subscribe(Arc::new(move |_old, _new| {
        // Simulate slow observer work
        thread::sleep(Duration::from_millis(20));
        slow_count_clone.fetch_add(1, Ordering::SeqCst);
    }))?;

    // Test synchronous updates (will block)
    let start = Instant::now();
    for i in 1..=3 {
        perf_counter.set(i)?;
    }
    let sync_duration = start.elapsed();
    println!("🐌 Sync updates (3 changes): {:?}", sync_duration);

    // Test asynchronous updates (won't block)
    let start = Instant::now();
    for i in 4..=6 {
        perf_counter.set_async(i)?;
    }
    let async_duration = start.elapsed();
    println!("🚀 Async updates (3 changes): {:?}", async_duration);

    println!("⚡ Speedup: {:.1}x faster", 
             sync_duration.as_secs_f64() / async_duration.as_secs_f64());

    // Wait for async observers to complete
    thread::sleep(Duration::from_millis(100));
    println!("🔔 Slow observer notifications: {}\n", slow_notifications.load(Ordering::SeqCst));

    // Test 3: Concurrent reads while updating
    println!("📚 Concurrent reads test...\n");
    {
        let read_counter = Arc::new(ObservableProperty::new(1000i32));
        let read_count = Arc::new(AtomicUsize::new(0));
        let num_readers = 8;
        let reads_per_reader = 50;

        // Start reader threads
        let reader_handles: Vec<_> = (0..num_readers).map(|reader_id| {
            let counter_clone = read_counter.clone();
            let count_clone = read_count.clone();

            thread::spawn(move || {
                for _ in 0..reads_per_reader {
                    match counter_clone.get() {
                        Ok(value) => {
                            count_clone.fetch_add(1, Ordering::SeqCst);
                            if value % 100 == 0 {
                                println!("📖 Reader {} saw value: {}", reader_id, value);
                            }
                        }
                        Err(e) => eprintln!("Reader {} error: {}", reader_id, e),
                    }
                    thread::sleep(Duration::from_millis(1));
                }
            })
        }).collect();

        // Start writer thread
        let writer_handle = {
            let counter_clone = read_counter.clone();
            thread::spawn(move || {
                for i in 1..=20 {
                    let new_value = 1000 + i * 50;
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Writer error: {}", e);
                    } else {
                        println!("✍️  Writer set value to: {}", new_value);
                    }
                    thread::sleep(Duration::from_millis(25));
                }
            })
        };

        // Wait for all threads
        for handle in reader_handles {
            if let Err(e) = handle.join() {
                eprintln!("Reader thread panicked: {:?}", e);
            }
        }

        if let Err(e) = writer_handle.join() {
            eprintln!("Writer thread panicked: {:?}", e);
        }

        println!("📊 Total successful reads: {}", read_count.load(Ordering::SeqCst));
    }

    // Cleanup
    counter.unsubscribe(observer1_id)?;
    counter.unsubscribe(observer2_id)?;
    counter.unsubscribe(even_observer_id)?;

    println!("\n✅ Multithreaded example completed successfully!");
    Ok(())
}

pub fn subscribe( &self, observer: Observer<T>, ) -> Result<ObserverId, PropertyError>

Subscribes an observer function to be called when the property changes

The observer function will be called with the old and new values whenever the property is modified via set() or set_async().

Arguments

• observer - A function wrapped in Arc that takes (&T, &T) parameters

Returns

Ok(ObserverId) containing a unique identifier for this observer, or Err(PropertyError) if the lock is poisoned.

Examples

use observable_property::ObservableProperty;
use std::sync::Arc;

let property = ObservableProperty::new(0);

let observer_id = property.subscribe(Arc::new(|old_value, new_value| {
    println!("Property changed from {} to {}", old_value, new_value);
})).unwrap();

// Later, unsubscribe using the returned ID
property.unsubscribe(observer_id).unwrap();

Examples found in repository

examples/basic.rs (lines 52-54)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Basic Example ===\n");

    // Create a person
    let person = Person::new("Alice".to_string(), 25);

    // Subscribe to name changes
    let name_observer_id = person.name.subscribe(Arc::new(|old_name, new_name| {
        println!("📝 Name changed: '{}' → '{}'", old_name, new_name);
    }))?;

    // Subscribe to age changes
    let age_observer_id = person.age.subscribe(Arc::new(|old_age, new_age| {
        println!("🎂 Age changed: {} → {}", old_age, new_age);
    }))?;

    // Subscribe to only significant age changes (milestones)
    let milestone_observer_id = person.age.subscribe_filtered(
        Arc::new(|_old_age, new_age| {
            println!("🎉 Milestone reached! {} is now a special age: {}", 
                    if *new_age >= 18 { "Adult" } else { "Child" }, new_age);
        }),
        |old_age, new_age| {
            // Notify on milestone ages
            let milestones = [18, 21, 30, 40, 50, 65];
            milestones.contains(new_age) || 
            (milestones.contains(old_age) && !milestones.contains(new_age))
        }
    )?;

    println!("Initial state:");
    println!("  Name: {}", person.get_name()?);
    println!("  Age: {}\n", person.get_age()?);

    // Demonstrate property changes
    println!("Making changes...\n");

    // Change name
    person.set_name("Alice Johnson".to_string())?;

    // Age up a few times
    person.celebrate_birthday()?;
    person.celebrate_birthday()?;
    person.celebrate_birthday()?;

    // Change to milestone age
    person.set_age(21)?;

    // Direct property access
    println!("\nDirect property access:");
    let simple_property = ObservableProperty::new(100);
    
    simple_property.subscribe(Arc::new(|old, new| {
        println!("💰 Value changed: {} → {}", old, new);
    }))?;

    simple_property.set(200)?;
    simple_property.set(150)?;

    // Cleanup observers
    person.name.unsubscribe(name_observer_id)?;
    person.age.unsubscribe(age_observer_id)?;
    person.age.unsubscribe(milestone_observer_id)?;

    println!("\n✅ Example completed successfully!");
    Ok(())
}

examples/multithreaded.rs (lines 20-23)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Multithreaded Example ===\n");

    // Shared counter with multiple observers
    let counter = Arc::new(ObservableProperty::new(0i32));
    let notification_count = Arc::new(AtomicUsize::new(0));

    // Add observers that track notifications
    let count_clone = notification_count.clone();
    let observer1_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔴 Observer 1: {} → {}", old, new);
    }))?;

    let count_clone = notification_count.clone();
    let observer2_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔵 Observer 2: {} → {}", old, new);
    }))?;

    // Filtered observer for even numbers only
    let count_clone = notification_count.clone();
    let even_observer_id = counter.subscribe_filtered(
        Arc::new(move |old, new| {
            count_clone.fetch_add(1, Ordering::SeqCst);
            println!("🟢 Even Observer: {} → {} (new value is even!)", old, new);
        }),
        |_old, new| new % 2 == 0
    )?;

    println!("📊 Starting concurrent updates...\n");

    // Test 1: Multiple threads updating the counter
    {
        let num_threads = 4;
        let updates_per_thread = 5;
        let barrier = Arc::new(Barrier::new(num_threads));

        let handles: Vec<_> = (0..num_threads).map(|thread_id| {
            let counter_clone = counter.clone();
            let barrier_clone = barrier.clone();

            thread::spawn(move || {
                // Wait for all threads to be ready
                barrier_clone.wait();

                for i in 0..updates_per_thread {
                    let new_value = (thread_id * 100 + i) as i32;
                    
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Thread {} failed to set counter: {}", thread_id, e);
                    } else {
                        println!("🧵 Thread {} set counter to {}", thread_id, new_value);
                    }

                    // Small delay to see interleaving
                    thread::sleep(Duration::from_millis(10));
                }
            })
        }).collect();

        // Wait for all threads to complete
        for handle in handles {
            if let Err(e) = handle.join() {
                eprintln!("Thread panicked: {:?}", e);
            }
        }
    }

    thread::sleep(Duration::from_millis(100)); // Let notifications settle

    println!("\n📈 Current counter value: {}", counter.get()?);
    println!("🔔 Total notifications sent: {}\n", notification_count.load(Ordering::SeqCst));

    // Test 2: Performance comparison between sync and async notifications
    println!("⚡ Performance Test: Sync vs Async notifications\n");

    // Create a property with a slow observer
    let perf_counter = ObservableProperty::new(0i32);
    let slow_notifications = Arc::new(AtomicUsize::new(0));
    let slow_count_clone = slow_notifications.clone();

    perf_counter.subscribe(Arc::new(move |_old, _new| {
        // Simulate slow observer work
        thread::sleep(Duration::from_millis(20));
        slow_count_clone.fetch_add(1, Ordering::SeqCst);
    }))?;

    // Test synchronous updates (will block)
    let start = Instant::now();
    for i in 1..=3 {
        perf_counter.set(i)?;
    }
    let sync_duration = start.elapsed();
    println!("🐌 Sync updates (3 changes): {:?}", sync_duration);

    // Test asynchronous updates (won't block)
    let start = Instant::now();
    for i in 4..=6 {
        perf_counter.set_async(i)?;
    }
    let async_duration = start.elapsed();
    println!("🚀 Async updates (3 changes): {:?}", async_duration);

    println!("⚡ Speedup: {:.1}x faster", 
             sync_duration.as_secs_f64() / async_duration.as_secs_f64());

    // Wait for async observers to complete
    thread::sleep(Duration::from_millis(100));
    println!("🔔 Slow observer notifications: {}\n", slow_notifications.load(Ordering::SeqCst));

    // Test 3: Concurrent reads while updating
    println!("📚 Concurrent reads test...\n");
    {
        let read_counter = Arc::new(ObservableProperty::new(1000i32));
        let read_count = Arc::new(AtomicUsize::new(0));
        let num_readers = 8;
        let reads_per_reader = 50;

        // Start reader threads
        let reader_handles: Vec<_> = (0..num_readers).map(|reader_id| {
            let counter_clone = read_counter.clone();
            let count_clone = read_count.clone();

            thread::spawn(move || {
                for _ in 0..reads_per_reader {
                    match counter_clone.get() {
                        Ok(value) => {
                            count_clone.fetch_add(1, Ordering::SeqCst);
                            if value % 100 == 0 {
                                println!("📖 Reader {} saw value: {}", reader_id, value);
                            }
                        }
                        Err(e) => eprintln!("Reader {} error: {}", reader_id, e),
                    }
                    thread::sleep(Duration::from_millis(1));
                }
            })
        }).collect();

        // Start writer thread
        let writer_handle = {
            let counter_clone = read_counter.clone();
            thread::spawn(move || {
                for i in 1..=20 {
                    let new_value = 1000 + i * 50;
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Writer error: {}", e);
                    } else {
                        println!("✍️  Writer set value to: {}", new_value);
                    }
                    thread::sleep(Duration::from_millis(25));
                }
            })
        };

        // Wait for all threads
        for handle in reader_handles {
            if let Err(e) = handle.join() {
                eprintln!("Reader thread panicked: {:?}", e);
            }
        }

        if let Err(e) = writer_handle.join() {
            eprintln!("Writer thread panicked: {:?}", e);
        }

        println!("📊 Total successful reads: {}", read_count.load(Ordering::SeqCst));
    }

    // Cleanup
    counter.unsubscribe(observer1_id)?;
    counter.unsubscribe(observer2_id)?;
    counter.unsubscribe(even_observer_id)?;

    println!("\n✅ Multithreaded example completed successfully!");
    Ok(())
}

pub fn unsubscribe(&self, id: ObserverId) -> Result<bool, PropertyError>

Removes an observer by its ID

Arguments

• id - The observer ID returned by subscribe()

Returns

Ok(bool) where true means the observer was found and removed, false means no observer with that ID existed. Returns Err(PropertyError) if the lock is poisoned.

Examples

use observable_property::ObservableProperty;
use std::sync::Arc;

let property = ObservableProperty::new(0);
let id = property.subscribe(Arc::new(|_, _| {})).unwrap();

let was_removed = property.unsubscribe(id).unwrap();
assert!(was_removed); // Observer existed and was removed

let was_removed_again = property.unsubscribe(id).unwrap();
assert!(!was_removed_again); // Observer no longer exists

Examples found in repository

examples/basic.rs (line 105)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Basic Example ===\n");

    // Create a person
    let person = Person::new("Alice".to_string(), 25);

    // Subscribe to name changes
    let name_observer_id = person.name.subscribe(Arc::new(|old_name, new_name| {
        println!("📝 Name changed: '{}' → '{}'", old_name, new_name);
    }))?;

    // Subscribe to age changes
    let age_observer_id = person.age.subscribe(Arc::new(|old_age, new_age| {
        println!("🎂 Age changed: {} → {}", old_age, new_age);
    }))?;

    // Subscribe to only significant age changes (milestones)
    let milestone_observer_id = person.age.subscribe_filtered(
        Arc::new(|_old_age, new_age| {
            println!("🎉 Milestone reached! {} is now a special age: {}", 
                    if *new_age >= 18 { "Adult" } else { "Child" }, new_age);
        }),
        |old_age, new_age| {
            // Notify on milestone ages
            let milestones = [18, 21, 30, 40, 50, 65];
            milestones.contains(new_age) || 
            (milestones.contains(old_age) && !milestones.contains(new_age))
        }
    )?;

    println!("Initial state:");
    println!("  Name: {}", person.get_name()?);
    println!("  Age: {}\n", person.get_age()?);

    // Demonstrate property changes
    println!("Making changes...\n");

    // Change name
    person.set_name("Alice Johnson".to_string())?;

    // Age up a few times
    person.celebrate_birthday()?;
    person.celebrate_birthday()?;
    person.celebrate_birthday()?;

    // Change to milestone age
    person.set_age(21)?;

    // Direct property access
    println!("\nDirect property access:");
    let simple_property = ObservableProperty::new(100);
    
    simple_property.subscribe(Arc::new(|old, new| {
        println!("💰 Value changed: {} → {}", old, new);
    }))?;

    simple_property.set(200)?;
    simple_property.set(150)?;

    // Cleanup observers
    person.name.unsubscribe(name_observer_id)?;
    person.age.unsubscribe(age_observer_id)?;
    person.age.unsubscribe(milestone_observer_id)?;

    println!("\n✅ Example completed successfully!");
    Ok(())
}

examples/multithreaded.rs (line 182)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Multithreaded Example ===\n");

    // Shared counter with multiple observers
    let counter = Arc::new(ObservableProperty::new(0i32));
    let notification_count = Arc::new(AtomicUsize::new(0));

    // Add observers that track notifications
    let count_clone = notification_count.clone();
    let observer1_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔴 Observer 1: {} → {}", old, new);
    }))?;

    let count_clone = notification_count.clone();
    let observer2_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔵 Observer 2: {} → {}", old, new);
    }))?;

    // Filtered observer for even numbers only
    let count_clone = notification_count.clone();
    let even_observer_id = counter.subscribe_filtered(
        Arc::new(move |old, new| {
            count_clone.fetch_add(1, Ordering::SeqCst);
            println!("🟢 Even Observer: {} → {} (new value is even!)", old, new);
        }),
        |_old, new| new % 2 == 0
    )?;

    println!("📊 Starting concurrent updates...\n");

    // Test 1: Multiple threads updating the counter
    {
        let num_threads = 4;
        let updates_per_thread = 5;
        let barrier = Arc::new(Barrier::new(num_threads));

        let handles: Vec<_> = (0..num_threads).map(|thread_id| {
            let counter_clone = counter.clone();
            let barrier_clone = barrier.clone();

            thread::spawn(move || {
                // Wait for all threads to be ready
                barrier_clone.wait();

                for i in 0..updates_per_thread {
                    let new_value = (thread_id * 100 + i) as i32;
                    
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Thread {} failed to set counter: {}", thread_id, e);
                    } else {
                        println!("🧵 Thread {} set counter to {}", thread_id, new_value);
                    }

                    // Small delay to see interleaving
                    thread::sleep(Duration::from_millis(10));
                }
            })
        }).collect();

        // Wait for all threads to complete
        for handle in handles {
            if let Err(e) = handle.join() {
                eprintln!("Thread panicked: {:?}", e);
            }
        }
    }

    thread::sleep(Duration::from_millis(100)); // Let notifications settle

    println!("\n📈 Current counter value: {}", counter.get()?);
    println!("🔔 Total notifications sent: {}\n", notification_count.load(Ordering::SeqCst));

    // Test 2: Performance comparison between sync and async notifications
    println!("⚡ Performance Test: Sync vs Async notifications\n");

    // Create a property with a slow observer
    let perf_counter = ObservableProperty::new(0i32);
    let slow_notifications = Arc::new(AtomicUsize::new(0));
    let slow_count_clone = slow_notifications.clone();

    perf_counter.subscribe(Arc::new(move |_old, _new| {
        // Simulate slow observer work
        thread::sleep(Duration::from_millis(20));
        slow_count_clone.fetch_add(1, Ordering::SeqCst);
    }))?;

    // Test synchronous updates (will block)
    let start = Instant::now();
    for i in 1..=3 {
        perf_counter.set(i)?;
    }
    let sync_duration = start.elapsed();
    println!("🐌 Sync updates (3 changes): {:?}", sync_duration);

    // Test asynchronous updates (won't block)
    let start = Instant::now();
    for i in 4..=6 {
        perf_counter.set_async(i)?;
    }
    let async_duration = start.elapsed();
    println!("🚀 Async updates (3 changes): {:?}", async_duration);

    println!("⚡ Speedup: {:.1}x faster", 
             sync_duration.as_secs_f64() / async_duration.as_secs_f64());

    // Wait for async observers to complete
    thread::sleep(Duration::from_millis(100));
    println!("🔔 Slow observer notifications: {}\n", slow_notifications.load(Ordering::SeqCst));

    // Test 3: Concurrent reads while updating
    println!("📚 Concurrent reads test...\n");
    {
        let read_counter = Arc::new(ObservableProperty::new(1000i32));
        let read_count = Arc::new(AtomicUsize::new(0));
        let num_readers = 8;
        let reads_per_reader = 50;

        // Start reader threads
        let reader_handles: Vec<_> = (0..num_readers).map(|reader_id| {
            let counter_clone = read_counter.clone();
            let count_clone = read_count.clone();

            thread::spawn(move || {
                for _ in 0..reads_per_reader {
                    match counter_clone.get() {
                        Ok(value) => {
                            count_clone.fetch_add(1, Ordering::SeqCst);
                            if value % 100 == 0 {
                                println!("📖 Reader {} saw value: {}", reader_id, value);
                            }
                        }
                        Err(e) => eprintln!("Reader {} error: {}", reader_id, e),
                    }
                    thread::sleep(Duration::from_millis(1));
                }
            })
        }).collect();

        // Start writer thread
        let writer_handle = {
            let counter_clone = read_counter.clone();
            thread::spawn(move || {
                for i in 1..=20 {
                    let new_value = 1000 + i * 50;
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Writer error: {}", e);
                    } else {
                        println!("✍️  Writer set value to: {}", new_value);
                    }
                    thread::sleep(Duration::from_millis(25));
                }
            })
        };

        // Wait for all threads
        for handle in reader_handles {
            if let Err(e) = handle.join() {
                eprintln!("Reader thread panicked: {:?}", e);
            }
        }

        if let Err(e) = writer_handle.join() {
            eprintln!("Writer thread panicked: {:?}", e);
        }

        println!("📊 Total successful reads: {}", read_count.load(Ordering::SeqCst));
    }

    // Cleanup
    counter.unsubscribe(observer1_id)?;
    counter.unsubscribe(observer2_id)?;
    counter.unsubscribe(even_observer_id)?;

    println!("\n✅ Multithreaded example completed successfully!");
    Ok(())
}

pub fn subscribe_filtered<F>( &self, observer: Observer<T>, filter: F, ) -> Result<ObserverId, PropertyError>

where F: Fn(&T, &T) -> bool + Send + Sync + 'static,

Subscribes an observer that only gets called when a filter condition is met

This is useful for observing only specific types of changes, such as when a value increases or crosses a threshold.

Arguments

• observer - The observer function to call when the filter passes
• filter - A predicate function that receives (old_value, new_value) and returns bool

Returns

Ok(ObserverId) for the filtered observer, or Err(PropertyError) if the lock is poisoned.

Examples

use observable_property::ObservableProperty;
use std::sync::Arc;

let property = ObservableProperty::new(0);

// Only notify when value increases
let id = property.subscribe_filtered(
    Arc::new(|old, new| println!("Value increased: {} -> {}", old, new)),
    |old, new| new > old
).unwrap();

property.set(10).unwrap(); // Triggers observer (0 -> 10)
property.set(5).unwrap();  // Does NOT trigger observer (10 -> 5)
property.set(15).unwrap(); // Triggers observer (5 -> 15)

Examples found in repository

examples/basic.rs (lines 62-73)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Basic Example ===\n");

    // Create a person
    let person = Person::new("Alice".to_string(), 25);

    // Subscribe to name changes
    let name_observer_id = person.name.subscribe(Arc::new(|old_name, new_name| {
        println!("📝 Name changed: '{}' → '{}'", old_name, new_name);
    }))?;

    // Subscribe to age changes
    let age_observer_id = person.age.subscribe(Arc::new(|old_age, new_age| {
        println!("🎂 Age changed: {} → {}", old_age, new_age);
    }))?;

    // Subscribe to only significant age changes (milestones)
    let milestone_observer_id = person.age.subscribe_filtered(
        Arc::new(|_old_age, new_age| {
            println!("🎉 Milestone reached! {} is now a special age: {}", 
                    if *new_age >= 18 { "Adult" } else { "Child" }, new_age);
        }),
        |old_age, new_age| {
            // Notify on milestone ages
            let milestones = [18, 21, 30, 40, 50, 65];
            milestones.contains(new_age) || 
            (milestones.contains(old_age) && !milestones.contains(new_age))
        }
    )?;

    println!("Initial state:");
    println!("  Name: {}", person.get_name()?);
    println!("  Age: {}\n", person.get_age()?);

    // Demonstrate property changes
    println!("Making changes...\n");

    // Change name
    person.set_name("Alice Johnson".to_string())?;

    // Age up a few times
    person.celebrate_birthday()?;
    person.celebrate_birthday()?;
    person.celebrate_birthday()?;

    // Change to milestone age
    person.set_age(21)?;

    // Direct property access
    println!("\nDirect property access:");
    let simple_property = ObservableProperty::new(100);
    
    simple_property.subscribe(Arc::new(|old, new| {
        println!("💰 Value changed: {} → {}", old, new);
    }))?;

    simple_property.set(200)?;
    simple_property.set(150)?;

    // Cleanup observers
    person.name.unsubscribe(name_observer_id)?;
    person.age.unsubscribe(age_observer_id)?;
    person.age.unsubscribe(milestone_observer_id)?;

    println!("\n✅ Example completed successfully!");
    Ok(())
}

examples/multithreaded.rs (lines 33-39)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("=== Observable Property Multithreaded Example ===\n");

    // Shared counter with multiple observers
    let counter = Arc::new(ObservableProperty::new(0i32));
    let notification_count = Arc::new(AtomicUsize::new(0));

    // Add observers that track notifications
    let count_clone = notification_count.clone();
    let observer1_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔴 Observer 1: {} → {}", old, new);
    }))?;

    let count_clone = notification_count.clone();
    let observer2_id = counter.subscribe(Arc::new(move |old, new| {
        count_clone.fetch_add(1, Ordering::SeqCst);
        println!("🔵 Observer 2: {} → {}", old, new);
    }))?;

    // Filtered observer for even numbers only
    let count_clone = notification_count.clone();
    let even_observer_id = counter.subscribe_filtered(
        Arc::new(move |old, new| {
            count_clone.fetch_add(1, Ordering::SeqCst);
            println!("🟢 Even Observer: {} → {} (new value is even!)", old, new);
        }),
        |_old, new| new % 2 == 0
    )?;

    println!("📊 Starting concurrent updates...\n");

    // Test 1: Multiple threads updating the counter
    {
        let num_threads = 4;
        let updates_per_thread = 5;
        let barrier = Arc::new(Barrier::new(num_threads));

        let handles: Vec<_> = (0..num_threads).map(|thread_id| {
            let counter_clone = counter.clone();
            let barrier_clone = barrier.clone();

            thread::spawn(move || {
                // Wait for all threads to be ready
                barrier_clone.wait();

                for i in 0..updates_per_thread {
                    let new_value = (thread_id * 100 + i) as i32;
                    
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Thread {} failed to set counter: {}", thread_id, e);
                    } else {
                        println!("🧵 Thread {} set counter to {}", thread_id, new_value);
                    }

                    // Small delay to see interleaving
                    thread::sleep(Duration::from_millis(10));
                }
            })
        }).collect();

        // Wait for all threads to complete
        for handle in handles {
            if let Err(e) = handle.join() {
                eprintln!("Thread panicked: {:?}", e);
            }
        }
    }

    thread::sleep(Duration::from_millis(100)); // Let notifications settle

    println!("\n📈 Current counter value: {}", counter.get()?);
    println!("🔔 Total notifications sent: {}\n", notification_count.load(Ordering::SeqCst));

    // Test 2: Performance comparison between sync and async notifications
    println!("⚡ Performance Test: Sync vs Async notifications\n");

    // Create a property with a slow observer
    let perf_counter = ObservableProperty::new(0i32);
    let slow_notifications = Arc::new(AtomicUsize::new(0));
    let slow_count_clone = slow_notifications.clone();

    perf_counter.subscribe(Arc::new(move |_old, _new| {
        // Simulate slow observer work
        thread::sleep(Duration::from_millis(20));
        slow_count_clone.fetch_add(1, Ordering::SeqCst);
    }))?;

    // Test synchronous updates (will block)
    let start = Instant::now();
    for i in 1..=3 {
        perf_counter.set(i)?;
    }
    let sync_duration = start.elapsed();
    println!("🐌 Sync updates (3 changes): {:?}", sync_duration);

    // Test asynchronous updates (won't block)
    let start = Instant::now();
    for i in 4..=6 {
        perf_counter.set_async(i)?;
    }
    let async_duration = start.elapsed();
    println!("🚀 Async updates (3 changes): {:?}", async_duration);

    println!("⚡ Speedup: {:.1}x faster", 
             sync_duration.as_secs_f64() / async_duration.as_secs_f64());

    // Wait for async observers to complete
    thread::sleep(Duration::from_millis(100));
    println!("🔔 Slow observer notifications: {}\n", slow_notifications.load(Ordering::SeqCst));

    // Test 3: Concurrent reads while updating
    println!("📚 Concurrent reads test...\n");
    {
        let read_counter = Arc::new(ObservableProperty::new(1000i32));
        let read_count = Arc::new(AtomicUsize::new(0));
        let num_readers = 8;
        let reads_per_reader = 50;

        // Start reader threads
        let reader_handles: Vec<_> = (0..num_readers).map(|reader_id| {
            let counter_clone = read_counter.clone();
            let count_clone = read_count.clone();

            thread::spawn(move || {
                for _ in 0..reads_per_reader {
                    match counter_clone.get() {
                        Ok(value) => {
                            count_clone.fetch_add(1, Ordering::SeqCst);
                            if value % 100 == 0 {
                                println!("📖 Reader {} saw value: {}", reader_id, value);
                            }
                        }
                        Err(e) => eprintln!("Reader {} error: {}", reader_id, e),
                    }
                    thread::sleep(Duration::from_millis(1));
                }
            })
        }).collect();

        // Start writer thread
        let writer_handle = {
            let counter_clone = read_counter.clone();
            thread::spawn(move || {
                for i in 1..=20 {
                    let new_value = 1000 + i * 50;
                    if let Err(e) = counter_clone.set(new_value) {
                        eprintln!("Writer error: {}", e);
                    } else {
                        println!("✍️  Writer set value to: {}", new_value);
                    }
                    thread::sleep(Duration::from_millis(25));
                }
            })
        };

        // Wait for all threads
        for handle in reader_handles {
            if let Err(e) = handle.join() {
                eprintln!("Reader thread panicked: {:?}", e);
            }
        }

        if let Err(e) = writer_handle.join() {
            eprintln!("Writer thread panicked: {:?}", e);
        }

        println!("📊 Total successful reads: {}", read_count.load(Ordering::SeqCst));
    }

    // Cleanup
    counter.unsubscribe(observer1_id)?;
    counter.unsubscribe(observer2_id)?;
    counter.unsubscribe(even_observer_id)?;

    println!("\n✅ Multithreaded example completed successfully!");
    Ok(())
}

Trait Implementations

impl<T: Clone> Clone for ObservableProperty<T>

fn clone(&self) -> Self

Creates a new reference to the same observable property

This creates a new ObservableProperty instance that shares the same underlying data with the original. Changes made through either instance will be visible to observers subscribed through both instances.

Examples

use observable_property::ObservableProperty;
use std::sync::Arc;

let property1 = ObservableProperty::new(42);
let property2 = property1.clone();

property2.subscribe(Arc::new(|old, new| {
    println!("Observer on property2 saw change: {} -> {}", old, new);
})).unwrap();

// This change through property1 will trigger the observer on property2
property1.set(100).unwrap();

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T: Clone + Debug + Send + Sync + 'static> Debug for ObservableProperty<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Debug implementation that shows the current value if accessible