rust_widgets 1.1.3

// SPDX-FileCopyrightText: Copyright (c) 2026 Mike Li/Mikewolfli/Wei Li(mikewolfli@163.com)
// SPDX-License-Identifier: MIT

use crate::compat::HashMap;
use crate::compat::Mutex;
use crate::data_binding::traits::*;
use alloc::sync::{Arc, Weak};
use core::sync::atomic::{AtomicBool, Ordering};

/// A reactive binding that notifies listeners when the value changes.
///
/// Think of it as a "signal + value" container. When the value is updated via
/// [`set`](Binding::set), all registered listeners are notified with their
/// subscription key.
///
/// Internally the binding's state is guarded by a `Mutex`, so all mutation
/// methods take `&self` rather than `&mut self`.  This makes it possible to
/// safely share a `Binding` across threads or use it with shared references.
pub struct Binding<T: Clone + Send + 'static> {
    inner: Arc<Mutex<BindingInner<T>>>,
}

struct BindingInner<T: Clone + Send + 'static> {
    value: T,
    listeners: HashMap<String, BoxedListener>,
}

impl<T: Clone + Send + 'static> Binding<T> {
    /// Create a new binding with an initial value.
    pub fn new(value: T) -> Self {
        Self { inner: Arc::new(Mutex::new(BindingInner { value, listeners: HashMap::new() })) }
    }

    /// Get the current value.
    ///
    /// For `T: Copy` types, use [`get_copy`](Self::get_copy) to avoid the clone.
    #[inline(always)]
    pub fn get(&self) -> T {
        self.inner.lock().unwrap_or_else(|e| e.into_inner()).value.clone()
    }

    /// Set a new value and notify all listeners.
    ///
    /// Notifications are dispatched **outside** the Mutex lock to prevent
    /// re-entrancy deadlocks (e.g. when a TwoWayListener tries to lock the
    /// same binding's Mutex while propagating a value change).
    ///
    /// Listeners are temporarily removed from the map, notified, then
    /// restored (unless a new listener was subscribed under the same key
    /// during notification, in which case the new one takes precedence).
    pub fn set(&self, value: T) {
        // ── Phase 1: Lock, update value, take all listeners ──
        let mut listeners: Vec<(String, BoxedListener)>;
        {
            let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
            inner.value = value;
            listeners = core::mem::take(&mut inner.listeners).into_iter().collect();
        } // Mutex lock released.

        // ── Phase 2: Notify outside lock (safe from re-entrancy) ──
        for (key, ref mut listener) in &mut listeners {
            listener.on_value_changed(key, "set");
        }

        // ── Phase 3: Restore listeners that weren't re-subscribed ──
        {
            let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
            for (key, listener) in listeners {
                // If no new listener was subscribed under this key
                // during notification, put the original one back.
                inner.listeners.entry(key).or_insert(listener);
            }
        }
    }

    /// Subscribe to value changes.
    ///
    /// `key` is an identifier used to later unsubscribe. If a listener with
    /// the same key already exists, it is replaced.
    pub fn subscribe(&self, key: &str, listener: BoxedListener) {
        self.inner
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .listeners
            .insert(key.to_string(), listener);
    }

    /// Remove a listener by its subscription key.
    pub fn unsubscribe(&self, key: &str) {
        self.inner.lock().unwrap_or_else(|e| e.into_inner()).listeners.remove(key);
    }

    /// Create a two-way binding between this binding and another.
    ///
    /// Whenever either binding's value changes, the other is updated to match.
    /// Uses an atomic synchronization guard to prevent infinite notification
    /// loops. The two-way connection uses `Weak` references to avoid reference
    /// cycles and prevent use-after-free if one binding is dropped.
    pub fn bind_to(&self, other: &Binding<T>)
    where
        T: PartialEq,
    {
        let syncing = Arc::new(AtomicBool::new(false));

        let self_weak = Arc::downgrade(&self.inner);
        let other_weak = Arc::downgrade(&other.inner);

        let listener_self_key = format!("__two_way_self_{:p}", Arc::as_ptr(&self.inner));
        let listener_other_key = format!("__two_way_other_{:p}", Arc::as_ptr(&other.inner));

        self.subscribe(
            &listener_self_key,
            Box::new(TwoWayListener::new(syncing.clone(), self_weak.clone(), other_weak.clone())),
        );
        other.subscribe(
            &listener_other_key,
            Box::new(TwoWayListener::new(syncing, other_weak, self_weak)),
        );
    }

    /// Get the current value without cloning (available for `Copy` types).
    ///
    /// This avoids the `.clone()` that [`get`](Self::get) always performs.
    #[inline(always)]
    pub fn get_copy(&self) -> T
    where
        T: Copy,
    {
        self.inner.lock().unwrap_or_else(|e| e.into_inner()).value
    }

    /// Return the number of currently registered listeners.
    pub fn listener_count(&self) -> usize {
        self.inner.lock().unwrap_or_else(|e| e.into_inner()).listeners.len()
    }
}

