libobs_wrapper\signals/
mod.rs

1//! Per-object OBS signal subscriptions.
2//!
3//! Signal state used to live in process-global maps keyed by native pointer addresses.
4//! Besides unnecessary global locking, that let callback-only raw pointers escape
5//! through async channels.  A signal manager now owns its own hubs and passes stable
6//! hub addresses directly to libobs as callback data.
7
8mod handler;
9mod traits;
10
11pub use traits::*;
12
13use crossbeam_channel::{bounded, Receiver, RecvError, Sender, TryRecvError, TrySendError};
14use std::sync::Mutex;
15
16const SIGNAL_QUEUE_CAPACITY: usize = 32;
17
18/// Opaque identity of an object mentioned by an OBS callback.
19///
20/// This is deliberately not a raw pointer and cannot be dereferenced. Callback data is
21/// often borrowed only for the duration of the C callback; exposing that pointer as a
22/// sendable Rust value allowed use-after-free in otherwise safe code.
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
24pub struct SignalObjectId(usize);
25
26impl SignalObjectId {
27    /// Construct an opaque callback identity from a callback-borrowed native pointer.
28    ///
29    /// # Safety
30    /// `ptr` must be a pointer supplied by libobs for the current callback. The returned
31    /// identity is never dereferenceable and does not extend the native object's lifetime.
32    #[doc(hidden)]
33    pub unsafe fn from_raw<T>(ptr: *mut T) -> Self {
34        Self(ptr as usize)
35    }
36
37    pub fn is_null(self) -> bool {
38        self.0 == 0
39    }
40}
41
42/// Synchronous receiver for an OBS signal subscription.
43///
44/// Each subscriber has a bounded queue. If it falls behind, new callback events are
45/// dropped for that subscriber rather than blocking the OBS callback thread.
46#[derive(Debug)]
47pub struct SignalReceiver<T> {
48    receiver: Receiver<T>,
49}
50
51impl<T> SignalReceiver<T> {
52    pub fn recv(&self) -> Result<T, RecvError> {
53        self.receiver.recv()
54    }
55
56    /// Compatibility alias for the old Tokio broadcast receiver call sites.
57    pub fn blocking_recv(&self) -> Result<T, RecvError> {
58        self.recv()
59    }
60
61    pub fn try_recv(&self) -> Result<T, TryRecvError> {
62        self.receiver.try_recv()
63    }
64}
65
66/// A small per-object fan-out hub used by generated signal managers.
67#[doc(hidden)]
68#[derive(Debug)]
69pub struct SignalHub<T> {
70    subscribers: Mutex<Vec<Sender<T>>>,
71}
72
73impl<T> Default for SignalHub<T> {
74    fn default() -> Self {
75        Self {
76            subscribers: Mutex::new(Vec::new()),
77        }
78    }
79}
80
81impl<T> SignalHub<T> {
82    #[doc(hidden)]
83    pub fn new() -> Self {
84        Self::default()
85    }
86
87    #[doc(hidden)]
88    pub fn subscribe(&self) -> SignalReceiver<T> {
89        let (tx, rx) = bounded(SIGNAL_QUEUE_CAPACITY);
90        match self.subscribers.lock() {
91            Ok(mut subscribers) => subscribers.push(tx),
92            Err(poisoned) => poisoned.into_inner().push(tx),
93        }
94        SignalReceiver { receiver: rx }
95    }
96}
97
98impl<T: Clone> SignalHub<T> {
99    #[doc(hidden)]
100    pub fn publish(&self, value: T) {
101        let mut subscribers = match self.subscribers.lock() {
102            Ok(subscribers) => subscribers,
103            Err(poisoned) => poisoned.into_inner(),
104        };
105        subscribers.retain(|subscriber| match subscriber.try_send(value.clone()) {
106            Ok(()) | Err(TrySendError::Full(_)) => true,
107            Err(TrySendError::Disconnected(_)) => false,
108        });
109    }
110}
111
112/// Generate a signal manager whose subscription state is owned by that manager.
113#[macro_export]
114macro_rules! impl_signal_manager {
115    ($handler_getter: expr, $name: ident for $ptr: ty, [
116        $($(#[$attr:meta])* $signal_name: literal: { $($inner_def:tt)* }),* $(,)*
117    ]) => {
118        paste::paste! {
119            $($crate::__signals_impl_signal!($ptr, $signal_name, $($inner_def)*);)*
120
121            $(
122                extern "C" fn [<$signal_name:snake _handler>](
123                    hub_ptr: *mut std::ffi::c_void,
124                    __internal_calldata: *mut libobs::calldata_t,
125                ) {
126                    let Some(hub) = (unsafe {
127                        // Safety: libobs receives this address from Arc::as_ptr during
128                        // connect and the manager keeps the Arc alive until disconnect.
129                        (hub_ptr as *const $crate::signals::SignalHub<
130                            [<__Private $signal_name:camel Type>]
131                        >).as_ref()
132                    }) else {
133                        log::warn!("Null signal hub for {}", $signal_name);
134                        return;
135                    };
136
137                    let value = unsafe {
138                        // Safety: libobs invokes the callback with valid calldata for
139                        // the duration of this call.
140                        [<$signal_name:snake _handler_inner>](__internal_calldata)
141                    };
142                    match value {
143                        Ok(value) => hub.publish(value),
144                        Err(err) => log::warn!("Error processing signal {}: {:?}", $signal_name, err),
145                    }
146                }
147            )*
148
149            #[derive(Debug)]
150            pub struct $name {
151                runtime: $crate::runtime::ObsRuntime,
152                pointer: $crate::unsafe_send::SmartPointerSendable<$ptr>,
153                $([<$signal_name:snake _hub>]: std::sync::Arc<
154                    $crate::signals::SignalHub<[<__Private $signal_name:camel Type>]>
155                >,)*
156            }
157
158            impl $name {
159                pub(crate) fn new(
160                    smart_ptr: &$crate::unsafe_send::SmartPointerSendable<$ptr>,
161                    runtime: $crate::runtime::ObsRuntime,
162                ) -> Result<Self, $crate::utils::ObsError> {
163                    use $crate::utils::ObsString;
164                    let smart_ptr = smart_ptr.clone();
165                    $(
166                        let [<$signal_name:snake _hub>] = std::sync::Arc::new(
167                            $crate::signals::SignalHub::<[<__Private $signal_name:camel Type>]>::new()
168                        );
169                    )*
170
171                    // Connect the complete manager in one actor command. A queue/shutdown
172                    // failure therefore happens before any callback userdata is registered,
173                    // eliminating partially-constructed managers with dangling C userdata.
174                    let connect_ptr = smart_ptr.clone();
175                    $(let [<$signal_name:snake _connect_hub>] = [<$signal_name:snake _hub>].clone();)*
176                    $crate::run_with_obs!(
177                        runtime,
178                        (connect_ptr, $([<$signal_name:snake _connect_hub>],)*),
179                        move || {
180                            $(
181                                let handler = ($handler_getter)(connect_ptr.clone());
182                                let signal = ObsString::new($signal_name);
183                                unsafe {
184                                    // SAFETY: handler and signal are valid on the OBS actor;
185                                    // each Arc allocation has a stable address until disconnect.
186                                    libobs::signal_handler_connect(
187                                        handler,
188                                        signal.as_c_str().as_ptr(),
189                                        Some([<$signal_name:snake _handler>]),
190                                        std::sync::Arc::as_ptr(&[<$signal_name:snake _connect_hub>])
191                                            as *mut std::ffi::c_void,
192                                    );
193                                }
194                            )*
195                        }
196                    )?;
197
198                    Ok(Self {
199                        pointer: smart_ptr,
200                        runtime,
201                        $([<$signal_name:snake _hub>],)*
202                    })
203                }
204
205                $(
206                    $(#[$attr])*
207                    pub fn [<on_ $signal_name:snake>](&self) -> Result<
208                        $crate::signals::SignalReceiver<[<__Private $signal_name:camel Type>]>,
209                        $crate::utils::ObsError,
210                    > {
211                        Ok(self.[<$signal_name:snake _hub>].subscribe())
212                    }
213                )*
214            }
215
216            impl Drop for $name {
217                fn drop(&mut self) {
218                    log::trace!("Dropping signal manager {}...", stringify!($name));
219                    let ptr = self.pointer.clone();
220                    let runtime = self.runtime.clone();
221                    $(let [<$signal_name:snake _hub>] = self.[<$signal_name:snake _hub>].clone();)*
222
223                    let cleanup_runtime = runtime.clone();
224                    runtime.defer_obs_cleanup(move || {
225                        let result = cleanup_runtime.run_with_obs_result(move || {
226                            let handler = ($handler_getter)(ptr.clone());
227                            $(
228                                let signal = $crate::utils::ObsString::new($signal_name);
229                                unsafe {
230                                    // SAFETY: This closure is explicitly executed through the OBS runtime;
231                                    // the signal string and captured hub Arc remain alive for disconnect.
232                                    libobs::signal_handler_disconnect(
233                                        handler,
234                                        signal.as_c_str().as_ptr(),
235                                        Some([<$signal_name:snake _handler>]),
236                                        std::sync::Arc::as_ptr(&[<$signal_name:snake _hub>])
237                                            as *mut std::ffi::c_void,
238                                    );
239                                }
240                            )*
241                        });
242                        if let Err(err) = result {
243                            log::error!(
244                                "Failed to disconnect {} signals on the OBS actor: {:?}",
245                                stringify!($name),
246                                err
247                            );
248                        }
249                    });
250                }
251            }
252        }
253    };
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259
260    #[test]
261    fn slow_signal_subscriber_drops_new_events_without_blocking() {
262        let hub = SignalHub::new();
263        let receiver = hub.subscribe();
264        for value in 0..(SIGNAL_QUEUE_CAPACITY + 5) {
265            hub.publish(value);
266        }
267
268        let drained: Vec<_> = (0..SIGNAL_QUEUE_CAPACITY)
269            .map(|_| receiver.try_recv().expect("queued signal"))
270            .collect();
271        assert_eq!(drained, (0..SIGNAL_QUEUE_CAPACITY).collect::<Vec<_>>());
272        assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));
273    }
274
275    #[test]
276    fn disconnected_signal_subscribers_are_removed() {
277        let hub = SignalHub::new();
278        let receiver = hub.subscribe();
279        drop(receiver);
280        hub.publish(1_u8);
281        let subscribers = hub
282            .subscribers
283            .lock()
284            .unwrap_or_else(|poisoned| poisoned.into_inner());
285        assert!(subscribers.is_empty());
286    }
287}
libobs_wrapper\signals/mod.rs

libobs_wrapper\signals/
mod.rs
