use std::collections::HashMap;
use std::fmt;
use std::mem;
use std::panic;
use std::sync::{Arc, Mutex, RwLock};
use std::thread;
use std::time::{Duration, Instant};
#[cfg(feature = "serde")]
use serde::{Deserialize, Serialize};
#[cfg(feature = "debug")]
use backtrace::Backtrace;
#[cfg(feature = "async")]
use std::pin::Pin;
#[cfg(feature = "async")]
use std::task::{Context, Poll};
const MAX_THREADS: usize = 4;
const MAX_OBSERVERS: usize = 10_000;
pub trait PropertyPersistence: Send + Sync + 'static {
type Value: Clone + Send + Sync + 'static;
fn load(&self) -> Result<Self::Value, Box<dyn std::error::Error + Send + Sync>>;
fn save(&self, value: &Self::Value) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;
}
#[derive(Debug, Clone)]
pub enum PropertyError {
ReadLockError {
context: String,
},
WriteLockError {
context: String,
},
ObserverNotFound {
id: usize,
},
PoisonedLock,
ObserverError {
reason: String,
},
ThreadPoolExhausted,
InvalidConfiguration {
reason: String,
},
PersistenceError {
reason: String,
},
NoHistory {
reason: String,
},
ValidationError {
reason: String,
},
}
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)
}
PropertyError::ThreadPoolExhausted => {
write!(f, "Thread pool is exhausted and cannot spawn more observers")
}
PropertyError::InvalidConfiguration { reason } => {
write!(f, "Invalid configuration: {}", reason)
}
PropertyError::PersistenceError { reason } => {
write!(f, "Persistence failed: {}", reason)
}
PropertyError::NoHistory { reason } => {
write!(f, "No history available: {}", reason)
}
PropertyError::ValidationError { reason } => {
write!(f, "Validation failed: {}", reason)
}
}
}
}
impl std::error::Error for PropertyError {}
pub type Observer<T> = Arc<dyn Fn(&T, &T) + Send + Sync>;
enum ObserverRef<T: Clone + Send + Sync + 'static> {
Strong(Observer<T>),
Weak(std::sync::Weak<dyn Fn(&T, &T) + Send + Sync>),
}
impl<T: Clone + Send + Sync + 'static> ObserverRef<T> {
fn try_call(&self) -> Option<Observer<T>> {
match self {
ObserverRef::Strong(arc) => Some(arc.clone()),
ObserverRef::Weak(weak) => weak.upgrade(),
}
}
}
pub type ObserverId = usize;
#[derive(Debug, Clone)]
pub struct PropertyMetrics {
pub total_changes: usize,
pub observer_calls: usize,
pub avg_notification_time: Duration,
}
pub struct Subscription<T: Clone + Send + Sync + 'static> {
inner: Arc<RwLock<InnerProperty<T>>>,
id: ObserverId,
}
impl<T: Clone + Send + Sync + 'static> std::fmt::Debug for Subscription<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Subscription")
.field("id", &self.id)
.field("inner", &"[ObservableProperty]")
.finish()
}
}
impl<T: Clone + Send + Sync + 'static> Drop for Subscription<T> {
fn drop(&mut self) {
match self.inner.write() {
Ok(mut guard) => {
guard.observers.remove(&self.id);
}
Err(poisoned) => {
let mut guard = poisoned.into_inner();
guard.observers.remove(&self.id);
}
}
}
}
#[cfg(feature = "async")]
pub trait Stream {
type Item;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
}
pub struct ObservableProperty<T>
where
T: Clone + Send + Sync + 'static,
{
inner: Arc<RwLock<InnerProperty<T>>>,
max_threads: usize,
max_observers: usize,
}
struct InnerProperty<T>
where
T: Clone + Send + Sync + 'static,
{
value: T,
observers: HashMap<ObserverId, ObserverRef<T>>,
next_id: ObserverId,
history: Option<Vec<T>>,
history_size: usize,
total_changes: usize,
observer_calls: usize,
notification_times: Vec<Duration>,
#[cfg(feature = "debug")]
debug_logging_enabled: bool,
#[cfg(feature = "debug")]
change_logs: Vec<ChangeLog>,
batch_depth: usize,
batch_initial_value: Option<T>,
eq_fn: Option<Arc<dyn Fn(&T, &T) -> bool + Send + Sync>>,
validator: Option<Arc<dyn Fn(&T) -> Result<(), String> + Send + Sync>>,
event_log: Option<Vec<PropertyEvent<T>>>,
event_log_size: usize,
}
#[cfg(feature = "debug")]
#[derive(Clone)]
struct ChangeLog {
timestamp: Instant,
old_value_repr: String,
new_value_repr: String,
backtrace: String,
thread_id: String,
}
#[derive(Clone, Debug)]
pub struct PropertyEvent<T: Clone> {
pub timestamp: Instant,
pub old_value: T,
pub new_value: T,
pub event_number: usize,
pub thread_id: String,
}
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,
history: None,
history_size: 0,
total_changes: 0,
observer_calls: 0,
notification_times: Vec::new(),
#[cfg(feature = "debug")]
debug_logging_enabled: false,
#[cfg(feature = "debug")]
change_logs: Vec::new(),
batch_depth: 0,
batch_initial_value: None,
eq_fn: None,
validator: None,
event_log: None,
event_log_size: 0,
})),
max_threads: MAX_THREADS,
max_observers: MAX_OBSERVERS,
}
}
pub fn with_equality<F>(initial_value: T, eq_fn: F) -> Self
where
F: Fn(&T, &T) -> bool + Send + Sync + 'static,
{
Self {
inner: Arc::new(RwLock::new(InnerProperty {
value: initial_value,
observers: HashMap::new(),
next_id: 0,
history: None,
history_size: 0,
total_changes: 0,
observer_calls: 0,
notification_times: Vec::new(),
#[cfg(feature = "debug")]
debug_logging_enabled: false,
#[cfg(feature = "debug")]
change_logs: Vec::new(),
batch_depth: 0,
batch_initial_value: None,
eq_fn: Some(Arc::new(eq_fn)),
validator: None,
event_log: None,
event_log_size: 0,
})),
max_threads: MAX_THREADS,
max_observers: MAX_OBSERVERS,
}
}
pub fn with_validator<F>(
initial_value: T,
validator: F,
) -> Result<Self, PropertyError>
where
F: Fn(&T) -> Result<(), String> + Send + Sync + 'static,
{
validator(&initial_value).map_err(|reason| PropertyError::ValidationError { reason })?;
Ok(Self {
inner: Arc::new(RwLock::new(InnerProperty {
value: initial_value,
observers: HashMap::new(),
next_id: 0,
history: None,
history_size: 0,
total_changes: 0,
observer_calls: 0,
notification_times: Vec::new(),
#[cfg(feature = "debug")]
debug_logging_enabled: false,
#[cfg(feature = "debug")]
change_logs: Vec::new(),
batch_depth: 0,
batch_initial_value: None,
eq_fn: None,
validator: Some(Arc::new(validator)),
event_log: None,
event_log_size: 0,
})),
max_threads: MAX_THREADS,
max_observers: MAX_OBSERVERS,
})
}
pub fn with_max_threads(initial_value: T, max_threads: usize) -> Self {
let max_threads = if max_threads == 0 {
MAX_THREADS
} else {
max_threads
};
Self {
inner: Arc::new(RwLock::new(InnerProperty {
value: initial_value,
observers: HashMap::new(),
next_id: 0,
history: None,
history_size: 0,
total_changes: 0,
observer_calls: 0,
notification_times: Vec::new(),
#[cfg(feature = "debug")]
debug_logging_enabled: false,
#[cfg(feature = "debug")]
change_logs: Vec::new(),
batch_depth: 0,
batch_initial_value: None,
eq_fn: None,
validator: None,
event_log: None,
event_log_size: 0,
})),
max_threads,
max_observers: MAX_OBSERVERS,
}
}
pub fn with_persistence<P>(initial_value: T, persistence: P) -> Self
where
P: PropertyPersistence<Value = T>,
{
let value = persistence.load().unwrap_or_else(|e| {
eprintln!(
"Failed to load persisted value, using initial value: {}",
e
);
initial_value.clone()
});
let property = Self::new(value);
let persistence = Arc::new(persistence);
if let Err(e) = property.subscribe(Arc::new(move |_old, new| {
if let Err(save_err) = persistence.save(new) {
eprintln!("Failed to persist property value: {}", save_err);
}
})) {
eprintln!("Failed to subscribe persistence observer: {}", e);
}
property
}
pub fn with_history(initial_value: T, history_size: usize) -> Self {
Self {
inner: Arc::new(RwLock::new(InnerProperty {
value: initial_value,
observers: HashMap::new(),
next_id: 0,
history: if history_size > 0 {
Some(Vec::with_capacity(history_size))
} else {
None
},
history_size,
total_changes: 0,
observer_calls: 0,
notification_times: Vec::new(),
#[cfg(feature = "debug")]
debug_logging_enabled: false,
#[cfg(feature = "debug")]
change_logs: Vec::new(),
batch_depth: 0,
batch_initial_value: None,
eq_fn: None,
validator: None,
event_log: None,
event_log_size: 0,
})),
max_threads: MAX_THREADS,
max_observers: MAX_OBSERVERS,
}
}
pub fn with_event_log(initial_value: T, event_log_size: usize) -> Self {
Self {
inner: Arc::new(RwLock::new(InnerProperty {
value: initial_value,
observers: HashMap::new(),
next_id: 0,
history: None,
history_size: 0,
total_changes: 0,
observer_calls: 0,
notification_times: Vec::new(),
#[cfg(feature = "debug")]
debug_logging_enabled: false,
#[cfg(feature = "debug")]
change_logs: Vec::new(),
batch_depth: 0,
batch_initial_value: None,
eq_fn: None,
validator: None,
event_log: Some(Vec::with_capacity(if event_log_size > 0 { event_log_size } else { 16 })),
event_log_size,
})),
max_threads: MAX_THREADS,
max_observers: MAX_OBSERVERS,
}
}
pub fn undo(&self) -> Result<(), PropertyError> {
let (old_value, new_value, observers_snapshot, dead_observer_ids) = {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
let validator = prop.validator.clone();
let history = prop.history.as_mut().ok_or_else(|| PropertyError::NoHistory {
reason: "History tracking is not enabled for this property".to_string(),
})?;
if history.is_empty() {
return Err(PropertyError::NoHistory {
reason: "No history available to undo".to_string(),
});
}
let previous_value = history.pop().unwrap();
if let Some(validator) = validator {
validator(&previous_value).map_err(|reason| {
history.push(previous_value.clone());
PropertyError::ValidationError {
reason: format!("Cannot undo to invalid historical value: {}", reason)
}
})?;
}
let old_value = mem::replace(&mut prop.value, previous_value.clone());
let mut observers_snapshot = Vec::new();
let mut dead_ids = Vec::new();
for (id, observer_ref) in &prop.observers {
if let Some(observer) = observer_ref.try_call() {
observers_snapshot.push(observer);
} else {
dead_ids.push(*id);
}
}
(old_value, previous_value, observers_snapshot, dead_ids)
};
for observer in observers_snapshot {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_value, &new_value);
})) {
eprintln!("Observer panic during undo: {:?}", e);
}
}
if !dead_observer_ids.is_empty() {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
for id in dead_observer_ids {
prop.observers.remove(&id);
}
}
Ok(())
}
pub fn get_history(&self) -> Vec<T> {
match self.inner.read() {
Ok(prop) => prop.history.as_ref().map_or(Vec::new(), |h| h.clone()),
Err(poisoned) => {
let prop = poisoned.into_inner();
prop.history.as_ref().map_or(Vec::new(), |h| h.clone())
}
}
}
pub fn get_event_log(&self) -> Vec<PropertyEvent<T>> {
match self.inner.read() {
Ok(prop) => prop.event_log.as_ref().map_or(Vec::new(), |log| log.clone()),
Err(poisoned) => {
let prop = poisoned.into_inner();
prop.event_log.as_ref().map_or(Vec::new(), |log| log.clone())
}
}
}
pub fn get(&self) -> Result<T, PropertyError> {
match self.inner.read() {
Ok(prop) => Ok(prop.value.clone()),
Err(poisoned) => {
Ok(poisoned.into_inner().value.clone())
}
}
}
pub fn get_metrics(&self) -> Result<PropertyMetrics, PropertyError> {
match self.inner.read() {
Ok(prop) => {
let avg_notification_time = if prop.notification_times.is_empty() {
Duration::from_secs(0)
} else {
let total: Duration = prop.notification_times.iter().sum();
total / prop.notification_times.len() as u32
};
Ok(PropertyMetrics {
total_changes: prop.total_changes,
observer_calls: prop.observer_calls,
avg_notification_time,
})
}
Err(poisoned) => {
let prop = poisoned.into_inner();
let avg_notification_time = if prop.notification_times.is_empty() {
Duration::from_secs(0)
} else {
let total: Duration = prop.notification_times.iter().sum();
total / prop.notification_times.len() as u32
};
Ok(PropertyMetrics {
total_changes: prop.total_changes,
observer_calls: prop.observer_calls,
avg_notification_time,
})
}
}
}
pub fn set(&self, new_value: T) -> Result<(), PropertyError> {
{
let prop = match self.inner.read() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
if let Some(validator) = &prop.validator {
validator(&new_value).map_err(|reason| PropertyError::ValidationError { reason })?;
}
}
let notification_start = Instant::now();
let (old_value, observers_snapshot, dead_observer_ids, in_batch) = {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => {
poisoned.into_inner()
}
};
let values_equal = if let Some(eq_fn) = &prop.eq_fn {
eq_fn(&prop.value, &new_value)
} else {
false };
if values_equal {
return Ok(());
}
let in_batch = prop.batch_depth > 0;
let old_value = mem::replace(&mut prop.value, new_value.clone());
prop.total_changes += 1;
let event_num = prop.total_changes - 1;
let history_size = prop.history_size;
if let Some(history) = &mut prop.history {
history.push(old_value.clone());
if history.len() > history_size {
let overflow = history.len() - history_size;
history.drain(0..overflow);
}
}
let event_log_size = prop.event_log_size;
if let Some(event_log) = &mut prop.event_log {
let event = PropertyEvent {
timestamp: Instant::now(),
old_value: old_value.clone(),
new_value: new_value.clone(),
event_number: event_num, thread_id: format!("{:?}", thread::current().id()),
};
event_log.push(event);
if event_log_size > 0 && event_log.len() > event_log_size {
let overflow = event_log.len() - event_log_size;
event_log.drain(0..overflow);
}
}
let mut observers_snapshot = Vec::new();
let mut dead_ids = Vec::new();
if !in_batch {
for (id, observer_ref) in &prop.observers {
if let Some(observer) = observer_ref.try_call() {
observers_snapshot.push(observer);
} else {
dead_ids.push(*id);
}
}
}
(old_value, observers_snapshot, dead_ids, in_batch)
};
if in_batch {
return Ok(());
}
let observer_count = observers_snapshot.len();
for observer in observers_snapshot {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_value, &new_value);
})) {
eprintln!("Observer panic: {:?}", e);
}
}
let notification_time = notification_start.elapsed();
{
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
prop.observer_calls += observer_count;
prop.notification_times.push(notification_time);
}
if !dead_observer_ids.is_empty() {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
for id in dead_observer_ids {
prop.observers.remove(&id);
}
}
Ok(())
}
pub fn set_async(&self, new_value: T) -> Result<(), PropertyError> {
{
let prop = match self.inner.read() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
if let Some(validator) = &prop.validator {
validator(&new_value).map_err(|reason| PropertyError::ValidationError { reason })?;
}
}
let notification_start = Instant::now();
let (old_value, observers_snapshot, dead_observer_ids, in_batch) = {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => {
poisoned.into_inner()
}
};
let values_equal = if let Some(eq_fn) = &prop.eq_fn {
eq_fn(&prop.value, &new_value)
} else {
false };
if values_equal {
return Ok(());
}
let in_batch = prop.batch_depth > 0;
let old_value = mem::replace(&mut prop.value, new_value.clone());
prop.total_changes += 1;
let event_num = prop.total_changes - 1;
let history_size = prop.history_size;
if let Some(history) = &mut prop.history {
history.push(old_value.clone());
if history.len() > history_size {
let overflow = history.len() - history_size;
history.drain(0..overflow);
}
}
let event_log_size = prop.event_log_size;
if let Some(event_log) = &mut prop.event_log {
let event = PropertyEvent {
timestamp: Instant::now(),
old_value: old_value.clone(),
new_value: new_value.clone(),
event_number: event_num, thread_id: format!("{:?}", thread::current().id()),
};
event_log.push(event);
if event_log_size > 0 && event_log.len() > event_log_size {
let overflow = event_log.len() - event_log_size;
event_log.drain(0..overflow);
}
}
let mut observers_snapshot = Vec::new();
let mut dead_ids = Vec::new();
if !in_batch {
for (id, observer_ref) in &prop.observers {
if let Some(observer) = observer_ref.try_call() {
observers_snapshot.push(observer);
} else {
dead_ids.push(*id);
}
}
}
(old_value, observers_snapshot, dead_ids, in_batch)
};
if in_batch {
return Ok(());
}
if observers_snapshot.is_empty() {
if !dead_observer_ids.is_empty() {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
for id in dead_observer_ids {
prop.observers.remove(&id);
}
}
return Ok(());
}
let observers_per_thread = observers_snapshot.len().div_ceil(self.max_threads);
let observer_count = observers_snapshot.len();
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);
}
}
});
}
let notification_time = notification_start.elapsed();
{
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
prop.observer_calls += observer_count;
prop.notification_times.push(notification_time);
}
if !dead_observer_ids.is_empty() {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
for id in dead_observer_ids {
prop.observers.remove(&id);
}
}
Ok(())
}
pub fn begin_update(&self) -> Result<(), PropertyError> {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
if prop.batch_depth == 0 {
prop.batch_initial_value = Some(prop.value.clone());
}
prop.batch_depth += 1;
Ok(())
}
pub fn end_update(&self) -> Result<(), PropertyError> {
let notification_start = Instant::now();
let (should_notify, old_value, new_value, observers_snapshot, dead_observer_ids) = {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
if prop.batch_depth == 0 {
return Err(PropertyError::InvalidConfiguration {
reason: "end_update() called without matching begin_update()".to_string(),
});
}
prop.batch_depth -= 1;
if prop.batch_depth == 0 {
if let Some(initial_value) = prop.batch_initial_value.take() {
let current_value = prop.value.clone();
let mut observers_snapshot = Vec::new();
let mut dead_ids = Vec::new();
for (id, observer_ref) in &prop.observers {
if let Some(observer) = observer_ref.try_call() {
observers_snapshot.push(observer);
} else {
dead_ids.push(*id);
}
}
(true, initial_value, current_value, observers_snapshot, dead_ids)
} else {
(false, prop.value.clone(), prop.value.clone(), Vec::new(), Vec::new())
}
} else {
(false, prop.value.clone(), prop.value.clone(), Vec::new(), Vec::new())
}
};
if should_notify && !observers_snapshot.is_empty() {
let observer_count = observers_snapshot.len();
for observer in observers_snapshot {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_value, &new_value);
})) {
eprintln!("Observer panic: {:?}", e);
}
}
let notification_time = notification_start.elapsed();
{
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
prop.observer_calls += observer_count;
prop.notification_times.push(notification_time);
}
if !dead_observer_ids.is_empty() {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
for id in dead_observer_ids {
prop.observers.remove(&id);
}
}
}
Ok(())
}
pub fn subscribe(&self, observer: Observer<T>) -> Result<ObserverId, PropertyError> {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => {
poisoned.into_inner()
}
};
if prop.observers.len() >= self.max_observers {
return Err(PropertyError::InvalidConfiguration {
reason: format!(
"Maximum observer limit ({}) exceeded. Current observers: {}. \
Consider unsubscribing unused observers to free resources.",
self.max_observers,
prop.observers.len()
),
});
}
let id = prop.next_id;
prop.next_id = prop.next_id.wrapping_add(1);
prop.observers.insert(id, ObserverRef::Strong(observer));
Ok(id)
}
pub fn unsubscribe(&self, id: ObserverId) -> Result<bool, PropertyError> {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => {
poisoned.into_inner()
}
};
let was_present = prop.observers.remove(&id).is_some();
Ok(was_present)
}
pub fn subscribe_weak(
&self,
observer: std::sync::Weak<dyn Fn(&T, &T) + Send + Sync>,
) -> Result<ObserverId, PropertyError> {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => {
poisoned.into_inner()
}
};
if prop.observers.len() >= self.max_observers {
return Err(PropertyError::InvalidConfiguration {
reason: format!(
"Maximum observer limit ({}) exceeded. Current observers: {}. \
Consider unsubscribing unused observers to free resources.",
self.max_observers,
prop.observers.len()
),
});
}
let id = prop.next_id;
prop.next_id = prop.next_id.wrapping_add(1);
prop.observers.insert(id, ObserverRef::Weak(observer));
Ok(id)
}
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)
}
pub fn subscribe_debounced(
&self,
observer: Observer<T>,
debounce_duration: Duration,
) -> Result<ObserverId, PropertyError> {
let last_change_time = Arc::new(Mutex::new(Instant::now()));
let pending_values = Arc::new(Mutex::new(None::<(T, T)>));
let debounced_observer = Arc::new(move |old_val: &T, new_val: &T| {
{
let mut last_time = last_change_time.lock().unwrap();
*last_time = Instant::now();
let mut pending = pending_values.lock().unwrap();
*pending = Some((old_val.clone(), new_val.clone()));
}
let last_change_time_thread = last_change_time.clone();
let pending_values_thread = pending_values.clone();
let observer_thread = observer.clone();
let duration = debounce_duration;
thread::spawn(move || {
thread::sleep(duration);
let should_notify = {
let last_time = last_change_time_thread.lock().unwrap();
last_time.elapsed() >= duration
};
if should_notify {
let values = {
let mut pending = pending_values_thread.lock().unwrap();
pending.take()
};
if let Some((old, new)) = values {
let _ = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer_thread(&old, &new);
}));
}
}
});
});
self.subscribe(debounced_observer)
}
pub fn subscribe_throttled(
&self,
observer: Observer<T>,
throttle_interval: Duration,
) -> Result<ObserverId, PropertyError> {
let last_notify_time = Arc::new(Mutex::new(None::<Instant>));
let pending_notification = Arc::new(Mutex::new(None::<(T, T)>));
let throttled_observer = Arc::new(move |old_val: &T, new_val: &T| {
let should_notify_now = {
let last_time = last_notify_time.lock().unwrap();
match *last_time {
None => true, Some(last) => last.elapsed() >= throttle_interval,
}
};
if should_notify_now {
{
let mut last_time = last_notify_time.lock().unwrap();
*last_time = Some(Instant::now());
}
let _ = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(old_val, new_val);
}));
} else {
{
let mut pending = pending_notification.lock().unwrap();
*pending = Some((old_val.clone(), new_val.clone()));
}
let last_notify_time_thread = last_notify_time.clone();
let pending_notification_thread = pending_notification.clone();
let observer_thread = observer.clone();
let interval = throttle_interval;
thread::spawn(move || {
let wait_duration = {
let last_time = last_notify_time_thread.lock().unwrap();
if let Some(last) = *last_time {
let elapsed = last.elapsed();
if elapsed < interval {
interval - elapsed
} else {
Duration::from_millis(0)
}
} else {
Duration::from_millis(0)
}
};
if wait_duration > Duration::from_millis(0) {
thread::sleep(wait_duration);
}
let should_notify = {
let last_time = last_notify_time_thread.lock().unwrap();
match *last_time {
Some(last) => last.elapsed() >= interval,
None => true,
}
};
if should_notify {
let values = {
let mut pending = pending_notification_thread.lock().unwrap();
pending.take()
};
if let Some((old, new)) = values {
{
let mut last_time = last_notify_time_thread.lock().unwrap();
*last_time = Some(Instant::now());
}
let _ = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer_thread(&old, &new);
}));
}
}
});
}
});
self.subscribe(throttled_observer)
}
pub fn notify_observers_batch(&self, changes: Vec<(T, T)>) -> Result<(), PropertyError> {
let (observers_snapshot, dead_observer_ids) = {
let prop = match self.inner.read() {
Ok(guard) => guard,
Err(poisoned) => {
poisoned.into_inner()
}
};
let mut observers = Vec::new();
let mut dead_ids = Vec::new();
for (id, observer_ref) in &prop.observers {
if let Some(observer) = observer_ref.try_call() {
observers.push(observer);
} else {
dead_ids.push(*id);
}
}
(observers, dead_ids)
};
for (old_val, new_val) in changes {
for observer in &observers_snapshot {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_val, &new_val);
})) {
eprintln!("Observer panic in batch notification: {:?}", e);
}
}
}
if !dead_observer_ids.is_empty() {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
for id in dead_observer_ids {
prop.observers.remove(&id);
}
}
Ok(())
}
pub fn subscribe_with_subscription(
&self,
observer: Observer<T>,
) -> Result<Subscription<T>, PropertyError> {
let id = self.subscribe(observer)?;
Ok(Subscription {
inner: Arc::clone(&self.inner),
id,
})
}
pub fn subscribe_filtered_with_subscription<F>(
&self,
observer: Observer<T>,
filter: F,
) -> Result<Subscription<T>, PropertyError>
where
F: Fn(&T, &T) -> bool + Send + Sync + 'static,
{
let id = self.subscribe_filtered(observer, filter)?;
Ok(Subscription {
inner: Arc::clone(&self.inner),
id,
})
}
pub fn with_config(initial_value: T, max_threads: usize, max_observers: usize) -> Self {
Self {
inner: Arc::new(RwLock::new(InnerProperty {
value: initial_value,
observers: HashMap::new(),
next_id: 0,
history: None,
history_size: 0,
total_changes: 0,
observer_calls: 0,
notification_times: Vec::new(),
#[cfg(feature = "debug")]
debug_logging_enabled: false,
#[cfg(feature = "debug")]
change_logs: Vec::new(),
batch_depth: 0,
batch_initial_value: None,
eq_fn: None,
validator: None,
event_log: None,
event_log_size: 0,
})),
max_threads: if max_threads == 0 { MAX_THREADS } else { max_threads },
max_observers: if max_observers == 0 { MAX_OBSERVERS } else { max_observers },
}
}
pub fn observer_count(&self) -> usize {
match self.inner.read() {
Ok(prop) => prop.observers.len(),
Err(poisoned) => {
poisoned.into_inner().observers.len()
}
}
}
pub fn try_get(&self) -> Option<T> {
self.get().ok()
}
pub fn modify<F>(&self, f: F) -> Result<(), PropertyError>
where
F: FnOnce(&mut T),
{
let (old_value, new_value, observers_snapshot, dead_observer_ids) = {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => {
poisoned.into_inner()
}
};
let old_value = prop.value.clone();
f(&mut prop.value);
let new_value = prop.value.clone();
if let Some(validator) = &prop.validator {
validator(&new_value).map_err(|reason| {
prop.value = old_value.clone();
PropertyError::ValidationError { reason }
})?;
}
let values_equal = if let Some(eq_fn) = &prop.eq_fn {
eq_fn(&old_value, &new_value)
} else {
false };
if values_equal {
return Ok(());
}
prop.total_changes += 1;
let event_num = prop.total_changes - 1;
let history_size = prop.history_size;
if let Some(history) = &mut prop.history {
history.push(old_value.clone());
if history.len() > history_size {
let overflow = history.len() - history_size;
history.drain(0..overflow);
}
}
let event_log_size = prop.event_log_size;
if let Some(event_log) = &mut prop.event_log {
let event = PropertyEvent {
timestamp: Instant::now(),
old_value: old_value.clone(),
new_value: new_value.clone(),
event_number: event_num, thread_id: format!("{:?}", thread::current().id()),
};
event_log.push(event);
if event_log_size > 0 && event_log.len() > event_log_size {
let overflow = event_log.len() - event_log_size;
event_log.drain(0..overflow);
}
}
let mut observers = Vec::new();
let mut dead_ids = Vec::new();
for (id, observer_ref) in &prop.observers {
if let Some(observer) = observer_ref.try_call() {
observers.push(observer);
} else {
dead_ids.push(*id);
}
}
(old_value, new_value, observers, dead_ids)
};
for observer in observers_snapshot {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_value, &new_value);
})) {
eprintln!("Observer panic in modify: {:?}", e);
}
}
if !dead_observer_ids.is_empty() {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
for id in dead_observer_ids {
prop.observers.remove(&id);
}
}
Ok(())
}
pub fn map<U, F>(&self, transform: F) -> Result<ObservableProperty<U>, PropertyError>
where
U: Clone + Send + Sync + 'static,
F: Fn(&T) -> U + Send + Sync + 'static,
{
let initial_value = self.get()?;
let derived = ObservableProperty::new(transform(&initial_value));
let derived_clone = derived.clone();
let transform = Arc::new(transform);
self.subscribe(Arc::new(move |_old, new| {
let transformed = transform(new);
if let Err(e) = derived_clone.set(transformed) {
eprintln!("Failed to update derived property: {}", e);
}
}))?;
Ok(derived)
}
pub fn update_batch<F>(&self, f: F) -> Result<(), PropertyError>
where
F: FnOnce(&mut T) -> Vec<T>,
{
let (initial_value, intermediate_states, observers_snapshot, dead_observer_ids) = {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => {
poisoned.into_inner()
}
};
let initial_value = prop.value.clone();
let states = f(&mut prop.value);
if let Some(final_state) = states.last() {
prop.value = final_state.clone();
let history_size = prop.history_size;
if let Some(history) = &mut prop.history {
history.push(initial_value.clone());
if history.len() > history_size {
let overflow = history.len() - history_size;
history.drain(0..overflow);
}
}
} else {
prop.value = initial_value.clone();
}
let mut observers = Vec::new();
let mut dead_ids = Vec::new();
for (id, observer_ref) in &prop.observers {
if let Some(observer) = observer_ref.try_call() {
observers.push(observer);
} else {
dead_ids.push(*id);
}
}
(initial_value, states, observers, dead_ids)
};
if !intermediate_states.is_empty() {
let mut previous_state = initial_value;
for current_state in intermediate_states {
for observer in &observers_snapshot {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&previous_state, ¤t_state);
})) {
eprintln!("Observer panic in update_batch: {:?}", e);
}
}
previous_state = current_state;
}
}
if !dead_observer_ids.is_empty() {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
for id in dead_observer_ids {
prop.observers.remove(&id);
}
}
Ok(())
}
pub fn bind_bidirectional(
&self,
other: &ObservableProperty<T>,
) -> Result<(), PropertyError>
where
T: PartialEq,
{
let self_inner = Arc::clone(&self.inner);
other.subscribe(Arc::new(move |_old, new| {
let should_update = {
match self_inner.read() {
Ok(prop) => &prop.value != new,
Err(poisoned) => &poisoned.into_inner().value != new,
}
};
if should_update {
let mut prop = match self_inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
let old_value = mem::replace(&mut prop.value, new.clone());
let history_size = prop.history_size;
if let Some(history) = &mut prop.history {
history.push(old_value.clone());
if history.len() > history_size {
let overflow = history.len() - history_size;
history.drain(0..overflow);
}
}
let mut observers = Vec::new();
for (_id, observer_ref) in &prop.observers {
if let Some(observer) = observer_ref.try_call() {
observers.push(observer);
}
}
drop(prop);
for observer in observers {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_value, new);
})) {
eprintln!("Observer panic in bidirectional binding: {:?}", e);
}
}
}
}))?;
let other_inner = Arc::clone(&other.inner);
self.subscribe(Arc::new(move |_old, new| {
let should_update = {
match other_inner.read() {
Ok(prop) => &prop.value != new,
Err(poisoned) => &poisoned.into_inner().value != new,
}
};
if should_update {
let mut prop = match other_inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
let old_value = mem::replace(&mut prop.value, new.clone());
let history_size = prop.history_size;
if let Some(history) = &mut prop.history {
history.push(old_value.clone());
if history.len() > history_size {
let overflow = history.len() - history_size;
history.drain(0..overflow);
}
}
let mut observers = Vec::new();
for (_id, observer_ref) in &prop.observers {
if let Some(observer) = observer_ref.try_call() {
observers.push(observer);
}
}
drop(prop);
for observer in observers {
if let Err(e) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
observer(&old_value, new);
})) {
eprintln!("Observer panic in bidirectional binding: {:?}", e);
}
}
}
}))?;
Ok(())
}
#[cfg(feature = "async")]
pub fn to_stream(&self) -> PropertyStream<T> {
use std::sync::mpsc;
let (tx, rx) = mpsc::channel::<T>();
let inner = self.inner.clone();
let current_value = inner
.read()
.or_else(|poisoned| Ok::<_, ()>(poisoned.into_inner()))
.map(|prop| prop.value.clone())
.ok();
let subscription_id = self
.subscribe(Arc::new(move |_old, new| {
let _ = tx.send(new.clone());
}))
.ok();
PropertyStream {
rx,
current_value,
subscription_id,
property: self.inner.clone(),
waker: Arc::new(Mutex::new(None)),
}
}
}
#[cfg(feature = "async")]
pub struct PropertyStream<T>
where
T: Clone + Send + Sync + 'static,
{
rx: std::sync::mpsc::Receiver<T>,
current_value: Option<T>,
subscription_id: Option<ObserverId>,
property: Arc<RwLock<InnerProperty<T>>>,
waker: Arc<Mutex<Option<std::task::Waker>>>,
}
#[cfg(feature = "async")]
impl<T: Clone + Send + Sync + Unpin + 'static> Stream for PropertyStream<T> {
type Item = T;
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
let this = self.get_mut();
if let Some(value) = this.current_value.take() {
return Poll::Ready(Some(value));
}
match this.rx.try_recv() {
Ok(value) => Poll::Ready(Some(value)),
Err(std::sync::mpsc::TryRecvError::Empty) => {
if let Ok(mut waker_lock) = this.waker.lock() {
*waker_lock = Some(cx.waker().clone());
}
Poll::Pending
}
Err(std::sync::mpsc::TryRecvError::Disconnected) => Poll::Ready(None),
}
}
}
#[cfg(feature = "async")]
impl<T: Clone + Send + Sync + 'static> Drop for PropertyStream<T> {
fn drop(&mut self) {
if let Some(id) = self.subscription_id {
if let Ok(mut prop) = self.property.write().or_else(|poisoned| Ok::<_, ()>(poisoned.into_inner())) {
prop.observers.remove(&id);
}
}
}
}
#[cfg(feature = "async")]
impl<T: Clone + Send + Sync + Unpin + 'static> ObservableProperty<T> {
#[cfg(feature = "async")]
pub fn wait_for<F>(&self, predicate: F) -> WaitForFuture<T, F>
where
F: Fn(&T) -> bool + Send + Sync + Unpin + 'static,
{
WaitForFuture::new(self.inner.clone(), predicate, self)
}
}
#[cfg(feature = "async")]
pub struct WaitForFuture<T, F>
where
T: Clone + Send + Sync + Unpin + 'static,
F: Fn(&T) -> bool + Send + Sync + Unpin + 'static,
{
property: Arc<RwLock<InnerProperty<T>>>,
rx: std::sync::mpsc::Receiver<T>,
subscription_id: Option<ObserverId>,
result: Option<T>,
_phantom: std::marker::PhantomData<F>,
}
#[cfg(feature = "async")]
impl<T, F> WaitForFuture<T, F>
where
T: Clone + Send + Sync + Unpin + 'static,
F: Fn(&T) -> bool + Send + Sync + Unpin + 'static,
{
fn new<P>(property: Arc<RwLock<InnerProperty<T>>>, predicate: F, obs_property: &P) -> Self
where
P: HasInner<T>,
{
if let Ok(current) = property
.read()
.or_else(|poisoned| Ok::<_, ()>(poisoned.into_inner()))
.map(|prop| prop.value.clone())
{
if predicate(¤t) {
let (tx, rx) = std::sync::mpsc::channel();
let _ = tx.send(current.clone());
return Self {
property,
rx,
subscription_id: None,
result: Some(current),
_phantom: std::marker::PhantomData,
};
}
}
let predicate = Arc::new(predicate);
let (tx, rx) = std::sync::mpsc::channel();
let subscription_id = obs_property
.subscribe_internal(Arc::new(move |_old, new| {
if predicate(new) {
let _ = tx.send(new.clone());
}
}))
.ok();
Self {
property,
rx,
subscription_id,
result: None,
_phantom: std::marker::PhantomData,
}
}
}
#[cfg(feature = "async")]
impl<T, F> std::future::Future for WaitForFuture<T, F>
where
T: Clone + Send + Sync + Unpin + 'static,
F: Fn(&T) -> bool + Send + Sync + Unpin + 'static,
{
type Output = T;
fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut();
if let Some(result) = this.result.take() {
return Poll::Ready(result);
}
match this.rx.try_recv() {
Ok(value) => {
if let Some(id) = this.subscription_id.take() {
if let Ok(mut prop) = this.property.write().or_else(|poisoned| Ok::<_, ()>(poisoned.into_inner())) {
prop.observers.remove(&id);
}
}
Poll::Ready(value)
}
Err(std::sync::mpsc::TryRecvError::Empty) => Poll::Pending,
Err(std::sync::mpsc::TryRecvError::Disconnected) => {
let value = this
.property
.read()
.or_else(|poisoned| Ok::<_, ()>(poisoned.into_inner()))
.map(|prop| prop.value.clone())
.unwrap();
Poll::Ready(value)
}
}
}
}
#[cfg(feature = "async")]
impl<T, F> Drop for WaitForFuture<T, F>
where
T: Clone + Send + Sync + Unpin + 'static,
F: Fn(&T) -> bool + Send + Sync + Unpin + 'static,
{
fn drop(&mut self) {
if let Some(id) = self.subscription_id.take() {
if let Ok(mut prop) = self.property.write().or_else(|poisoned| Ok::<_, ()>(poisoned.into_inner())) {
prop.observers.remove(&id);
}
}
}
}
#[cfg(feature = "async")]
trait HasInner<T: Clone + Send + Sync + 'static> {
fn subscribe_internal(&self, observer: Observer<T>) -> Result<ObserverId, PropertyError>;
}
#[cfg(feature = "async")]
impl<T: Clone + Send + Sync + 'static> HasInner<T> for ObservableProperty<T> {
fn subscribe_internal(&self, observer: Observer<T>) -> Result<ObserverId, PropertyError> {
self.subscribe(observer)
}
}
#[cfg(feature = "debug")]
impl<T: Clone + Send + Sync + std::fmt::Debug + 'static> ObservableProperty<T> {
pub fn log_change(&self, old_value: &T, new_value: &T, label: &str) {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
if !prop.debug_logging_enabled {
return;
}
let backtrace = Backtrace::new();
let thread_id = format!("{:?}", thread::current().id());
let old_value_repr = format!("{:?}", old_value);
let new_value_repr = format!("{:?} ({})", new_value, label);
prop.change_logs.push(ChangeLog {
timestamp: Instant::now(),
old_value_repr,
new_value_repr,
backtrace: format!("{:?}", backtrace),
thread_id,
});
}
pub fn enable_change_logging(&self) {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
prop.debug_logging_enabled = true;
}
pub fn disable_change_logging(&self) {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
prop.debug_logging_enabled = false;
}
pub fn clear_change_logs(&self) {
let mut prop = match self.inner.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
prop.change_logs.clear();
}
pub fn get_change_logs(&self) -> Vec<String> {
let prop = match self.inner.read() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
if prop.change_logs.is_empty() {
return Vec::new();
}
let first_timestamp = prop.change_logs[0].timestamp;
prop.change_logs.iter().map(|log| {
let elapsed = log.timestamp.duration_since(first_timestamp);
format!(
"[+{:.3}s] Thread: {}\n {} -> {}\n Stack trace:\n{}\n",
elapsed.as_secs_f64(),
log.thread_id,
log.old_value_repr,
log.new_value_repr,
log.backtrace
)
}).collect()
}
pub fn print_change_logs(&self) {
println!("===== Property Change Logs =====");
for log in self.get_change_logs() {
println!("{}", log);
}
println!("================================");
}
}
impl<T: Clone + Default + Send + Sync + 'static> ObservableProperty<T> {
pub fn get_or_default(&self) -> T {
self.get().unwrap_or_default()
}
}
impl<T: Clone + Send + Sync + 'static> Clone for ObservableProperty<T> {
fn clone(&self) -> Self {
Self {
inner: Arc::clone(&self.inner),
max_threads: self.max_threads,
max_observers: self.max_observers,
}
}
}
impl<T: Clone + std::fmt::Debug + Send + Sync + 'static> std::fmt::Debug for ObservableProperty<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.get() {
Ok(value) => f
.debug_struct("ObservableProperty")
.field("value", &value)
.field("observers_count", &"[hidden]")
.field("max_threads", &self.max_threads)
.field("max_observers", &self.max_observers)
.finish(),
Err(_) => f
.debug_struct("ObservableProperty")
.field("value", &"[inaccessible]")
.field("observers_count", &"[hidden]")
.field("max_threads", &self.max_threads)
.field("max_observers", &self.max_observers)
.finish(),
}
}
}
#[cfg(feature = "serde")]
impl<T: Clone + Send + Sync + 'static + Serialize> Serialize for ObservableProperty<T> {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
match self.get() {
Ok(value) => value.serialize(serializer),
Err(_) => Err(serde::ser::Error::custom("Failed to read property value")),
}
}
}
#[cfg(feature = "serde")]
impl<'de, T: Clone + Send + Sync + 'static + Deserialize<'de>> Deserialize<'de> for ObservableProperty<T> {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = T::deserialize(deserializer)?;
Ok(ObservableProperty::new(value))
}
}
pub fn computed<T, U, F>(
dependencies: Vec<Arc<ObservableProperty<T>>>,
compute_fn: F,
) -> Result<Arc<ObservableProperty<U>>, PropertyError>
where
T: Clone + Send + Sync + 'static,
U: Clone + Send + Sync + 'static,
F: Fn(&[T]) -> U + Send + Sync + 'static,
{
let initial_values: Result<Vec<T>, PropertyError> =
dependencies.iter().map(|dep| dep.get()).collect();
let initial_values = initial_values?;
let initial_computed = compute_fn(&initial_values);
let computed_property = Arc::new(ObservableProperty::new(initial_computed));
let compute_fn = Arc::new(compute_fn);
for dependency in dependencies.iter() {
let deps_clone = dependencies.clone();
let computed_clone = computed_property.clone();
let compute_fn_clone = compute_fn.clone();
dependency.subscribe(Arc::new(move |_old, _new| {
let current_values: Result<Vec<T>, PropertyError> =
deps_clone.iter().map(|dep| dep.get()).collect();
if let Ok(values) = current_values {
let new_computed = compute_fn_clone(&values);
if let Err(e) = computed_clone.set(new_computed) {
eprintln!("Error updating computed property: {}", e);
}
}
}))?;
}
Ok(computed_property)
}
#[cfg(test)]
pub mod testing {
use super::*;
use std::sync::mpsc::{channel, Receiver};
use std::time::Duration;
pub fn await_notification<T: Clone + Send + Sync + 'static>(
property: &ObservableProperty<T>,
) {
let (tx, rx) = channel();
let _subscription = property
.subscribe_with_subscription(Arc::new(move |_old, _new| {
let _ = tx.send(());
}))
.expect("Failed to subscribe for await_notification");
rx.recv_timeout(Duration::from_secs(5))
.expect("Timeout waiting for notification");
}
pub fn collect_changes<T: Clone + Send + Sync + 'static>(
property: &ObservableProperty<T>,
) -> ChangeCollector<T> {
ChangeCollector::new(property)
}
pub struct ChangeCollector<T: Clone + Send + Sync + 'static> {
receiver: Receiver<(T, T)>,
_subscription: Subscription<T>,
}
impl<T: Clone + Send + Sync + 'static> ChangeCollector<T> {
fn new(property: &ObservableProperty<T>) -> Self {
let (tx, rx) = channel();
let subscription = property
.subscribe_with_subscription(Arc::new(move |old, new| {
let _ = tx.send((old.clone(), new.clone()));
}))
.expect("Failed to subscribe for collect_changes");
ChangeCollector {
receiver: rx,
_subscription: subscription,
}
}
pub fn changes(&self) -> Vec<(T, T)> {
let mut changes = Vec::new();
while let Ok(change) = self.receiver.try_recv() {
changes.push(change);
}
changes
}
pub fn wait_for_changes(&self, count: usize) -> Vec<(T, T)> {
let mut changes = Vec::new();
let timeout = Duration::from_secs(5);
let start = std::time::Instant::now();
while changes.len() < count {
if start.elapsed() > timeout {
panic!(
"Timeout waiting for {} changes, only received {}",
count,
changes.len()
);
}
match self.receiver.recv_timeout(Duration::from_millis(100)) {
Ok(change) => changes.push(change),
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => continue,
Err(e) => panic!("Error receiving change: {}", e),
}
}
changes
}
pub fn count(&self) -> usize {
self.changes().len()
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
#[test]
fn test_property_creation_and_basic_operations() {
let prop = ObservableProperty::new(42);
match prop.get() {
Ok(value) => assert_eq!(value, 42),
Err(e) => panic!("Failed to get initial value: {}", e),
}
if let Err(e) = prop.set(100) {
panic!("Failed to set value: {}", e);
}
match prop.get() {
Ok(value) => assert_eq!(value, 100),
Err(e) => panic!("Failed to get updated value: {}", e),
}
}
#[test]
fn test_observer_subscription_and_notification() {
let prop = ObservableProperty::new("initial".to_string());
let notification_count = Arc::new(AtomicUsize::new(0));
let last_old_value = Arc::new(RwLock::new(String::new()));
let last_new_value = Arc::new(RwLock::new(String::new()));
let count_clone = notification_count.clone();
let old_clone = last_old_value.clone();
let new_clone = last_new_value.clone();
let observer_id = match prop.subscribe(Arc::new(move |old, new| {
count_clone.fetch_add(1, Ordering::SeqCst);
if let Ok(mut old_val) = old_clone.write() {
*old_val = old.clone();
}
if let Ok(mut new_val) = new_clone.write() {
*new_val = new.clone();
}
})) {
Ok(id) => id,
Err(e) => panic!("Failed to subscribe observer: {}", e),
};
if let Err(e) = prop.set("changed".to_string()) {
panic!("Failed to set property value: {}", e);
}
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
match last_old_value.read() {
Ok(old_val) => assert_eq!(*old_val, "initial"),
Err(e) => panic!("Failed to read old value: {:?}", e),
}
match last_new_value.read() {
Ok(new_val) => assert_eq!(*new_val, "changed"),
Err(e) => panic!("Failed to read new value: {:?}", e),
}
match prop.unsubscribe(observer_id) {
Ok(was_present) => assert!(was_present),
Err(e) => panic!("Failed to unsubscribe observer: {}", e),
}
if let Err(e) = prop.set("not_notified".to_string()) {
panic!("Failed to set property value after unsubscribe: {}", e);
}
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
}
#[test]
fn test_filtered_observer() {
let prop = ObservableProperty::new(0i32);
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
let observer_id = match prop.subscribe_filtered(
Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
}),
|old, new| new > old,
) {
Ok(id) => id,
Err(e) => panic!("Failed to subscribe filtered observer: {}", e),
};
if let Err(e) = prop.set(5) {
panic!("Failed to set property value to 5: {}", e);
}
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
if let Err(e) = prop.set(3) {
panic!("Failed to set property value to 3: {}", e);
}
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
if let Err(e) = prop.set(10) {
panic!("Failed to set property value to 10: {}", e);
}
assert_eq!(notification_count.load(Ordering::SeqCst), 2);
match prop.unsubscribe(observer_id) {
Ok(_) => {}
Err(e) => panic!("Failed to unsubscribe filtered observer: {}", e),
}
}
#[test]
fn test_thread_safety_concurrent_reads() {
let prop = Arc::new(ObservableProperty::new(42i32));
let num_threads = 10;
let reads_per_thread = 100;
let handles: Vec<_> = (0..num_threads)
.map(|_| {
let prop_clone = prop.clone();
thread::spawn(move || {
for _ in 0..reads_per_thread {
match prop_clone.get() {
Ok(value) => assert_eq!(value, 42),
Err(e) => panic!("Failed to read property value: {}", e),
}
thread::sleep(Duration::from_millis(1));
}
})
})
.collect();
for handle in handles {
if let Err(e) = handle.join() {
panic!("Thread failed to complete: {:?}", e);
}
}
}
#[test]
fn test_async_set_performance() {
let prop = ObservableProperty::new(0i32);
let slow_observer_count = Arc::new(AtomicUsize::new(0));
let count_clone = slow_observer_count.clone();
let _id = match prop.subscribe(Arc::new(move |_, _| {
thread::sleep(Duration::from_millis(50));
count_clone.fetch_add(1, Ordering::SeqCst);
})) {
Ok(id) => id,
Err(e) => panic!("Failed to subscribe slow observer: {}", e),
};
let start = std::time::Instant::now();
if let Err(e) = prop.set(1) {
panic!("Failed to set property value synchronously: {}", e);
}
let sync_duration = start.elapsed();
let start = std::time::Instant::now();
if let Err(e) = prop.set_async(2) {
panic!("Failed to set property value asynchronously: {}", e);
}
let async_duration = start.elapsed();
assert!(async_duration < sync_duration);
assert!(async_duration.as_millis() < 10);
thread::sleep(Duration::from_millis(100));
assert_eq!(slow_observer_count.load(Ordering::SeqCst), 2);
}
#[test]
fn test_lock_poisoning() {
let prop = Arc::new(ObservableProperty::new(0));
let prop_clone = prop.clone();
let poison_thread = thread::spawn(move || {
let _guard = prop_clone.inner.write().expect("Failed to acquire write lock for poisoning test");
panic!("Deliberate panic to poison the lock");
});
let _ = poison_thread.join();
match prop.get() {
Ok(value) => assert_eq!(value, 0), Err(e) => panic!("get() should succeed with graceful degradation, got error: {:?}", e),
}
match prop.set(42) {
Ok(_) => {}, Err(e) => panic!("set() should succeed with graceful degradation, got error: {:?}", e),
}
assert_eq!(prop.get().expect("Failed to get value after set"), 42);
match prop.subscribe(Arc::new(|_, _| {})) {
Ok(_) => {}, Err(e) => panic!("subscribe() should succeed with graceful degradation, got error: {:?}", e),
}
}
#[test]
fn test_observer_panic_isolation() {
let prop = ObservableProperty::new(0);
let call_counts = Arc::new(AtomicUsize::new(0));
let panic_observer_id = prop
.subscribe(Arc::new(|_, _| {
panic!("This observer deliberately panics");
}))
.expect("Failed to subscribe panic observer");
let counts = call_counts.clone();
let normal_observer_id = prop
.subscribe(Arc::new(move |_, _| {
counts.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to subscribe normal observer");
prop.set(42).expect("Failed to set property value");
assert_eq!(call_counts.load(Ordering::SeqCst), 1);
prop.unsubscribe(panic_observer_id).expect("Failed to unsubscribe panic observer");
prop.unsubscribe(normal_observer_id).expect("Failed to unsubscribe normal observer");
}
#[test]
fn test_unsubscribe_nonexistent_observer() {
let property = ObservableProperty::new(0);
let valid_id = property.subscribe(Arc::new(|_, _| {})).expect("Failed to subscribe test observer");
let nonexistent_id = valid_id + 1000;
match property.unsubscribe(nonexistent_id) {
Ok(was_present) => {
assert!(
!was_present,
"Unsubscribe should return false for nonexistent ID"
);
}
Err(e) => panic!("Unsubscribe returned error: {:?}", e),
}
property.unsubscribe(valid_id).expect("Failed first unsubscribe");
let result = property.unsubscribe(valid_id).expect("Failed second unsubscribe");
assert!(!result, "Second unsubscribe should return false");
}
#[test]
fn test_observer_id_wraparound() {
let prop = ObservableProperty::new(0);
let id1 = prop.subscribe(Arc::new(|_, _| {})).expect("Failed to subscribe observer 1");
let id2 = prop.subscribe(Arc::new(|_, _| {})).expect("Failed to subscribe observer 2");
let id3 = prop.subscribe(Arc::new(|_, _| {})).expect("Failed to subscribe observer 3");
assert!(id2 > id1, "Observer IDs should increment");
assert!(id3 > id2, "Observer IDs should continue incrementing");
assert_eq!(id2, id1 + 1, "Observer IDs should increment by 1");
assert_eq!(id3, id2 + 1, "Observer IDs should increment by 1");
prop.unsubscribe(id1).expect("Failed to unsubscribe observer 1");
prop.unsubscribe(id2).expect("Failed to unsubscribe observer 2");
prop.unsubscribe(id3).expect("Failed to unsubscribe observer 3");
}
#[test]
fn test_concurrent_subscribe_unsubscribe() {
let prop = Arc::new(ObservableProperty::new(0));
let num_threads = 8;
let operations_per_thread = 100;
let handles: Vec<_> = (0..num_threads)
.map(|thread_id| {
let prop_clone = prop.clone();
thread::spawn(move || {
let mut local_ids = Vec::new();
for i in 0..operations_per_thread {
let observer_id = prop_clone
.subscribe(Arc::new(move |old, new| {
let _ = thread_id + i + old + new;
}))
.expect("Subscribe should succeed");
local_ids.push(observer_id);
if i % 10 == 0 && !local_ids.is_empty() {
let idx = i % local_ids.len();
let id_to_remove = local_ids.remove(idx);
prop_clone
.unsubscribe(id_to_remove)
.expect("Unsubscribe should succeed");
}
}
for id in local_ids {
prop_clone
.unsubscribe(id)
.expect("Final cleanup should succeed");
}
})
})
.collect();
for handle in handles {
handle.join().expect("Thread should complete successfully");
}
prop.set(42)
.expect("Property should still work after concurrent operations");
}
#[test]
fn test_multiple_observer_panics_isolation() {
let prop = ObservableProperty::new(0);
let successful_calls = Arc::new(AtomicUsize::new(0));
let _panic_id1 = prop
.subscribe(Arc::new(|_, _| {
panic!("First panic observer");
}))
.expect("Failed to subscribe first panic observer");
let _panic_id2 = prop
.subscribe(Arc::new(|_, _| {
panic!("Second panic observer");
}))
.expect("Failed to subscribe second panic observer");
let count1 = successful_calls.clone();
let _success_id1 = prop
.subscribe(Arc::new(move |_, _| {
count1.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to subscribe first success observer");
let count2 = successful_calls.clone();
let _success_id2 = prop
.subscribe(Arc::new(move |_, _| {
count2.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to subscribe second success observer");
prop.set(42).expect("Failed to set property value for panic isolation test");
assert_eq!(successful_calls.load(Ordering::SeqCst), 2);
}
#[test]
fn test_async_observer_panic_isolation() {
let prop = ObservableProperty::new(0);
let successful_calls = Arc::new(AtomicUsize::new(0));
let _panic_id = prop
.subscribe(Arc::new(|_, _| {
panic!("Async panic observer");
}))
.expect("Failed to subscribe async panic observer");
let count = successful_calls.clone();
let _success_id = prop
.subscribe(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to subscribe async success observer");
prop.set_async(42).expect("Failed to set property value asynchronously");
thread::sleep(Duration::from_millis(100));
assert_eq!(successful_calls.load(Ordering::SeqCst), 1);
}
#[test]
fn test_very_large_observer_count() {
let prop = ObservableProperty::new(0);
let total_calls = Arc::new(AtomicUsize::new(0));
let observer_count = 1000;
let mut observer_ids = Vec::with_capacity(observer_count);
for i in 0..observer_count {
let count = total_calls.clone();
let id = prop
.subscribe(Arc::new(move |old, new| {
count.fetch_add(1, Ordering::SeqCst);
assert_eq!(*old, 0);
assert_eq!(*new, i + 1);
}))
.expect("Failed to subscribe large observer count test observer");
observer_ids.push(id);
}
prop.set(observer_count).expect("Failed to set property value for large observer count test");
assert_eq!(total_calls.load(Ordering::SeqCst), observer_count);
for id in observer_ids {
prop.unsubscribe(id).expect("Failed to unsubscribe observer in large count test");
}
}
#[test]
fn test_observer_with_mutable_state() {
let prop = ObservableProperty::new(0);
let call_history = Arc::new(RwLock::new(Vec::new()));
let history = call_history.clone();
let observer_id = prop
.subscribe(Arc::new(move |old, new| {
if let Ok(mut hist) = history.write() {
hist.push((*old, *new));
}
}))
.expect("Failed to subscribe mutable state observer");
prop.set(1).expect("Failed to set property to 1");
prop.set(2).expect("Failed to set property to 2");
prop.set(3).expect("Failed to set property to 3");
let history = call_history.read().expect("Failed to read call history");
assert_eq!(history.len(), 3);
assert_eq!(history[0], (0, 1));
assert_eq!(history[1], (1, 2));
assert_eq!(history[2], (2, 3));
prop.unsubscribe(observer_id).expect("Failed to unsubscribe mutable state observer");
}
#[test]
fn test_subscription_automatic_cleanup() {
let prop = ObservableProperty::new(0);
let call_count = Arc::new(AtomicUsize::new(0));
{
let count = call_count.clone();
let _subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription for automatic cleanup test");
prop.set(1).expect("Failed to set property value in subscription test");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
prop.set(2).expect("Failed to set property value after subscription dropped");
assert_eq!(call_count.load(Ordering::SeqCst), 1); }
#[test]
fn test_subscription_explicit_drop() {
let prop = ObservableProperty::new(0);
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription for explicit drop test");
prop.set(1).expect("Failed to set property value before explicit drop");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
drop(subscription);
prop.set(2).expect("Failed to set property value after explicit drop");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[test]
fn test_multiple_subscriptions_with_cleanup() {
let prop = ObservableProperty::new(0);
let call_count1 = Arc::new(AtomicUsize::new(0));
let call_count2 = Arc::new(AtomicUsize::new(0));
let call_count3 = Arc::new(AtomicUsize::new(0));
let count1 = call_count1.clone();
let count2 = call_count2.clone();
let count3 = call_count3.clone();
let subscription1 = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count1.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create first subscription");
let subscription2 = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count2.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create second subscription");
let subscription3 = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count3.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create third subscription");
prop.set(1).expect("Failed to set property value with all subscriptions");
assert_eq!(call_count1.load(Ordering::SeqCst), 1);
assert_eq!(call_count2.load(Ordering::SeqCst), 1);
assert_eq!(call_count3.load(Ordering::SeqCst), 1);
drop(subscription2);
prop.set(2).expect("Failed to set property value with partial subscriptions");
assert_eq!(call_count1.load(Ordering::SeqCst), 2);
assert_eq!(call_count2.load(Ordering::SeqCst), 1); assert_eq!(call_count3.load(Ordering::SeqCst), 2);
drop(subscription1);
drop(subscription3);
prop.set(3).expect("Failed to set property value with no subscriptions");
assert_eq!(call_count1.load(Ordering::SeqCst), 2);
assert_eq!(call_count2.load(Ordering::SeqCst), 1);
assert_eq!(call_count3.load(Ordering::SeqCst), 2);
}
#[test]
fn test_subscription_drop_with_poisoned_lock() {
let prop = Arc::new(ObservableProperty::new(0));
let prop_clone = prop.clone();
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription for poisoned lock test");
let poison_thread = thread::spawn(move || {
let _guard = prop_clone.inner.write().expect("Failed to acquire write lock for poisoning test");
panic!("Deliberate panic to poison the lock");
});
let _ = poison_thread.join();
drop(subscription);
}
#[test]
fn test_subscription_vs_manual_unsubscribe() {
let prop = ObservableProperty::new(0);
let auto_count = Arc::new(AtomicUsize::new(0));
let manual_count = Arc::new(AtomicUsize::new(0));
let manual_count_clone = manual_count.clone();
let manual_id = prop
.subscribe(Arc::new(move |_, _| {
manual_count_clone.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create manual subscription");
let auto_count_clone = auto_count.clone();
let _auto_subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
auto_count_clone.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create automatic subscription");
prop.set(1).expect("Failed to set property value with both subscriptions");
assert_eq!(manual_count.load(Ordering::SeqCst), 1);
assert_eq!(auto_count.load(Ordering::SeqCst), 1);
prop.unsubscribe(manual_id).expect("Failed to manually unsubscribe");
prop.set(2).expect("Failed to set property value after manual unsubscribe");
assert_eq!(manual_count.load(Ordering::SeqCst), 1); assert_eq!(auto_count.load(Ordering::SeqCst), 2);
}
#[test]
fn test_subscribe_with_subscription_error_handling() {
let prop = Arc::new(ObservableProperty::new(0));
let prop_clone = prop.clone();
let poison_thread = thread::spawn(move || {
let _guard = prop_clone.inner.write().expect("Failed to acquire write lock for poisoning test");
panic!("Deliberate panic to poison the lock");
});
let _ = poison_thread.join();
let result = prop.subscribe_with_subscription(Arc::new(|_, _| {}));
assert!(result.is_ok(), "subscribe_with_subscription should succeed with graceful degradation");
}
#[test]
fn test_subscription_with_property_cloning() {
let prop1 = ObservableProperty::new(0);
let prop2 = prop1.clone();
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let _subscription = prop1
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription for cloned property test");
prop2.set(42).expect("Failed to set property value through prop2");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
prop1.set(100).expect("Failed to set property value through prop1");
assert_eq!(call_count.load(Ordering::SeqCst), 2);
}
#[test]
fn test_subscription_in_conditional_blocks() {
let prop = ObservableProperty::new(0);
let call_count = Arc::new(AtomicUsize::new(0));
let should_subscribe = true;
if should_subscribe {
let count = call_count.clone();
let _subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription in conditional block");
prop.set(1).expect("Failed to set property value in conditional block");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
prop.set(2).expect("Failed to set property value after conditional block");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[test]
fn test_subscription_with_early_return() {
fn test_function(
prop: &ObservableProperty<i32>,
should_return_early: bool,
) -> Result<(), PropertyError> {
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let _subscription = prop.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))?;
prop.set(1)?;
assert_eq!(call_count.load(Ordering::SeqCst), 1);
if should_return_early {
return Ok(()); }
prop.set(2)?;
assert_eq!(call_count.load(Ordering::SeqCst), 2);
Ok(())
}
let prop = ObservableProperty::new(0);
test_function(&prop, true).expect("Failed to test early return");
prop.set(10).expect("Failed to set property value after early return");
test_function(&prop, false).expect("Failed to test normal exit");
prop.set(20).expect("Failed to set property value after normal exit");
}
#[test]
fn test_subscription_move_semantics() {
let prop = ObservableProperty::new(0);
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription for move semantics test");
prop.set(1).expect("Failed to set property value before move");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
let moved_subscription = subscription;
prop.set(2).expect("Failed to set property value after move");
assert_eq!(call_count.load(Ordering::SeqCst), 2);
drop(moved_subscription);
prop.set(3).expect("Failed to set property value after moved subscription drop");
assert_eq!(call_count.load(Ordering::SeqCst), 2);
}
#[test]
fn test_filtered_subscription_automatic_cleanup() {
let prop = ObservableProperty::new(0);
let call_count = Arc::new(AtomicUsize::new(0));
{
let count = call_count.clone();
let _subscription = prop
.subscribe_filtered_with_subscription(
Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}),
|old, new| new > old, )
.expect("Failed to create filtered subscription");
prop.set(5).expect("Failed to set property value to 5 in filtered test");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
prop.set(3).expect("Failed to set property value to 3 in filtered test");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
prop.set(10).expect("Failed to set property value to 10 in filtered test");
assert_eq!(call_count.load(Ordering::SeqCst), 2);
}
prop.set(20).expect("Failed to set property value after filtered subscription cleanup");
assert_eq!(call_count.load(Ordering::SeqCst), 2);
}
#[test]
fn test_multiple_filtered_subscriptions() {
let prop = ObservableProperty::new(10);
let increase_count = Arc::new(AtomicUsize::new(0));
let decrease_count = Arc::new(AtomicUsize::new(0));
let significant_change_count = Arc::new(AtomicUsize::new(0));
let inc_count = increase_count.clone();
let dec_count = decrease_count.clone();
let sig_count = significant_change_count.clone();
let _increase_sub = prop
.subscribe_filtered_with_subscription(
Arc::new(move |_, _| {
inc_count.fetch_add(1, Ordering::SeqCst);
}),
|old, new| new > old,
)
.expect("Failed to create increase subscription");
let _decrease_sub = prop
.subscribe_filtered_with_subscription(
Arc::new(move |_, _| {
dec_count.fetch_add(1, Ordering::SeqCst);
}),
|old, new| new < old,
)
.expect("Failed to create decrease subscription");
let _significant_sub = prop
.subscribe_filtered_with_subscription(
Arc::new(move |_, _| {
sig_count.fetch_add(1, Ordering::SeqCst);
}),
|old, new| ((*new as i32) - (*old as i32)).abs() > 5,
)
.expect("Failed to create significant change subscription");
prop.set(15).expect("Failed to set property to 15 in multiple filtered test"); assert_eq!(increase_count.load(Ordering::SeqCst), 1);
assert_eq!(decrease_count.load(Ordering::SeqCst), 0);
assert_eq!(significant_change_count.load(Ordering::SeqCst), 0);
prop.set(25).expect("Failed to set property to 25 in multiple filtered test"); assert_eq!(increase_count.load(Ordering::SeqCst), 2);
assert_eq!(decrease_count.load(Ordering::SeqCst), 0);
assert_eq!(significant_change_count.load(Ordering::SeqCst), 1);
prop.set(5).expect("Failed to set property to 5 in multiple filtered test"); assert_eq!(increase_count.load(Ordering::SeqCst), 2);
assert_eq!(decrease_count.load(Ordering::SeqCst), 1);
assert_eq!(significant_change_count.load(Ordering::SeqCst), 2);
prop.set(3).expect("Failed to set property to 3 in multiple filtered test"); assert_eq!(increase_count.load(Ordering::SeqCst), 2);
assert_eq!(decrease_count.load(Ordering::SeqCst), 2);
assert_eq!(significant_change_count.load(Ordering::SeqCst), 2);
}
#[test]
fn test_filtered_subscription_complex_filter() {
let prop = ObservableProperty::new(0.0f64);
let call_count = Arc::new(AtomicUsize::new(0));
let values_received = Arc::new(RwLock::new(Vec::new()));
let count = call_count.clone();
let values = values_received.clone();
let _subscription = prop
.subscribe_filtered_with_subscription(
Arc::new(move |old, new| {
count.fetch_add(1, Ordering::SeqCst);
if let Ok(mut v) = values.write() {
v.push((*old, *new));
}
}),
|old, new| {
let old_int = old.floor() as i32;
let new_int = new.floor() as i32;
old_int != new_int && (new - old).abs() > 0.5_f64
},
)
.expect("Failed to create complex filtered subscription");
prop.set(0.3).expect("Failed to set property to 0.3 in complex filter test");
prop.set(0.7).expect("Failed to set property to 0.7 in complex filter test");
assert_eq!(call_count.load(Ordering::SeqCst), 0);
prop.set(1.3).expect("Failed to set property to 1.3 in complex filter test"); assert_eq!(call_count.load(Ordering::SeqCst), 1);
prop.set(1.9).expect("Failed to set property to 1.9 in complex filter test");
prop.set(2.1).expect("Failed to set property to 2.1 in complex filter test"); assert_eq!(call_count.load(Ordering::SeqCst), 1);
prop.set(3.5).expect("Failed to set property to 3.5 in complex filter test");
assert_eq!(call_count.load(Ordering::SeqCst), 2);
let values = values_received.read().expect("Failed to read values in complex filter test");
assert_eq!(values.len(), 2);
assert_eq!(values[0], (0.7, 1.3));
assert_eq!(values[1], (2.1, 3.5));
}
#[test]
fn test_filtered_subscription_error_handling() {
let prop = Arc::new(ObservableProperty::new(0));
let prop_clone = prop.clone();
let poison_thread = thread::spawn(move || {
let _guard = prop_clone.inner.write().expect("Failed to acquire write lock for filtered subscription poison test");
panic!("Deliberate panic to poison the lock");
});
let _ = poison_thread.join();
let result = prop.subscribe_filtered_with_subscription(Arc::new(|_, _| {}), |_, _| true);
assert!(result.is_ok(), "subscribe_filtered_with_subscription should succeed with graceful degradation");
}
#[test]
fn test_filtered_subscription_vs_manual_filtered() {
let prop = ObservableProperty::new(0);
let auto_count = Arc::new(AtomicUsize::new(0));
let manual_count = Arc::new(AtomicUsize::new(0));
let manual_count_clone = manual_count.clone();
let manual_id = prop
.subscribe_filtered(
Arc::new(move |_, _| {
manual_count_clone.fetch_add(1, Ordering::SeqCst);
}),
|old, new| new > old,
)
.expect("Failed to create manual filtered subscription");
let auto_count_clone = auto_count.clone();
let _auto_subscription = prop
.subscribe_filtered_with_subscription(
Arc::new(move |_, _| {
auto_count_clone.fetch_add(1, Ordering::SeqCst);
}),
|old, new| new > old,
)
.expect("Failed to create automatic filtered subscription");
prop.set(5).expect("Failed to set property to 5 in filtered vs manual test");
assert_eq!(manual_count.load(Ordering::SeqCst), 1);
assert_eq!(auto_count.load(Ordering::SeqCst), 1);
prop.set(3).expect("Failed to set property to 3 in filtered vs manual test");
assert_eq!(manual_count.load(Ordering::SeqCst), 1);
assert_eq!(auto_count.load(Ordering::SeqCst), 1);
prop.set(10).expect("Failed to set property to 10 in filtered vs manual test");
assert_eq!(manual_count.load(Ordering::SeqCst), 2);
assert_eq!(auto_count.load(Ordering::SeqCst), 2);
prop.unsubscribe(manual_id).expect("Failed to unsubscribe manual filtered observer");
prop.set(15).expect("Failed to set property to 15 after manual cleanup");
assert_eq!(manual_count.load(Ordering::SeqCst), 2); assert_eq!(auto_count.load(Ordering::SeqCst), 3);
}
#[test]
fn test_filtered_subscription_with_panicking_filter() {
let prop = ObservableProperty::new(0);
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let _subscription = prop
.subscribe_filtered_with_subscription(
Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}),
|_, new| {
if *new == 42 {
panic!("Filter panic on 42");
}
true },
)
.expect("Failed to create panicking filter subscription");
prop.set(1).expect("Failed to set property to 1 in panicking filter test");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
prop.set(42).expect("Failed to set property to 42 in panicking filter test");
prop.set(2).expect("Failed to set property to 2 after filter panic");
}
#[test]
fn test_subscription_thread_safety() {
let prop = Arc::new(ObservableProperty::new(0));
let num_threads = 8;
let operations_per_thread = 50;
let total_calls = Arc::new(AtomicUsize::new(0));
let handles: Vec<_> = (0..num_threads)
.map(|thread_id| {
let prop_clone = prop.clone();
let calls_clone = total_calls.clone();
thread::spawn(move || {
let mut local_subscriptions = Vec::new();
for i in 0..operations_per_thread {
let calls = calls_clone.clone();
let subscription = prop_clone
.subscribe_with_subscription(Arc::new(move |old, new| {
calls.fetch_add(1, Ordering::SeqCst);
let _ = thread_id + i + old + new;
}))
.expect("Should be able to create subscription");
local_subscriptions.push(subscription);
prop_clone
.set(thread_id * 1000 + i)
.expect("Should be able to set value");
if i % 5 == 0 && !local_subscriptions.is_empty() {
local_subscriptions.remove(0); }
}
})
})
.collect();
for handle in handles {
handle.join().expect("Thread should complete successfully");
}
prop.set(9999).expect("Property should still work");
println!(
"Total observer calls: {}",
total_calls.load(Ordering::SeqCst)
);
}
#[test]
fn test_subscription_cross_thread_drop() {
let prop = Arc::new(ObservableProperty::new(0));
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription for cross-thread drop test");
prop.set(1).expect("Failed to set property value in cross-thread drop test");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
let prop_clone = prop.clone();
let call_count_clone = call_count.clone();
let handle = thread::spawn(move || {
prop_clone.set(2).expect("Failed to set property value in other thread");
assert_eq!(call_count_clone.load(Ordering::SeqCst), 2);
drop(subscription);
prop_clone.set(3).expect("Failed to set property value after drop in other thread");
assert_eq!(call_count_clone.load(Ordering::SeqCst), 2); });
handle.join().expect("Failed to join cross-thread drop test thread");
prop.set(4).expect("Failed to set property value after thread join");
assert_eq!(call_count.load(Ordering::SeqCst), 2);
}
#[test]
fn test_concurrent_subscription_creation_and_property_changes() {
let prop = Arc::new(ObservableProperty::new(0));
let total_notifications = Arc::new(AtomicUsize::new(0));
let num_subscriber_threads = 4;
let num_setter_threads = 2;
let operations_per_thread = 25;
let subscriber_handles: Vec<_> = (0..num_subscriber_threads)
.map(|_| {
let prop_clone = prop.clone();
let notifications_clone = total_notifications.clone();
thread::spawn(move || {
for _ in 0..operations_per_thread {
let notifications = notifications_clone.clone();
let _subscription = prop_clone
.subscribe_with_subscription(Arc::new(move |_, _| {
notifications.fetch_add(1, Ordering::SeqCst);
}))
.expect("Should create subscription");
thread::sleep(Duration::from_millis(1));
}
})
})
.collect();
let setter_handles: Vec<_> = (0..num_setter_threads)
.map(|thread_id| {
let prop_clone = prop.clone();
thread::spawn(move || {
for i in 0..operations_per_thread * 2 {
prop_clone
.set(thread_id * 10000 + i)
.expect("Should set value");
thread::sleep(Duration::from_millis(1));
}
})
})
.collect();
for handle in subscriber_handles
.into_iter()
.chain(setter_handles.into_iter())
{
handle.join().expect("Thread should complete");
}
prop.set(99999).expect("Property should still work");
println!(
"Total notifications during concurrent test: {}",
total_notifications.load(Ordering::SeqCst)
);
}
#[test]
fn test_filtered_subscription_thread_safety() {
let prop = Arc::new(ObservableProperty::new(0));
let increase_notifications = Arc::new(AtomicUsize::new(0));
let decrease_notifications = Arc::new(AtomicUsize::new(0));
let num_threads = 6;
let handles: Vec<_> = (0..num_threads)
.map(|thread_id| {
let prop_clone = prop.clone();
let inc_notifications = increase_notifications.clone();
let dec_notifications = decrease_notifications.clone();
thread::spawn(move || {
let inc_count = inc_notifications.clone();
let _inc_subscription = prop_clone
.subscribe_filtered_with_subscription(
Arc::new(move |_, _| {
inc_count.fetch_add(1, Ordering::SeqCst);
}),
|old, new| new > old,
)
.expect("Should create filtered subscription");
let dec_count = dec_notifications.clone();
let _dec_subscription = prop_clone
.subscribe_filtered_with_subscription(
Arc::new(move |_, _| {
dec_count.fetch_add(1, Ordering::SeqCst);
}),
|old, new| new < old,
)
.expect("Should create filtered subscription");
let base_value = thread_id * 100;
for i in 0..20 {
let new_value = base_value + (i % 10); prop_clone.set(new_value).expect("Should set value");
thread::sleep(Duration::from_millis(1));
}
})
})
.collect();
for handle in handles {
handle.join().expect("Thread should complete");
}
let initial_inc = increase_notifications.load(Ordering::SeqCst);
let initial_dec = decrease_notifications.load(Ordering::SeqCst);
prop.set(1000).expect("Property should still work");
prop.set(2000).expect("Property should still work");
assert_eq!(increase_notifications.load(Ordering::SeqCst), initial_inc);
assert_eq!(decrease_notifications.load(Ordering::SeqCst), initial_dec);
println!(
"Increase notifications: {}, Decrease notifications: {}",
initial_inc, initial_dec
);
}
#[test]
fn test_subscription_with_async_property_changes() {
let prop = Arc::new(ObservableProperty::new(0));
let sync_notifications = Arc::new(AtomicUsize::new(0));
let async_notifications = Arc::new(AtomicUsize::new(0));
let sync_count = sync_notifications.clone();
let _sync_subscription = prop
.subscribe_with_subscription(Arc::new(move |old, new| {
sync_count.fetch_add(1, Ordering::SeqCst);
thread::sleep(Duration::from_millis(5));
println!("Sync observer: {} -> {}", old, new);
}))
.expect("Failed to create sync subscription");
let async_count = async_notifications.clone();
let _async_subscription = prop
.subscribe_with_subscription(Arc::new(move |old, new| {
async_count.fetch_add(1, Ordering::SeqCst);
println!("Async observer: {} -> {}", old, new);
}))
.expect("Failed to create async subscription");
let start = std::time::Instant::now();
prop.set(1).expect("Failed to set property value 1 in async test");
prop.set(2).expect("Failed to set property value 2 in async test");
let sync_duration = start.elapsed();
let start = std::time::Instant::now();
prop.set_async(3).expect("Failed to set property value 3 async");
prop.set_async(4).expect("Failed to set property value 4 async");
let async_duration = start.elapsed();
assert!(async_duration < sync_duration);
thread::sleep(Duration::from_millis(50));
assert_eq!(sync_notifications.load(Ordering::SeqCst), 4);
assert_eq!(async_notifications.load(Ordering::SeqCst), 4);
}
#[test]
fn test_subscription_creation_with_poisoned_lock() {
let prop = Arc::new(ObservableProperty::new(0));
let prop_clone = prop.clone();
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let existing_subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription before poisoning");
let poison_thread = thread::spawn(move || {
let _guard = prop_clone.inner.write().expect("Failed to acquire write lock for subscription poison test");
panic!("Deliberate panic to poison the lock");
});
let _ = poison_thread.join();
let result = prop.subscribe_with_subscription(Arc::new(|_, _| {}));
assert!(result.is_ok(), "subscribe_with_subscription should succeed with graceful degradation");
let filtered_result =
prop.subscribe_filtered_with_subscription(Arc::new(|_, _| {}), |_, _| true);
assert!(filtered_result.is_ok(), "subscribe_filtered_with_subscription should succeed with graceful degradation");
drop(existing_subscription);
}
#[test]
fn test_subscription_cleanup_behavior_with_poisoned_lock() {
let prop = Arc::new(ObservableProperty::new(0));
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription for cleanup behavior test");
prop.set(1).expect("Failed to set property value in cleanup behavior test");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
let prop_clone = prop.clone();
let poison_thread = thread::spawn(move || {
let _guard = prop_clone.inner.write().expect("Failed to acquire write lock for cleanup behavior poison test");
panic!("Deliberate panic to poison the lock");
});
let _ = poison_thread.join();
drop(subscription);
}
#[test]
fn test_multiple_subscription_cleanup_with_poisoned_lock() {
let prop = Arc::new(ObservableProperty::new(0));
let mut subscriptions = Vec::new();
for i in 0..5 {
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let subscription = prop
.subscribe_with_subscription(Arc::new(move |old, new| {
count.fetch_add(1, Ordering::SeqCst);
println!("Observer {}: {} -> {}", i, old, new);
}))
.expect("Failed to create subscription in multiple cleanup test");
subscriptions.push(subscription);
}
prop.set(42).expect("Failed to set property value in multiple cleanup test");
let prop_clone = prop.clone();
let poison_thread = thread::spawn(move || {
let _guard = prop_clone.inner.write().expect("Failed to acquire write lock for multiple cleanup poison test");
panic!("Deliberate panic to poison the lock");
});
let _ = poison_thread.join();
for subscription in subscriptions {
drop(subscription);
}
}
#[test]
fn test_subscription_behavior_before_and_after_poison() {
let prop = Arc::new(ObservableProperty::new(0));
let before_poison_count = Arc::new(AtomicUsize::new(0));
let after_poison_count = Arc::new(AtomicUsize::new(0));
let before_count = before_poison_count.clone();
let before_subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
before_count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription before poison test");
prop.set(1).expect("Failed to set property value before poison test");
assert_eq!(before_poison_count.load(Ordering::SeqCst), 1);
let prop_clone = prop.clone();
let poison_thread = thread::spawn(move || {
let _guard = prop_clone.inner.write().expect("Failed to acquire write lock for before/after poison test");
panic!("Deliberate panic to poison the lock");
});
let _ = poison_thread.join();
let after_count = after_poison_count.clone();
let after_result = prop.subscribe_with_subscription(Arc::new(move |_, _| {
after_count.fetch_add(1, Ordering::SeqCst);
}));
assert!(after_result.is_ok(), "subscribe_with_subscription should succeed with graceful degradation");
let _after_subscription = after_result.unwrap();
drop(before_subscription);
}
#[test]
fn test_concurrent_subscription_drops_with_poison() {
let prop = Arc::new(ObservableProperty::new(0));
let num_subscriptions = 10;
let mut subscriptions = Vec::new();
for i in 0..num_subscriptions {
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
println!("Observer {}", i);
}))
.expect("Failed to create subscription in concurrent drops test");
subscriptions.push(subscription);
}
let prop_clone = prop.clone();
let poison_thread = thread::spawn(move || {
let _guard = prop_clone.inner.write().expect("Failed to acquire write lock for concurrent drops poison test");
panic!("Deliberate panic to poison the lock");
});
let _ = poison_thread.join();
let handles: Vec<_> = subscriptions
.into_iter()
.enumerate()
.map(|(i, subscription)| {
thread::spawn(move || {
thread::sleep(Duration::from_millis(i as u64 % 5));
drop(subscription);
println!("Dropped subscription {}", i);
})
})
.collect();
for handle in handles {
handle
.join()
.expect("Drop thread should complete without panic");
}
}
#[test]
fn test_with_max_threads_creation() {
let prop1 = ObservableProperty::with_max_threads(42, 1);
let prop2 = ObservableProperty::with_max_threads("test".to_string(), 8);
let prop3 = ObservableProperty::with_max_threads(0.5_f64, 16);
assert_eq!(prop1.get().expect("Failed to get prop1 value"), 42);
assert_eq!(prop2.get().expect("Failed to get prop2 value"), "test");
assert_eq!(prop3.get().expect("Failed to get prop3 value"), 0.5);
}
#[test]
fn test_with_max_threads_zero_defaults_to_max_threads() {
let prop1 = ObservableProperty::with_max_threads(100, 0);
let prop2 = ObservableProperty::new(100);
assert_eq!(prop1.get().expect("Failed to get prop1 value"), 100);
assert_eq!(prop2.get().expect("Failed to get prop2 value"), 100);
}
#[test]
fn test_with_max_threads_basic_functionality() {
let prop = ObservableProperty::with_max_threads(0, 2);
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let _subscription = prop
.subscribe_with_subscription(Arc::new(move |old, new| {
count.fetch_add(1, Ordering::SeqCst);
assert_eq!(*old, 0);
assert_eq!(*new, 42);
}))
.expect("Failed to create subscription for max_threads test");
prop.set(42).expect("Failed to set value synchronously");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
prop.set_async(43).expect("Failed to set value asynchronously");
thread::sleep(Duration::from_millis(50));
assert_eq!(call_count.load(Ordering::SeqCst), 2);
}
#[test]
fn test_with_max_threads_async_performance() {
let prop = ObservableProperty::with_max_threads(0, 1); let slow_call_count = Arc::new(AtomicUsize::new(0));
let mut subscriptions = Vec::new();
for _ in 0..4 {
let count = slow_call_count.clone();
let subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
thread::sleep(Duration::from_millis(25)); count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create slow observer subscription");
subscriptions.push(subscription);
}
let start = std::time::Instant::now();
prop.set_async(42).expect("Failed to set value asynchronously");
let async_duration = start.elapsed();
assert!(async_duration.as_millis() < 50, "Async set should return quickly");
thread::sleep(Duration::from_millis(200));
assert_eq!(slow_call_count.load(Ordering::SeqCst), 4);
}
#[test]
fn test_with_max_threads_vs_regular_constructor() {
let prop_regular = ObservableProperty::new(42);
let prop_custom = ObservableProperty::with_max_threads(42, 4);
let count_regular = Arc::new(AtomicUsize::new(0));
let count_custom = Arc::new(AtomicUsize::new(0));
let count1 = count_regular.clone();
let _sub1 = prop_regular
.subscribe_with_subscription(Arc::new(move |_, _| {
count1.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create regular subscription");
let count2 = count_custom.clone();
let _sub2 = prop_custom
.subscribe_with_subscription(Arc::new(move |_, _| {
count2.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create custom subscription");
prop_regular.set(100).expect("Failed to set regular property");
prop_custom.set(100).expect("Failed to set custom property");
assert_eq!(count_regular.load(Ordering::SeqCst), 1);
assert_eq!(count_custom.load(Ordering::SeqCst), 1);
prop_regular.set_async(200).expect("Failed to set regular property async");
prop_custom.set_async(200).expect("Failed to set custom property async");
thread::sleep(Duration::from_millis(50));
assert_eq!(count_regular.load(Ordering::SeqCst), 2);
assert_eq!(count_custom.load(Ordering::SeqCst), 2);
}
#[test]
fn test_with_max_threads_large_values() {
let prop = ObservableProperty::with_max_threads(0, 1000);
let call_count = Arc::new(AtomicUsize::new(0));
let count = call_count.clone();
let _subscription = prop
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription for large max_threads test");
prop.set_async(42).expect("Failed to set value with large max_threads");
thread::sleep(Duration::from_millis(50));
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[test]
fn test_with_max_threads_clone_behavior() {
let prop1 = ObservableProperty::with_max_threads(42, 2);
let prop2 = prop1.clone();
let call_count1 = Arc::new(AtomicUsize::new(0));
let call_count2 = Arc::new(AtomicUsize::new(0));
let count1 = call_count1.clone();
let _sub1 = prop1
.subscribe_with_subscription(Arc::new(move |_, _| {
count1.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription for cloned property test");
let count2 = call_count2.clone();
let _sub2 = prop2
.subscribe_with_subscription(Arc::new(move |_, _| {
count2.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create subscription for original property test");
prop1.set_async(100).expect("Failed to set value through prop1");
thread::sleep(Duration::from_millis(50));
assert_eq!(call_count1.load(Ordering::SeqCst), 1);
assert_eq!(call_count2.load(Ordering::SeqCst), 1);
prop2.set_async(200).expect("Failed to set value through prop2");
thread::sleep(Duration::from_millis(50));
assert_eq!(call_count1.load(Ordering::SeqCst), 2);
assert_eq!(call_count2.load(Ordering::SeqCst), 2);
}
#[test]
fn test_with_max_threads_thread_safety() {
let prop = Arc::new(ObservableProperty::with_max_threads(0, 3));
let call_count = Arc::new(AtomicUsize::new(0));
let handles: Vec<_> = (0..5)
.map(|thread_id| {
let prop_clone = prop.clone();
let count_clone = call_count.clone();
thread::spawn(move || {
let count = count_clone.clone();
let _subscription = prop_clone
.subscribe_with_subscription(Arc::new(move |_, _| {
count.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to create thread-safe subscription");
prop_clone
.set_async(thread_id * 10)
.expect("Failed to set value from thread");
thread::sleep(Duration::from_millis(10));
})
})
.collect();
for handle in handles {
handle.join().expect("Thread should complete successfully");
}
thread::sleep(Duration::from_millis(100));
assert!(call_count.load(Ordering::SeqCst) > 0);
}
#[test]
fn test_with_max_threads_error_handling() {
let prop = ObservableProperty::with_max_threads(42, 2);
let _subscription = prop
.subscribe_with_subscription(Arc::new(|_, _| {
}))
.expect("Failed to create subscription for error handling test");
assert!(prop.set(100).is_ok());
assert!(prop.set_async(200).is_ok());
assert_eq!(prop.get().expect("Failed to get value after error test"), 200);
}
#[test]
fn test_observer_limit_enforcement() {
let prop = ObservableProperty::new(0);
let mut observer_ids = Vec::new();
for i in 0..100 {
let result = prop.subscribe(Arc::new(move |_, _| {
let _ = i; }));
assert!(result.is_ok(), "Should be able to add observer {}", i);
observer_ids.push(result.unwrap());
}
assert_eq!(observer_ids.len(), 100);
let result = prop.subscribe(Arc::new(|_, _| {}));
assert!(result.is_ok());
}
#[test]
fn test_observer_limit_error_message() {
let prop = ObservableProperty::new(0);
for _ in 0..10 {
assert!(prop.subscribe(Arc::new(|_, _| {})).is_ok());
}
}
#[test]
fn test_observer_limit_with_unsubscribe() {
let prop = ObservableProperty::new(0);
let mut ids = Vec::new();
for _ in 0..50 {
ids.push(prop.subscribe(Arc::new(|_, _| {})).expect("Failed to subscribe"));
}
for id in ids.iter().take(25) {
assert!(prop.unsubscribe(*id).expect("Failed to unsubscribe"));
}
for _ in 0..30 {
assert!(prop.subscribe(Arc::new(|_, _| {})).is_ok());
}
}
#[test]
fn test_observer_limit_with_raii_subscriptions() {
let prop = ObservableProperty::new(0);
let mut subscriptions = Vec::new();
for _ in 0..50 {
subscriptions.push(
prop.subscribe_with_subscription(Arc::new(|_, _| {}))
.expect("Failed to create subscription")
);
}
subscriptions.truncate(25);
for _ in 0..30 {
let _sub = prop.subscribe_with_subscription(Arc::new(|_, _| {}))
.expect("Failed to create subscription after RAII cleanup");
}
}
#[test]
fn test_filtered_subscription_respects_observer_limit() {
let prop = ObservableProperty::new(0);
for i in 0..50 {
if i % 2 == 0 {
assert!(prop.subscribe(Arc::new(|_, _| {})).is_ok());
} else {
assert!(prop.subscribe_filtered(Arc::new(|_, _| {}), |_, _| true).is_ok());
}
}
assert!(prop.subscribe_filtered(Arc::new(|_, _| {}), |_, _| true).is_ok());
}
#[test]
fn test_observer_limit_concurrent_subscriptions() {
let prop = Arc::new(ObservableProperty::new(0));
let success_count = Arc::new(AtomicUsize::new(0));
let handles: Vec<_> = (0..10)
.map(|_| {
let prop_clone = prop.clone();
let count_clone = success_count.clone();
thread::spawn(move || {
for _ in 0..10 {
if prop_clone.subscribe(Arc::new(|_, _| {})).is_ok() {
count_clone.fetch_add(1, Ordering::SeqCst);
}
thread::sleep(Duration::from_micros(10));
}
})
})
.collect();
for handle in handles {
handle.join().expect("Thread should complete");
}
assert_eq!(success_count.load(Ordering::SeqCst), 100);
}
#[test]
fn test_notify_observers_batch_releases_lock_early() {
use std::sync::atomic::AtomicBool;
let prop = Arc::new(ObservableProperty::new(0));
let call_count = Arc::new(AtomicUsize::new(0));
let started = Arc::new(AtomicBool::new(false));
let started_clone = started.clone();
let count_clone = call_count.clone();
prop.subscribe(Arc::new(move |_, _| {
started_clone.store(true, Ordering::SeqCst);
count_clone.fetch_add(1, Ordering::SeqCst);
thread::sleep(Duration::from_millis(50));
})).expect("Failed to subscribe");
let prop_clone = prop.clone();
let batch_handle = thread::spawn(move || {
prop_clone.notify_observers_batch(vec![(0, 1), (1, 2)]).expect("Failed to notify batch");
});
while !started.load(Ordering::SeqCst) {
thread::sleep(Duration::from_millis(1));
}
let subscribe_result = prop.subscribe(Arc::new(|_, _| {
}));
assert!(subscribe_result.is_ok(), "Should be able to subscribe while batch notification is in progress");
batch_handle.join().expect("Batch thread should complete");
assert_eq!(call_count.load(Ordering::SeqCst), 2);
}
#[test]
fn test_notify_observers_batch_panic_isolation() {
let prop = ObservableProperty::new(0);
let good_observer_count = Arc::new(AtomicUsize::new(0));
let count_clone = good_observer_count.clone();
prop.subscribe(Arc::new(|_, _| {
panic!("Deliberate panic in batch observer");
})).expect("Failed to subscribe panicking observer");
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
})).expect("Failed to subscribe good observer");
let result = prop.notify_observers_batch(vec![(0, 1), (1, 2), (2, 3)]);
assert!(result.is_ok());
assert_eq!(good_observer_count.load(Ordering::SeqCst), 3);
}
#[test]
fn test_notify_observers_batch_multiple_changes() {
let prop = ObservableProperty::new(0);
let received_changes = Arc::new(RwLock::new(Vec::new()));
let changes_clone = received_changes.clone();
prop.subscribe(Arc::new(move |old, new| {
if let Ok(mut changes) = changes_clone.write() {
changes.push((*old, *new));
}
})).expect("Failed to subscribe");
prop.notify_observers_batch(vec![
(0, 10),
(10, 20),
(20, 30),
(30, 40),
]).expect("Failed to notify batch");
let changes = received_changes.read().expect("Failed to read changes");
assert_eq!(changes.len(), 4);
assert_eq!(changes[0], (0, 10));
assert_eq!(changes[1], (10, 20));
assert_eq!(changes[2], (20, 30));
assert_eq!(changes[3], (30, 40));
}
#[test]
fn test_notify_observers_batch_empty() {
let prop = ObservableProperty::new(0);
let call_count = Arc::new(AtomicUsize::new(0));
let count_clone = call_count.clone();
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
})).expect("Failed to subscribe");
prop.notify_observers_batch(vec![]).expect("Failed with empty batch");
assert_eq!(call_count.load(Ordering::SeqCst), 0);
}
#[test]
fn test_debounced_observer_basic() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let last_value = Arc::new(RwLock::new(0));
let count_clone = notification_count.clone();
let value_clone = last_value.clone();
prop.subscribe_debounced(
Arc::new(move |_old, new| {
count_clone.fetch_add(1, Ordering::SeqCst);
if let Ok(mut val) = value_clone.write() {
*val = *new;
}
}),
Duration::from_millis(100)
).expect("Failed to subscribe debounced observer");
prop.set(42).expect("Failed to set value");
assert_eq!(notification_count.load(Ordering::SeqCst), 0);
thread::sleep(Duration::from_millis(150));
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
assert_eq!(*last_value.read().unwrap(), 42);
}
#[test]
fn test_debounced_observer_rapid_changes() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let last_value = Arc::new(RwLock::new(0));
let count_clone = notification_count.clone();
let value_clone = last_value.clone();
prop.subscribe_debounced(
Arc::new(move |_old, new| {
count_clone.fetch_add(1, Ordering::SeqCst);
if let Ok(mut val) = value_clone.write() {
*val = *new;
}
}),
Duration::from_millis(100)
).expect("Failed to subscribe debounced observer");
for i in 1..=10 {
prop.set(i).expect("Failed to set value");
thread::sleep(Duration::from_millis(20)); }
assert_eq!(notification_count.load(Ordering::SeqCst), 0);
thread::sleep(Duration::from_millis(150));
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
assert_eq!(*last_value.read().unwrap(), 10);
}
#[test]
fn test_debounced_observer_multiple_sequences() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let values = Arc::new(RwLock::new(Vec::new()));
let count_clone = notification_count.clone();
let values_clone = values.clone();
prop.subscribe_debounced(
Arc::new(move |_old, new| {
count_clone.fetch_add(1, Ordering::SeqCst);
if let Ok(mut vals) = values_clone.write() {
vals.push(*new);
}
}),
Duration::from_millis(100)
).expect("Failed to subscribe debounced observer");
prop.set(1).expect("Failed to set value");
prop.set(2).expect("Failed to set value");
prop.set(3).expect("Failed to set value");
thread::sleep(Duration::from_millis(150));
prop.set(4).expect("Failed to set value");
prop.set(5).expect("Failed to set value");
thread::sleep(Duration::from_millis(150));
assert_eq!(notification_count.load(Ordering::SeqCst), 2);
let vals = values.read().unwrap();
assert_eq!(vals.len(), 2);
assert_eq!(vals[0], 3); assert_eq!(vals[1], 5); }
#[test]
fn test_debounced_observer_with_string() {
let prop = ObservableProperty::new("".to_string());
let notification_count = Arc::new(AtomicUsize::new(0));
let last_value = Arc::new(RwLock::new(String::new()));
let count_clone = notification_count.clone();
let value_clone = last_value.clone();
prop.subscribe_debounced(
Arc::new(move |_old, new| {
count_clone.fetch_add(1, Ordering::SeqCst);
if let Ok(mut val) = value_clone.write() {
*val = new.clone();
}
}),
Duration::from_millis(100)
).expect("Failed to subscribe debounced observer");
prop.set("H".to_string()).expect("Failed to set value");
thread::sleep(Duration::from_millis(30));
prop.set("He".to_string()).expect("Failed to set value");
thread::sleep(Duration::from_millis(30));
prop.set("Hel".to_string()).expect("Failed to set value");
thread::sleep(Duration::from_millis(30));
prop.set("Hell".to_string()).expect("Failed to set value");
thread::sleep(Duration::from_millis(30));
prop.set("Hello".to_string()).expect("Failed to set value");
assert_eq!(notification_count.load(Ordering::SeqCst), 0);
thread::sleep(Duration::from_millis(150));
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
assert_eq!(*last_value.read().unwrap(), "Hello");
}
#[test]
fn test_debounced_observer_zero_duration() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe_debounced(
Arc::new(move |_old, _new| {
count_clone.fetch_add(1, Ordering::SeqCst);
}),
Duration::from_millis(0)
).expect("Failed to subscribe debounced observer");
prop.set(1).expect("Failed to set value");
thread::sleep(Duration::from_millis(10));
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
}
#[test]
fn test_throttled_observer_basic() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let values = Arc::new(RwLock::new(Vec::new()));
let count_clone = notification_count.clone();
let values_clone = values.clone();
prop.subscribe_throttled(
Arc::new(move |_old, new| {
count_clone.fetch_add(1, Ordering::SeqCst);
if let Ok(mut vals) = values_clone.write() {
vals.push(*new);
}
}),
Duration::from_millis(100)
).expect("Failed to subscribe throttled observer");
prop.set(1).expect("Failed to set value");
thread::sleep(Duration::from_millis(10));
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
prop.set(2).expect("Failed to set value");
thread::sleep(Duration::from_millis(10));
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
thread::sleep(Duration::from_millis(100));
assert_eq!(notification_count.load(Ordering::SeqCst), 2);
let vals = values.read().unwrap();
assert_eq!(vals.len(), 2);
assert_eq!(vals[0], 1);
assert_eq!(vals[1], 2);
}
#[test]
fn test_throttled_observer_continuous_changes() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe_throttled(
Arc::new(move |_old, _new| {
count_clone.fetch_add(1, Ordering::SeqCst);
}),
Duration::from_millis(100)
).expect("Failed to subscribe throttled observer");
for i in 1..=25 {
prop.set(i).expect("Failed to set value");
thread::sleep(Duration::from_millis(20));
}
thread::sleep(Duration::from_millis(150));
let count = notification_count.load(Ordering::SeqCst);
assert!(count >= 4, "Expected at least 4 notifications, got {}", count);
assert!(count <= 10, "Expected at most 10 notifications, got {}", count);
}
#[test]
fn test_throttled_observer_rate_limiting() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let values = Arc::new(RwLock::new(Vec::new()));
let count_clone = notification_count.clone();
let values_clone = values.clone();
prop.subscribe_throttled(
Arc::new(move |_old, new| {
count_clone.fetch_add(1, Ordering::SeqCst);
if let Ok(mut vals) = values_clone.write() {
vals.push(*new);
}
}),
Duration::from_millis(200)
).expect("Failed to subscribe throttled observer");
for i in 1..=20 {
prop.set(i).expect("Failed to set value");
thread::sleep(Duration::from_millis(10));
}
thread::sleep(Duration::from_millis(250));
let count = notification_count.load(Ordering::SeqCst);
assert!(count >= 1, "Expected at least 1 notification, got {}", count);
assert!(count <= 3, "Expected at most 3 notifications, got {}", count);
}
#[test]
fn test_throttled_observer_first_change_immediate() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let first_value = Arc::new(RwLock::new(None));
let count_clone = notification_count.clone();
let value_clone = first_value.clone();
prop.subscribe_throttled(
Arc::new(move |_old, new| {
count_clone.fetch_add(1, Ordering::SeqCst);
if let Ok(mut val) = value_clone.write() {
if val.is_none() {
*val = Some(*new);
}
}
}),
Duration::from_millis(100)
).expect("Failed to subscribe throttled observer");
prop.set(42).expect("Failed to set value");
thread::sleep(Duration::from_millis(10));
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
assert_eq!(*first_value.read().unwrap(), Some(42));
}
#[test]
fn test_throttled_vs_debounced_behavior() {
let prop = ObservableProperty::new(0);
let throttle_count = Arc::new(AtomicUsize::new(0));
let debounce_count = Arc::new(AtomicUsize::new(0));
let throttle_clone = throttle_count.clone();
let debounce_clone = debounce_count.clone();
prop.subscribe_throttled(
Arc::new(move |_old, _new| {
throttle_clone.fetch_add(1, Ordering::SeqCst);
}),
Duration::from_millis(100)
).expect("Failed to subscribe throttled observer");
prop.subscribe_debounced(
Arc::new(move |_old, _new| {
debounce_clone.fetch_add(1, Ordering::SeqCst);
}),
Duration::from_millis(100)
).expect("Failed to subscribe debounced observer");
for i in 1..=30 {
prop.set(i).expect("Failed to set value");
thread::sleep(Duration::from_millis(10));
}
thread::sleep(Duration::from_millis(150));
let throttle_notifications = throttle_count.load(Ordering::SeqCst);
let debounce_notifications = debounce_count.load(Ordering::SeqCst);
assert!(throttle_notifications >= 2,
"Throttled should have multiple notifications, got {}", throttle_notifications);
assert_eq!(debounce_notifications, 1,
"Debounced should have exactly 1 notification, got {}", debounce_notifications);
assert!(throttle_notifications > debounce_notifications,
"Throttled ({}) should have more notifications than debounced ({})",
throttle_notifications, debounce_notifications);
}
#[test]
fn test_throttled_observer_with_long_interval() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe_throttled(
Arc::new(move |_old, _new| {
count_clone.fetch_add(1, Ordering::SeqCst);
}),
Duration::from_secs(1)
).expect("Failed to subscribe throttled observer");
prop.set(1).expect("Failed to set value");
thread::sleep(Duration::from_millis(10));
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
for i in 2..=5 {
prop.set(i).expect("Failed to set value");
thread::sleep(Duration::from_millis(50));
}
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
thread::sleep(Duration::from_millis(1100));
let final_count = notification_count.load(Ordering::SeqCst);
assert!(final_count >= 1 && final_count <= 2,
"Expected 1-2 notifications, got {}", final_count);
}
#[test]
fn test_debounced_and_throttled_combined() {
let prop = ObservableProperty::new(0);
let debounce_values = Arc::new(RwLock::new(Vec::new()));
let throttle_values = Arc::new(RwLock::new(Vec::new()));
let debounce_clone = debounce_values.clone();
let throttle_clone = throttle_values.clone();
prop.subscribe_debounced(
Arc::new(move |_old, new| {
if let Ok(mut vals) = debounce_clone.write() {
vals.push(*new);
}
}),
Duration::from_millis(100)
).expect("Failed to subscribe debounced");
prop.subscribe_throttled(
Arc::new(move |_old, new| {
if let Ok(mut vals) = throttle_clone.write() {
vals.push(*new);
}
}),
Duration::from_millis(100)
).expect("Failed to subscribe throttled");
for i in 1..=10 {
prop.set(i).expect("Failed to set value");
thread::sleep(Duration::from_millis(25));
}
thread::sleep(Duration::from_millis(200));
let debounce_vals = debounce_values.read().unwrap();
let throttle_vals = throttle_values.read().unwrap();
assert_eq!(debounce_vals.len(), 1);
assert_eq!(debounce_vals[0], 10);
assert!(throttle_vals.len() >= 2,
"Throttled should have at least 2 values, got {}", throttle_vals.len());
}
#[test]
fn test_computed_basic() {
let a = Arc::new(ObservableProperty::new(5));
let b = Arc::new(ObservableProperty::new(10));
let sum = computed(
vec![a.clone(), b.clone()],
|values| values[0] + values[1]
).expect("Failed to create computed property");
assert_eq!(sum.get().unwrap(), 15);
a.set(7).expect("Failed to set a");
thread::sleep(Duration::from_millis(10));
assert_eq!(sum.get().unwrap(), 17);
b.set(3).expect("Failed to set b");
thread::sleep(Duration::from_millis(10));
assert_eq!(sum.get().unwrap(), 10);
}
#[test]
fn test_computed_with_observer() {
let width = Arc::new(ObservableProperty::new(10));
let height = Arc::new(ObservableProperty::new(5));
let area = computed(
vec![width.clone(), height.clone()],
|values| values[0] * values[1]
).expect("Failed to create computed property");
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
area.subscribe(Arc::new(move |_old, _new| {
count_clone.fetch_add(1, Ordering::SeqCst);
})).expect("Failed to subscribe");
assert_eq!(area.get().unwrap(), 50);
width.set(20).expect("Failed to set width");
thread::sleep(Duration::from_millis(10));
assert_eq!(area.get().unwrap(), 100);
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
height.set(8).expect("Failed to set height");
thread::sleep(Duration::from_millis(10));
assert_eq!(area.get().unwrap(), 160);
assert_eq!(notification_count.load(Ordering::SeqCst), 2);
}
#[test]
fn test_computed_string_concatenation() {
let first = Arc::new(ObservableProperty::new("Hello".to_string()));
let last = Arc::new(ObservableProperty::new("World".to_string()));
let full = computed(
vec![first.clone(), last.clone()],
|values| format!("{} {}", values[0], values[1])
).expect("Failed to create computed property");
assert_eq!(full.get().unwrap(), "Hello World");
first.set("Goodbye".to_string()).expect("Failed to set first");
thread::sleep(Duration::from_millis(10));
assert_eq!(full.get().unwrap(), "Goodbye World");
last.set("Rust".to_string()).expect("Failed to set last");
thread::sleep(Duration::from_millis(10));
assert_eq!(full.get().unwrap(), "Goodbye Rust");
}
#[test]
fn test_computed_chaining() {
let celsius = Arc::new(ObservableProperty::new(0.0));
let fahrenheit = computed(
vec![celsius.clone()],
|values| values[0] * 9.0 / 5.0 + 32.0
).expect("Failed to create fahrenheit");
let kelvin = computed(
vec![celsius.clone()],
|values| values[0] + 273.15
).expect("Failed to create kelvin");
assert_eq!(celsius.get().unwrap(), 0.0);
assert_eq!(fahrenheit.get().unwrap(), 32.0);
assert_eq!(kelvin.get().unwrap(), 273.15);
celsius.set(100.0).expect("Failed to set celsius");
thread::sleep(Duration::from_millis(10));
assert_eq!(fahrenheit.get().unwrap(), 212.0);
assert_eq!(kelvin.get().unwrap(), 373.15);
}
#[test]
fn test_computed_multiple_dependencies() {
let a = Arc::new(ObservableProperty::new(1));
let b = Arc::new(ObservableProperty::new(2));
let c = Arc::new(ObservableProperty::new(3));
let result = computed(
vec![a.clone(), b.clone(), c.clone()],
|values| values[0] + values[1] * values[2]
).expect("Failed to create computed property");
assert_eq!(result.get().unwrap(), 7);
a.set(5).expect("Failed to set a");
thread::sleep(Duration::from_millis(10));
assert_eq!(result.get().unwrap(), 11);
b.set(4).expect("Failed to set b");
thread::sleep(Duration::from_millis(10));
assert_eq!(result.get().unwrap(), 17);
c.set(2).expect("Failed to set c");
thread::sleep(Duration::from_millis(10));
assert_eq!(result.get().unwrap(), 13);
}
#[test]
fn test_computed_single_dependency() {
let number = Arc::new(ObservableProperty::new(5));
let doubled = computed(
vec![number.clone()],
|values| values[0] * 2
).expect("Failed to create computed property");
assert_eq!(doubled.get().unwrap(), 10);
number.set(7).expect("Failed to set number");
thread::sleep(Duration::from_millis(10));
assert_eq!(doubled.get().unwrap(), 14);
}
#[test]
fn test_computed_complex_calculation() {
let base = Arc::new(ObservableProperty::new(10.0_f64));
let rate = Arc::new(ObservableProperty::new(0.05_f64));
let years = Arc::new(ObservableProperty::new(2.0_f64));
let amount = computed(
vec![base.clone(), rate.clone(), years.clone()],
|values| values[0] * (1.0 + values[1]).powf(values[2])
).expect("Failed to create computed property");
let initial_amount = amount.get().unwrap();
assert!((initial_amount - 11.025_f64).abs() < 0.001);
base.set(100.0_f64).expect("Failed to set base");
thread::sleep(Duration::from_millis(10));
let new_amount = amount.get().unwrap();
assert!((new_amount - 110.25_f64).abs() < 0.001);
years.set(5.0_f64).expect("Failed to set years");
thread::sleep(Duration::from_millis(10));
let final_amount = amount.get().unwrap();
assert!((final_amount - 127.628_f64).abs() < 0.001);
}
#[test]
fn test_update_batch_basic() {
let prop = ObservableProperty::new(0);
let notifications = Arc::new(RwLock::new(Vec::new()));
let notifications_clone = notifications.clone();
prop.subscribe(Arc::new(move |old, new| {
if let Ok(mut notifs) = notifications_clone.write() {
notifs.push((*old, *new));
}
})).expect("Failed to subscribe");
prop.update_batch(|_current| {
vec![10, 20, 30]
}).expect("Failed to update_batch");
let notifs = notifications.read().unwrap();
assert_eq!(notifs.len(), 3);
assert_eq!(notifs[0], (0, 10));
assert_eq!(notifs[1], (10, 20));
assert_eq!(notifs[2], (20, 30));
assert_eq!(prop.get().unwrap(), 30);
}
#[test]
fn test_update_batch_empty_vec() {
let prop = ObservableProperty::new(42);
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
})).expect("Failed to subscribe");
prop.update_batch(|current| {
*current = 100; Vec::new()
}).expect("Failed to update_batch");
assert_eq!(notification_count.load(Ordering::SeqCst), 0);
assert_eq!(prop.get().unwrap(), 42); }
#[test]
fn test_update_batch_single_state() {
let prop = ObservableProperty::new(5);
let notifications = Arc::new(RwLock::new(Vec::new()));
let notifications_clone = notifications.clone();
prop.subscribe(Arc::new(move |old, new| {
if let Ok(mut notifs) = notifications_clone.write() {
notifs.push((*old, *new));
}
})).expect("Failed to subscribe");
prop.update_batch(|_current| {
vec![10]
}).expect("Failed to update_batch");
let notifs = notifications.read().unwrap();
assert_eq!(notifs.len(), 1);
assert_eq!(notifs[0], (5, 10));
assert_eq!(prop.get().unwrap(), 10);
}
#[test]
fn test_update_batch_string_transformation() {
let prop = ObservableProperty::new(String::from("hello"));
let notifications = Arc::new(RwLock::new(Vec::new()));
let notifications_clone = notifications.clone();
prop.subscribe(Arc::new(move |old, new| {
if let Ok(mut notifs) = notifications_clone.write() {
notifs.push((old.clone(), new.clone()));
}
})).expect("Failed to subscribe");
prop.update_batch(|current| {
let step1 = current.to_uppercase();
let step2 = format!("{}!", step1);
let step3 = format!("{} WORLD", step2);
vec![step1, step2, step3]
}).expect("Failed to update_batch");
let notifs = notifications.read().unwrap();
assert_eq!(notifs.len(), 3);
assert_eq!(notifs[0].0, "hello");
assert_eq!(notifs[0].1, "HELLO");
assert_eq!(notifs[1].0, "HELLO");
assert_eq!(notifs[1].1, "HELLO!");
assert_eq!(notifs[2].0, "HELLO!");
assert_eq!(notifs[2].1, "HELLO! WORLD");
assert_eq!(prop.get().unwrap(), "HELLO! WORLD");
}
#[test]
fn test_update_batch_multiple_observers() {
let prop = ObservableProperty::new(0);
let count1 = Arc::new(AtomicUsize::new(0));
let count2 = Arc::new(AtomicUsize::new(0));
let count1_clone = count1.clone();
let count2_clone = count2.clone();
prop.subscribe(Arc::new(move |_, _| {
count1_clone.fetch_add(1, Ordering::SeqCst);
})).expect("Failed to subscribe observer 1");
prop.subscribe(Arc::new(move |_, _| {
count2_clone.fetch_add(1, Ordering::SeqCst);
})).expect("Failed to subscribe observer 2");
prop.update_batch(|_current| {
vec![1, 2, 3, 4, 5]
}).expect("Failed to update_batch");
assert_eq!(count1.load(Ordering::SeqCst), 5);
assert_eq!(count2.load(Ordering::SeqCst), 5);
assert_eq!(prop.get().unwrap(), 5);
}
#[test]
fn test_update_batch_with_panicking_observer() {
let prop = ObservableProperty::new(0);
let good_observer_count = Arc::new(AtomicUsize::new(0));
let count_clone = good_observer_count.clone();
prop.subscribe(Arc::new(|old, _new| {
if *old == 1 {
panic!("Observer panic!");
}
})).expect("Failed to subscribe panicking observer");
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
})).expect("Failed to subscribe good observer");
prop.update_batch(|_current| {
vec![1, 2, 3]
}).expect("Failed to update_batch");
assert_eq!(good_observer_count.load(Ordering::SeqCst), 3);
assert_eq!(prop.get().unwrap(), 3);
}
#[test]
fn test_update_batch_thread_safety() {
let prop = Arc::new(ObservableProperty::new(0));
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
})).expect("Failed to subscribe");
let handles: Vec<_> = (0..5).map(|i| {
let prop_clone = prop.clone();
thread::spawn(move || {
prop_clone.update_batch(|_current| {
vec![i * 10 + 1, i * 10 + 2, i * 10 + 3]
}).expect("Failed to update_batch in thread");
})
}).collect();
for handle in handles {
handle.join().unwrap();
}
assert_eq!(notification_count.load(Ordering::SeqCst), 15);
}
#[test]
fn test_update_batch_with_weak_observers() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
let observer: Arc<dyn Fn(&i32, &i32) + Send + Sync> = Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
});
prop.subscribe_weak(Arc::downgrade(&observer))
.expect("Failed to subscribe weak observer");
prop.update_batch(|_current| {
vec![1, 2, 3]
}).expect("Failed to update_batch");
assert_eq!(notification_count.load(Ordering::SeqCst), 3);
drop(observer);
prop.update_batch(|_current| {
vec![4, 5, 6]
}).expect("Failed to update_batch");
assert_eq!(notification_count.load(Ordering::SeqCst), 3);
assert_eq!(prop.get().unwrap(), 6);
}
#[test]
fn test_change_coalescing_basic() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let last_old = Arc::new(RwLock::new(0));
let last_new = Arc::new(RwLock::new(0));
let count_clone = notification_count.clone();
let old_clone = last_old.clone();
let new_clone = last_new.clone();
prop.subscribe(Arc::new(move |old, new| {
count_clone.fetch_add(1, Ordering::SeqCst);
*old_clone.write().unwrap() = *old;
*new_clone.write().unwrap() = *new;
}))
.expect("Failed to subscribe");
prop.begin_update().expect("Failed to begin update");
prop.set(10).expect("Failed to set value");
prop.set(20).expect("Failed to set value");
prop.set(30).expect("Failed to set value");
assert_eq!(notification_count.load(Ordering::SeqCst), 0);
assert_eq!(prop.get().unwrap(), 30);
prop.end_update().expect("Failed to end update");
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
assert_eq!(*last_old.read().unwrap(), 0);
assert_eq!(*last_new.read().unwrap(), 30);
}
#[test]
fn test_change_coalescing_nested() {
let prop = ObservableProperty::new(100);
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to subscribe");
prop.begin_update().expect("Failed to begin update");
prop.set(110).expect("Failed to set");
prop.begin_update().expect("Failed to begin nested update");
prop.set(120).expect("Failed to set");
prop.set(130).expect("Failed to set");
prop.end_update().expect("Failed to end nested update");
assert_eq!(notification_count.load(Ordering::SeqCst), 0);
prop.set(140).expect("Failed to set");
prop.end_update().expect("Failed to end outer update");
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
assert_eq!(prop.get().unwrap(), 140);
}
#[test]
fn test_change_coalescing_without_begin() {
let prop = ObservableProperty::new(0);
let result = prop.end_update();
assert!(result.is_err());
if let Err(PropertyError::InvalidConfiguration { reason }) = result {
assert!(reason.contains("without matching begin_update"));
} else {
panic!("Expected InvalidConfiguration error");
}
}
#[test]
fn test_change_coalescing_multiple_cycles() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to subscribe");
prop.begin_update().expect("Failed to begin update 1");
prop.set(10).expect("Failed to set");
prop.set(20).expect("Failed to set");
prop.end_update().expect("Failed to end update 1");
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
prop.begin_update().expect("Failed to begin update 2");
prop.set(30).expect("Failed to set");
prop.set(40).expect("Failed to set");
prop.end_update().expect("Failed to end update 2");
assert_eq!(notification_count.load(Ordering::SeqCst), 2);
prop.set(50).expect("Failed to set");
assert_eq!(notification_count.load(Ordering::SeqCst), 3);
}
#[test]
fn test_change_coalescing_with_async() {
let prop = ObservableProperty::new(0);
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to subscribe");
prop.begin_update().expect("Failed to begin update");
prop.set(10).expect("Failed to set");
prop.set_async(20).expect("Failed to set async");
prop.set(30).expect("Failed to set");
assert_eq!(notification_count.load(Ordering::SeqCst), 0);
prop.end_update().expect("Failed to end update");
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
}
#[test]
fn test_change_coalescing_thread_safety() {
use std::thread;
let prop = Arc::new(ObservableProperty::new(0));
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
}))
.expect("Failed to subscribe");
let prop_clone = prop.clone();
let handle = thread::spawn(move || {
prop_clone.begin_update().expect("Failed to begin update");
prop_clone.set(10).expect("Failed to set");
prop_clone.set(20).expect("Failed to set");
prop_clone.end_update().expect("Failed to end update");
});
handle.join().expect("Thread panicked");
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
assert_eq!(prop.get().unwrap(), 20);
}
#[test]
fn test_with_validator_basic() {
let result = ObservableProperty::with_validator(25, |age| {
if *age <= 150 {
Ok(())
} else {
Err(format!("Age must be at most 150, got {}", age))
}
});
assert!(result.is_ok());
let prop = result.unwrap();
assert_eq!(prop.get().unwrap(), 25);
assert!(prop.set(30).is_ok());
assert_eq!(prop.get().unwrap(), 30);
let invalid_result = prop.set(200);
assert!(invalid_result.is_err());
assert_eq!(prop.get().unwrap(), 30); }
#[test]
fn test_with_validator_rejects_invalid_initial_value() {
let result = ObservableProperty::with_validator(200, |age| {
if *age <= 150 {
Ok(())
} else {
Err(format!("Age must be at most 150, got {}", age))
}
});
assert!(result.is_err());
match result {
Err(PropertyError::ValidationError { reason }) => {
assert!(reason.contains("200"));
}
_ => panic!("Expected ValidationError"),
}
}
#[test]
fn test_with_validator_string_validation() {
let result = ObservableProperty::with_validator("alice".to_string(), |name| {
if name.is_empty() {
return Err("Username cannot be empty".to_string());
}
if name.len() < 3 {
return Err(format!("Username must be at least 3 characters, got {}", name.len()));
}
if !name.chars().all(|c| c.is_alphanumeric() || c == '_') {
return Err("Username can only contain letters, numbers, and underscores".to_string());
}
Ok(())
});
assert!(result.is_ok());
let prop = result.unwrap();
assert!(prop.set("bob".to_string()).is_ok());
assert!(prop.set("ab".to_string()).is_err()); assert!(prop.set("user@123".to_string()).is_err()); assert_eq!(prop.get().unwrap(), "bob");
}
#[test]
fn test_with_validator_with_observers() {
let prop = ObservableProperty::with_validator(10, |val| {
if *val >= 0 && *val <= 100 {
Ok(())
} else {
Err(format!("Value must be between 0 and 100, got {}", val))
}
}).unwrap();
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
})).unwrap();
prop.set(50).unwrap();
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
prop.set(150).unwrap_err();
assert_eq!(notification_count.load(Ordering::SeqCst), 1); }
#[test]
fn test_with_equality_basic() {
let prop = ObservableProperty::with_equality(10i32, |a, b| (a - b).abs() <= 5);
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
})).unwrap();
prop.set(12).unwrap();
assert_eq!(notification_count.load(Ordering::SeqCst), 0);
prop.set(20).unwrap();
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
}
#[test]
fn test_with_equality_string_case_insensitive() {
let prop = ObservableProperty::with_equality("Hello".to_string(), |a, b| {
a.to_lowercase() == b.to_lowercase()
});
let notification_count = Arc::new(AtomicUsize::new(0));
let count_clone = notification_count.clone();
prop.subscribe(Arc::new(move |_, _| {
count_clone.fetch_add(1, Ordering::SeqCst);
})).unwrap();
prop.set("hello".to_string()).unwrap();
assert_eq!(notification_count.load(Ordering::SeqCst), 0);
prop.set("World".to_string()).unwrap();
assert_eq!(notification_count.load(Ordering::SeqCst), 1);
}
#[test]
fn test_with_history_basic() {
let prop = ObservableProperty::with_history(0, 5);
prop.set(10).unwrap();
prop.set(20).unwrap();
prop.set(30).unwrap();
assert_eq!(prop.get().unwrap(), 30);
prop.undo().unwrap();
assert_eq!(prop.get().unwrap(), 20);
prop.undo().unwrap();
assert_eq!(prop.get().unwrap(), 10);
prop.undo().unwrap();
assert_eq!(prop.get().unwrap(), 0);
}
#[test]
fn test_with_history_get_history() {
let prop = ObservableProperty::with_history("start".to_string(), 3);
prop.set("second".to_string()).unwrap();
prop.set("third".to_string()).unwrap();
prop.set("fourth".to_string()).unwrap();
let history = prop.get_history();
assert_eq!(history.len(), 3);
assert_eq!(history[0], "start");
assert_eq!(history[1], "second");
assert_eq!(history[2], "third");
assert_eq!(prop.get().unwrap(), "fourth");
}
#[test]
fn test_with_history_bounded_buffer() {
let prop = ObservableProperty::with_history(1, 2);
prop.set(2).unwrap();
prop.set(3).unwrap();
prop.set(4).unwrap();
let history = prop.get_history();
assert_eq!(history.len(), 2);
assert_eq!(history[0], 2);
assert_eq!(history[1], 3);
assert_eq!(prop.get().unwrap(), 4);
}
#[test]
fn test_undo_no_history() {
let prop = ObservableProperty::new(42);
let result = prop.undo();
assert!(result.is_err());
}
#[test]
fn test_undo_empty_history() {
let prop = ObservableProperty::with_history(42, 5);
let result = prop.undo();
assert!(result.is_err());
}
#[test]
fn test_history_with_observers() {
let prop = ObservableProperty::with_history(0, 5);
let notifications = Arc::new(RwLock::new(Vec::new()));
let notifs_clone = notifications.clone();
prop.subscribe(Arc::new(move |old, new| {
if let Ok(mut notifs) = notifs_clone.write() {
notifs.push((*old, *new));
}
})).unwrap();
prop.set(10).unwrap();
prop.set(20).unwrap();
prop.undo().unwrap();
let notifs = notifications.read().unwrap();
assert_eq!(notifs.len(), 3);
assert_eq!(notifs[0], (0, 10));
assert_eq!(notifs[1], (10, 20));
assert_eq!(notifs[2], (20, 10)); }
#[test]
fn test_with_event_log_basic() {
let counter = ObservableProperty::with_event_log(0, 0);
counter.set(1).unwrap();
counter.set(2).unwrap();
counter.set(3).unwrap();
let events = counter.get_event_log();
assert_eq!(events.len(), 3);
assert_eq!(events[0].old_value, 0);
assert_eq!(events[0].new_value, 1);
assert_eq!(events[0].event_number, 0);
assert_eq!(events[2].old_value, 2);
assert_eq!(events[2].new_value, 3);
assert_eq!(events[2].event_number, 2);
}
#[test]
fn test_with_event_log_bounded() {
let prop = ObservableProperty::with_event_log(100, 3);
prop.set(101).unwrap();
prop.set(102).unwrap();
prop.set(103).unwrap();
prop.set(104).unwrap();
let events = prop.get_event_log();
assert_eq!(events.len(), 3);
assert_eq!(events[0].old_value, 101);
assert_eq!(events[2].new_value, 104);
}
#[test]
fn test_event_log_timestamps() {
let prop = ObservableProperty::with_event_log(0, 0);
let before = Instant::now();
thread::sleep(Duration::from_millis(10));
prop.set(1).unwrap();
thread::sleep(Duration::from_millis(10));
prop.set(2).unwrap();
let after = Instant::now();
let events = prop.get_event_log();
assert_eq!(events.len(), 2);
assert!(events[0].timestamp >= before);
assert!(events[1].timestamp <= after);
assert!(events[1].timestamp >= events[0].timestamp);
}
#[test]
fn test_map_basic() {
let celsius = ObservableProperty::new(20.0);
let fahrenheit = celsius.map(|c| c * 9.0 / 5.0 + 32.0).unwrap();
assert_eq!(fahrenheit.get().unwrap(), 68.0);
celsius.set(25.0).unwrap();
thread::sleep(Duration::from_millis(10)); assert_eq!(fahrenheit.get().unwrap(), 77.0);
celsius.set(0.0).unwrap();
thread::sleep(Duration::from_millis(10));
assert_eq!(fahrenheit.get().unwrap(), 32.0);
}
#[test]
fn test_map_string_formatting() {
let count = ObservableProperty::new(42);
let message = count.map(|n| format!("Count: {}", n)).unwrap();
assert_eq!(message.get().unwrap(), "Count: 42");
count.set(100).unwrap();
thread::sleep(Duration::from_millis(10));
assert_eq!(message.get().unwrap(), "Count: 100");
}
#[test]
fn test_map_chaining() {
let base = ObservableProperty::new(10);
let doubled = base.map(|x| x * 2).unwrap();
let squared = doubled.map(|x| x * x).unwrap();
assert_eq!(squared.get().unwrap(), 400);
base.set(5).unwrap();
thread::sleep(Duration::from_millis(20));
assert_eq!(squared.get().unwrap(), 100); }
#[test]
fn test_map_type_conversion() {
let integer = ObservableProperty::new(42);
let float_value = integer.map(|i| *i as f64).unwrap();
let is_even = integer.map(|i| i % 2 == 0).unwrap();
assert_eq!(float_value.get().unwrap(), 42.0);
assert_eq!(is_even.get().unwrap(), true);
integer.set(43).unwrap();
thread::sleep(Duration::from_millis(10));
assert_eq!(is_even.get().unwrap(), false);
}
#[test]
fn test_modify_basic() {
let counter = ObservableProperty::new(0);
let notifications = Arc::new(RwLock::new(Vec::new()));
let notifs_clone = notifications.clone();
counter.subscribe(Arc::new(move |old, new| {
if let Ok(mut notifs) = notifs_clone.write() {
notifs.push((*old, *new));
}
})).unwrap();
counter.modify(|value| *value += 1).unwrap();
assert_eq!(counter.get().unwrap(), 1);
counter.modify(|value| *value *= 2).unwrap();
assert_eq!(counter.get().unwrap(), 2);
let notifs = notifications.read().unwrap();
assert_eq!(notifs.len(), 2);
assert_eq!(notifs[0], (0, 1));
assert_eq!(notifs[1], (1, 2));
}
#[test]
fn test_modify_with_validator() {
let prop = ObservableProperty::with_validator(10, |val| {
if *val >= 0 && *val <= 100 {
Ok(())
} else {
Err("Value must be between 0 and 100".to_string())
}
}).unwrap();
assert!(prop.modify(|v| *v += 5).is_ok());
assert_eq!(prop.get().unwrap(), 15);
let result = prop.modify(|v| *v += 100);
assert!(result.is_err());
assert_eq!(prop.get().unwrap(), 15); }
#[test]
fn test_modify_string() {
let text = ObservableProperty::new("hello".to_string());
text.modify(|s| {
*s = s.to_uppercase();
}).unwrap();
assert_eq!(text.get().unwrap(), "HELLO");
text.modify(|s| {
s.push_str(" WORLD");
}).unwrap();
assert_eq!(text.get().unwrap(), "HELLO WORLD");
}
#[test]
fn test_bind_bidirectional_basic() {
let prop1 = Arc::new(ObservableProperty::new(10));
let prop2 = Arc::new(ObservableProperty::new(10));
prop1.bind_bidirectional(&prop2).unwrap();
prop1.set(30).unwrap();
thread::sleep(Duration::from_millis(20));
assert_eq!(prop2.get().unwrap(), 30);
prop2.set(40).unwrap();
thread::sleep(Duration::from_millis(20));
assert_eq!(prop1.get().unwrap(), 40);
}
#[test]
fn test_bind_bidirectional_strings() {
let prop1 = Arc::new(ObservableProperty::new("first".to_string()));
let prop2 = Arc::new(ObservableProperty::new("first".to_string()));
prop1.bind_bidirectional(&prop2).unwrap();
prop2.set("updated".to_string()).unwrap();
thread::sleep(Duration::from_millis(20));
assert_eq!(prop1.get().unwrap(), "updated");
prop1.set("final".to_string()).unwrap();
thread::sleep(Duration::from_millis(20));
assert_eq!(prop2.get().unwrap(), "final");
}
#[test]
fn test_get_metrics_basic() {
let prop = ObservableProperty::new(0);
prop.subscribe(Arc::new(|_, _| {
thread::sleep(Duration::from_millis(5));
})).unwrap();
prop.set(1).unwrap();
prop.set(2).unwrap();
prop.set(3).unwrap();
let metrics = prop.get_metrics().unwrap();
assert_eq!(metrics.total_changes, 3);
assert_eq!(metrics.observer_calls, 3);
assert!(metrics.avg_notification_time.as_millis() >= 4);
}
#[test]
fn test_metrics_multiple_observers() {
let prop = ObservableProperty::new(0);
for _ in 0..3 {
prop.subscribe(Arc::new(|_, _| {})).unwrap();
}
prop.set(1).unwrap();
prop.set(2).unwrap();
let metrics = prop.get_metrics().unwrap();
assert_eq!(metrics.total_changes, 2);
assert_eq!(metrics.observer_calls, 6); }
#[test]
fn test_metrics_no_observers() {
let prop = ObservableProperty::new(0);
prop.set(1).unwrap();
prop.set(2).unwrap();
let metrics = prop.get_metrics().unwrap();
assert_eq!(metrics.total_changes, 2);
assert_eq!(metrics.observer_calls, 0);
}
#[test]
#[cfg(feature = "debug")]
fn test_enable_disable_change_logging() {
let prop = ObservableProperty::new(0);
prop.enable_change_logging();
prop.set(1).unwrap();
prop.set(2).unwrap();
prop.disable_change_logging();
prop.set(3).unwrap();
}
#[test]
fn test_observer_count() {
let prop = ObservableProperty::new(0);
assert_eq!(prop.observer_count(), 0);
let _id1 = prop.subscribe(Arc::new(|_, _| {})).unwrap();
assert_eq!(prop.observer_count(), 1);
let _id2 = prop.subscribe(Arc::new(|_, _| {})).unwrap();
assert_eq!(prop.observer_count(), 2);
let _id3 = prop.subscribe(Arc::new(|_, _| {})).unwrap();
assert_eq!(prop.observer_count(), 3);
}
#[test]
fn test_observer_count_after_unsubscribe() {
let prop = ObservableProperty::new(0);
let id1 = prop.subscribe(Arc::new(|_, _| {})).unwrap();
let id2 = prop.subscribe(Arc::new(|_, _| {})).unwrap();
assert_eq!(prop.observer_count(), 2);
prop.unsubscribe(id1).unwrap();
assert_eq!(prop.observer_count(), 1);
prop.unsubscribe(id2).unwrap();
assert_eq!(prop.observer_count(), 0);
}
#[test]
fn test_with_config_custom_limits() {
let prop = ObservableProperty::with_config(42, 2, 5);
for _ in 0..5 {
assert!(prop.subscribe(Arc::new(|_, _| {})).is_ok());
}
assert_eq!(prop.observer_count(), 5);
let result = prop.subscribe(Arc::new(|_, _| {}));
assert!(result.is_err());
}
#[test]
fn test_with_config_max_threads() {
let prop = ObservableProperty::with_config(0, 1, 100);
let call_count = Arc::new(AtomicUsize::new(0));
for _ in 0..4 {
let count = call_count.clone();
prop.subscribe(Arc::new(move |_, _| {
thread::sleep(Duration::from_millis(25));
count.fetch_add(1, Ordering::SeqCst);
})).unwrap();
}
let start = Instant::now();
prop.set_async(42).unwrap();
let duration = start.elapsed();
assert!(duration.as_millis() < 50);
thread::sleep(Duration::from_millis(150));
assert_eq!(call_count.load(Ordering::SeqCst), 4);
}
struct MockPersistence {
data: Arc<RwLock<Option<i32>>>,
}
impl PropertyPersistence for MockPersistence {
type Value = i32;
fn load(&self) -> Result<Self::Value, Box<dyn std::error::Error + Send + Sync>> {
self.data
.read()
.unwrap()
.ok_or_else(|| "No data".into())
}
fn save(&self, value: &Self::Value) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
*self.data.write().unwrap() = Some(*value);
Ok(())
}
}
#[test]
fn test_with_persistence_load_success() {
let storage = Arc::new(RwLock::new(Some(42)));
let persistence = MockPersistence { data: storage.clone() };
let prop = ObservableProperty::with_persistence(0, persistence);
assert_eq!(prop.get().unwrap(), 42);
}
#[test]
fn test_with_persistence_auto_save() {
let storage = Arc::new(RwLock::new(None));
let persistence = MockPersistence { data: storage.clone() };
let prop = ObservableProperty::with_persistence(10, persistence);
prop.set(20).unwrap();
thread::sleep(Duration::from_millis(10));
assert_eq!(*storage.read().unwrap(), Some(20));
prop.set(30).unwrap();
thread::sleep(Duration::from_millis(10));
assert_eq!(*storage.read().unwrap(), Some(30));
}
#[test]
fn test_with_persistence_load_failure_uses_default() {
let storage = Arc::new(RwLock::new(None));
let persistence = MockPersistence { data: storage };
let prop = ObservableProperty::with_persistence(99, persistence);
assert_eq!(prop.get().unwrap(), 99);
}
#[test]
fn test_computed_updates_immediately() {
let a = Arc::new(ObservableProperty::new(5));
let b = Arc::new(ObservableProperty::new(10));
let sum = computed(
vec![a.clone(), b.clone()],
|values| values[0] + values[1]
).unwrap();
assert_eq!(sum.get().unwrap(), 15);
a.set(7).unwrap();
thread::sleep(Duration::from_millis(10));
assert_eq!(sum.get().unwrap(), 17);
}
#[test]
fn test_computed_with_string() {
let first = Arc::new(ObservableProperty::new("Hello".to_string()));
let last = Arc::new(ObservableProperty::new("World".to_string()));
let full = computed(
vec![first.clone(), last.clone()],
|values| format!("{} {}", values[0], values[1])
).unwrap();
assert_eq!(full.get().unwrap(), "Hello World");
first.set("Goodbye".to_string()).unwrap();
thread::sleep(Duration::from_millis(10));
assert_eq!(full.get().unwrap(), "Goodbye World");
}
}