impl<T: Clone + Send + 'static> BindingInner<T> {
    /// Set value without notifying listeners.
    /// Used by TwoWayListener to propagate changes silently.
    fn set_no_notify(&mut self, value: T) {
        self.value = value;
    }
}

/// A listener that propagates value changes from one binding to another.
///
/// Used internally by [`Binding::bind_to`] to implement two-way synchronization.
/// Uses `Weak<Mutex<BindingInner<T>>>` internally so that if one binding is
/// dropped, the listener on the other safely detects this and becomes a no-op.
struct TwoWayListener<T: Clone + Send + 'static> {
    syncing: Arc<AtomicBool>,
    source: Weak<Mutex<BindingInner<T>>>,
    target: Weak<Mutex<BindingInner<T>>>,
}

impl<T: Clone + Send + 'static> TwoWayListener<T> {
    fn new(
        syncing: Arc<AtomicBool>,
        source: Weak<Mutex<BindingInner<T>>>,
        target: Weak<Mutex<BindingInner<T>>>,
    ) -> Self {
        Self { syncing, source, target }
    }
}

impl<T: Clone + Send + 'static + PartialEq> BindingListener for TwoWayListener<T> {
    fn on_value_changed(&mut self, _key: &str, _operation: &str) {
        // Re-entrancy guard: the first thread in wins, and the flag is cleared by
        // the RAII guard below on **every** exit — including a panic from
        // `set_no_notify`. A bare `store(false)` at the end of the body would be
        // skipped if anything panicked, leaving `syncing` stuck at `true` and
        // silently disabling this two-way binding forever.
        if self.syncing.swap(true, Ordering::SeqCst) {
            return;
        }
        let _reset = SyncingGuard { flag: &self.syncing };

        // Read value from source, then release source's Mutex lock BEFORE
        // locking the target.  This avoids a re-entrant-Mutex deadlock when
        // the outer `set()` already holds the source binding's lock.
        let val = self
            .source
            .upgrade()
            .map(|source| source.lock().unwrap_or_else(|e| e.into_inner()).value.clone());

        // If either binding has been dropped, skip gracefully.
        if let Some(val) = val {
            if let Some(target) = self.target.upgrade() {
                target.lock().unwrap_or_else(|e| e.into_inner()).set_no_notify(val);
            }
        }
    }
}

/// Clears the `syncing` flag when dropped, so the re-entrancy guard cannot be
/// left armed by an unwinding panic.
struct SyncingGuard<'a> {
    flag: &'a AtomicBool,
}

impl Drop for SyncingGuard<'_> {
    fn drop(&mut self) {
        self.flag.store(false, Ordering::SeqCst);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::compat::Mutex;
    use core::sync::atomic::AtomicI32;

    #[test]
    fn test_binding_get_set() {
        let b = Binding::new(42);
        assert_eq!(b.get(), 42);
        b.set(100);
        assert_eq!(b.get(), 100);
    }

    /// A panic while propagating must not leave the two-way re-entrancy guard
    /// permanently armed.
    ///
    /// `TwoWayListener` raises `syncing` on entry to suppress the echo from the
    /// reverse direction. If the propagation body panicked, the old code's final
    /// `store(false)` was skipped, so `syncing` stayed `true` and the binding
    /// silently stopped syncing forever. `SyncingGuard` resets it on every exit,
    /// which this test pins down by unwinding through the guarded section.
    #[test]
    fn syncing_guard_clears_flag_on_panic() {
        let flag = AtomicBool::new(false);
        assert!(!flag.swap(true, Ordering::SeqCst), "flag starts unset");

        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _reset = SyncingGuard { flag: &flag };
            panic!("boom inside the guarded section");
        }));

        assert!(result.is_err(), "the probe must panic");
        assert!(
            !flag.load(Ordering::SeqCst),
            "the guard must clear `syncing` while unwinding, otherwise the binding is dead forever"
        );
    }

    /// The guard clears the flag on the normal (non-panicking) path as well.
    #[test]
    fn syncing_guard_clears_flag_on_normal_exit() {
        let flag = AtomicBool::new(false);
        flag.store(true, Ordering::SeqCst);
        {
            let _reset = SyncingGuard { flag: &flag };
        }
        assert!(!flag.load(Ordering::SeqCst), "the guard must clear `syncing` on drop");
    }

    /// A two-way binding still synchronises after its listener ran once.
    #[test]
    fn two_way_listener_propagates_and_rearms() {
        let syncing = Arc::new(AtomicBool::new(false));
        let source: Arc<Mutex<BindingInner<i32>>> =
            Arc::new(Mutex::new(BindingInner { value: 7, listeners: HashMap::new() }));
        let target: Arc<Mutex<BindingInner<i32>>> =
            Arc::new(Mutex::new(BindingInner { value: 0, listeners: HashMap::new() }));

        let mut listener = TwoWayListener::new(
            Arc::clone(&syncing),
            Arc::downgrade(&source),
            Arc::downgrade(&target),
        );

        listener.on_value_changed("k", "set");
        assert_eq!(target.lock().unwrap().value, 7, "the value must propagate source -> target");
        assert!(!syncing.load(Ordering::SeqCst), "the guard must re-arm the listener");

        // A second call must still work (the guard did not get stuck).
        source.lock().unwrap().value = 9;
        listener.on_value_changed("k", "set");
        assert_eq!(target.lock().unwrap().value, 9, "a later change must still propagate");
        assert!(!syncing.load(Ordering::SeqCst));
    }

    #[test]
    fn test_binding_listener_notification() {
        let b = Binding::new("hello".to_string());
        let notified = Arc::new(AtomicBool::new(false));
        let n = notified.clone();
        let listener = Box::new(FnListener::new(move |_key, _op| {
            n.store(true, Ordering::SeqCst);
        }));
        b.subscribe("test", listener);
        b.set("world".to_string());
        assert!(notified.load(Ordering::SeqCst));
    }

    #[test]
    fn test_binding_unsubscribe() {
        let b = Binding::new(0);
        let count = Arc::new(AtomicI32::new(0));
        let c = count.clone();
        let listener = Box::new(FnListener::new(move |_key, _op| {
            c.fetch_add(1, Ordering::SeqCst);
        }));
        b.subscribe("test", listener);
        b.set(1);
        assert_eq!(count.load(Ordering::SeqCst), 1);
        b.unsubscribe("test");
        b.set(2);
        assert_eq!(count.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn test_binding_multiple_listeners() {
        let b = Binding::new(0);
        let count_a = Arc::new(AtomicI32::new(0));
        let count_b = Arc::new(AtomicI32::new(0));

        let ca = count_a.clone();
        b.subscribe(
            "a",
            Box::new(FnListener::new(move |_, _| {
                ca.fetch_add(1, Ordering::SeqCst);
            })),
        );
        let cb = count_b.clone();
        b.subscribe(
            "b",
            Box::new(FnListener::new(move |_, _| {
                cb.fetch_add(1, Ordering::SeqCst);
            })),
        );
        b.set(1);
        assert_eq!(count_a.load(Ordering::SeqCst), 1);
        assert_eq!(count_b.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn test_binding_listener_receive_key() {
        let b = Binding::new(0);
        let received_key = Arc::new(Mutex::new(String::new()));
        let rk = received_key.clone();
        let listener = Box::new(FnListener::new(move |key, _op| {
            *rk.lock().unwrap() = key.to_string();
        }));
        b.subscribe("my_key", listener);
        b.set(99);
        assert_eq!(*received_key.lock().unwrap(), "my_key");
    }

    #[test]
    fn test_binding_two_way_sync() {
        let a = Binding::new(10);
        let b = Binding::new(20);

        a.bind_to(&b);

        // Set a -> b should propagate
        a.set(30);
        assert_eq!(a.get(), 30);
        assert_eq!(b.get(), 30);

        // Set b -> a should propagate
        b.set(50);
        assert_eq!(a.get(), 50);
        assert_eq!(b.get(), 50);
    }

    #[test]
    fn test_binding_two_way_no_infinite_loop() {
        let a = Binding::new(0);
        let b = Binding::new(0);
        let a_count = Arc::new(AtomicI32::new(0));
        let b_count = Arc::new(AtomicI32::new(0));

        let ac = a_count.clone();
        a.subscribe(
            "a_count",
            Box::new(FnListener::new(move |_, _| {
                ac.fetch_add(1, Ordering::SeqCst);
            })),
        );
        let bc = b_count.clone();
        b.subscribe(
            "b_count",
            Box::new(FnListener::new(move |_, _| {
                bc.fetch_add(1, Ordering::SeqCst);
            })),
        );

        a.bind_to(&b);

        // Changing a should notify a's listeners once and propagate to b
        // via set_no_notify (b's listeners are NOT fired by the TwoWayListener).
        a.set(42);
        assert_eq!(a_count.load(Ordering::SeqCst), 1);
        assert_eq!(b_count.load(Ordering::SeqCst), 0);

        // Setting b directly should fire b's listeners
        b.set(100);
        assert_eq!(a_count.load(Ordering::SeqCst), 1); // a's listeners unchanged
        assert_eq!(b_count.load(Ordering::SeqCst), 1); // b's listener fired for b.set()
    }

    #[test]
    fn test_binding_drop_safety() {
        // Verify that dropping one binding doesn't cause UB in the other's listener.
        let a = Arc::new(Binding::new(10));
        let b = Arc::new(Binding::new(20));
        a.bind_to(&b);

        // Drop 'a' — b's listener holds a Weak to a's inner, which should
        // gracefully become a no-op.
        drop(a);

        // Setting b should not panic or cause UB
        b.set(99);
        assert_eq!(b.get(), 99);
    }
}