use pyo3::prelude::*;
use pyo3::types::PyList;
use std::sync::{Arc, Mutex};
mod file_watcher;
use file_watcher::*;
mod folder_watcher;
use folder_watcher::*;
pub mod clipboard;
pub mod keyboard_mouse;
#[pyclass]
pub struct Subscription {
disposed: Arc<Mutex<bool>>,
}
#[pymethods]
impl Subscription {
#[new]
fn new() -> Self {
Self {
disposed: Arc::new(Mutex::new(false)),
}
}
fn dispose(&self) {
*self.disposed.lock().unwrap() = true;
}
fn is_disposed(&self) -> bool {
*self.disposed.lock().unwrap()
}
}
type SubscribeFn = Arc<dyn Fn(PyObject) -> Py<Subscription> + Send + Sync>;
#[pyclass]
pub struct Observable {
subscribe_fn: SubscribeFn,
}
impl Observable {
fn new_impl<F>(f: F) -> Self
where
F: Fn(PyObject) -> Py<Subscription> + Send + Sync + 'static,
{
Self {
subscribe_fn: Arc::new(f),
}
}
}
#[pymethods]
impl Observable {
#[staticmethod]
fn from_subscribe_fn(py: Python<'_>, fn_obj: PyObject) -> PyResult<Self> {
let fn_handle = fn_obj.clone_ref(py);
Ok(Self::new_impl(move |observer: PyObject| {
Python::with_gil(|py| {
match fn_handle.call1(py, (observer,)) {
Ok(ret) => {
if let Ok(sub) = ret.extract::<Py<Subscription>>(py) {
return sub;
}
Py::new(py, Subscription::new()).unwrap()
}
Err(_) => Py::new(py, Subscription::new()).unwrap(),
}
})
}))
}
#[staticmethod]
fn of(value: PyObject) -> Self {
Self::new_impl(move |observer| {
Python::with_gil(|py| {
let _ = observer.call1(py, (value.clone_ref(py),));
Py::new(py, Subscription::new()).unwrap()
})
})
}
#[staticmethod]
fn empty() -> Self {
Self::new_impl(move |_| {
Python::with_gil(|py| Py::new(py, Subscription::new()).unwrap())
})
}
#[staticmethod]
fn never() -> Self {
Self::new_impl(move |_| {
Python::with_gil(|py| Py::new(py, Subscription::new()).unwrap())
})
}
#[staticmethod]
fn range(start: i64, count: i64) -> Self {
Self::new_impl(move |observer| {
Python::with_gil(|py| {
for i in 0..count {
let value = (start + i).to_object(py);
let _ = observer.call1(py, (value,));
}
Py::new(py, Subscription::new()).unwrap()
})
})
}
#[staticmethod]
fn repeat(value: PyObject, count: usize) -> Self {
Self::new_impl(move |observer| {
Python::with_gil(|py| {
for _ in 0..count {
let _ = observer.call1(py, (value.clone_ref(py),));
}
Py::new(py, Subscription::new()).unwrap()
})
})
}
#[staticmethod]
fn from_iter(values: Bound<'_, PyList>) -> Self {
let items: Vec<PyObject> = Python::with_gil(|py| {
values.iter().map(|item| item.unbind()).collect()
});
Self::new_impl(move |observer| {
Python::with_gil(|py| {
for item in &items {
let _ = observer.call1(py, (item.clone_ref(py),));
}
Py::new(py, Subscription::new()).unwrap()
})
})
}
#[pyo3(signature = (on_next, on_error=None, on_completed=None))]
fn subscribe(&self, on_next: PyObject, on_error: Option<PyObject>, on_completed: Option<PyObject>) -> Py<Subscription> {
(self.subscribe_fn)(on_next)
}
fn map(&self, mapper: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
Python::with_gil(|py| {
let mapper_clone = mapper.clone_ref(py);
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let result = mapper_clone.call1(py, (value,)).unwrap();
let _ = downstream_clone.call1(py, (result,));
Ok(())
});
source_fn(wrapped.to_object(py));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn filter(&self, predicate: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
Python::with_gil(|py| {
let predicate_clone = predicate.clone_ref(py);
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let should_pass: bool = predicate_clone
.call1(py, (value.clone_ref(py),))
.unwrap()
.extract(py)
.unwrap_or(false);
if should_pass {
let _ = downstream_clone.call1(py, (value,));
}
Ok(())
});
source_fn(wrapped.to_object(py));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn flat_map(&self, mapper: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
Python::with_gil(|py| {
let mapper_clone = mapper.clone_ref(py);
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let inner = mapper_clone.call1(py, (value,)).unwrap();
if let Ok(iter_obj) = inner.call_method0(py, "__iter__") {
let iter_bound = iter_obj.bind(py);
while let Ok(next_item) = iter_bound.call_method0("__next__") {
let _ = downstream_clone.call1(py, (next_item.unbind(),));
}
}
Ok(())
});
source_fn(wrapped.to_object(py));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn scan(&self, initial: PyObject, scanner: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
let accumulator: Arc<Mutex<Option<PyObject>>> = Arc::new(Mutex::new(None));
Python::with_gil(|py| {
*accumulator.lock().unwrap() = Some(initial.clone_ref(py));
let accumulator_clone = accumulator.clone();
let scanner_clone = scanner.clone_ref(py);
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let mut acc_guard = accumulator_clone.lock().unwrap();
if let Some(ref acc) = *acc_guard {
let new_acc = scanner_clone
.call1(py, (acc.clone_ref(py), value.clone_ref(py)))
.unwrap();
*acc_guard = Some(new_acc);
}
Ok(())
});
source_fn(wrapped.to_object(py));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn skip(&self, n: usize) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
let counter = Arc::new(Mutex::new(0usize));
Python::with_gil(|py| {
let counter_clone = counter.clone();
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let mut cnt = counter_clone.lock().unwrap();
if *cnt < n {
*cnt += 1;
return Ok(());
}
drop(cnt);
let _ = downstream_clone.call1(py, (value,));
Ok(())
});
source_fn(wrapped.to_object(py));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn take(&self, n: usize) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
let counter = Arc::new(Mutex::new(0usize));
Python::with_gil(|py| {
let counter_clone = counter.clone();
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let mut cnt = counter_clone.lock().unwrap();
if *cnt < n {
*cnt += 1;
drop(cnt);
let _ = downstream_clone.call1(py, (value,));
}
Ok(())
});
source_fn(wrapped.to_object(py));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn first(&self) -> Self {
self.take(1)
}
fn last(&self) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
Python::with_gil(|py| {
let last_value = Arc::new(Mutex::new(py.None()));
let have_value = Arc::new(Mutex::new(false));
let last_value_clone = last_value.clone();
let have_value_clone = have_value.clone();
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
*last_value_clone.lock().unwrap() = value.clone_ref(py);
*have_value_clone.lock().unwrap() = true;
Ok(())
});
source_fn(wrapped.to_object(py));
if *have_value.lock().unwrap() {
let last = last_value.lock().unwrap().clone_ref(py);
let _ = downstream_observer.call1(py, (last,));
}
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn count(&self) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
let counter = Arc::new(Mutex::new(0i64));
Python::with_gil(|py| {
let counter_clone = counter.clone();
let wrapped = RustObserver::new(move |_py, _value| {
*counter_clone.lock().unwrap() += 1;
Ok(())
});
source_fn(wrapped.to_object(py));
let final_count = *counter.lock().unwrap();
let _ = downstream_observer.call1(py, (final_count,));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn sum(&self) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
Python::with_gil(|py| {
let total = Arc::new(Mutex::new(py.None()));
let have_value = Arc::new(Mutex::new(false));
let total_clone = total.clone();
let have_value_clone = have_value.clone();
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let mut current = total_clone.lock().unwrap();
if !*have_value_clone.lock().unwrap() {
*current = value.clone_ref(py);
*have_value_clone.lock().unwrap() = true;
} else {
let current_value = current.clone_ref(py);
let new_value = current_value.call_method1(py, "__add__", (value,)).unwrap();
*current = new_value;
}
Ok(())
});
source_fn(wrapped.to_object(py));
if *have_value.lock().unwrap() {
let total_value = total.lock().unwrap().clone_ref(py);
let _ = downstream_clone.call1(py, (total_value,));
}
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn reduce(&self, initial: PyObject, reducer: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
let accumulator: Arc<Mutex<Option<PyObject>>> = Arc::new(Mutex::new(None));
Python::with_gil(|py| {
*accumulator.lock().unwrap() = Some(initial.clone_ref(py));
let accumulator_clone = accumulator.clone();
let reducer_clone = reducer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let mut acc_guard = accumulator_clone.lock().unwrap();
if let Some(ref acc) = *acc_guard {
let new_acc = reducer_clone
.call1(py, (acc.clone_ref(py), value.clone_ref(py)))
.unwrap();
*acc_guard = Some(new_acc);
}
Ok(())
});
source_fn(wrapped.to_object(py));
if let Some(final_acc) = accumulator.lock().unwrap().take() {
let _ = downstream_observer.call1(py, (final_acc,));
}
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn contains(&self, target: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
let found = Arc::new(Mutex::new(false));
Python::with_gil(|py| {
let found_clone = found.clone();
let target_clone = target.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
if !*found_clone.lock().unwrap() {
let eq = target_clone
.call_method1(py, "__eq__", (value.clone_ref(py),))
.unwrap();
let is_eq: bool = eq.extract(py).unwrap_or(false);
if is_eq {
*found_clone.lock().unwrap() = true;
}
}
Ok(())
});
source_fn(wrapped.to_object(py));
let result = if *found.lock().unwrap() { true.to_object(py) } else { false.to_object(py) };
let _ = downstream_observer.call1(py, (result,));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn all(&self, predicate: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
let all_pass = Arc::new(Mutex::new(true));
Python::with_gil(|py| {
let all_pass_clone = all_pass.clone();
let predicate_clone = predicate.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
if *all_pass_clone.lock().unwrap() {
let result = predicate_clone
.call1(py, (value.clone_ref(py),))
.unwrap();
let pass: bool = result.extract(py).unwrap_or(false);
if !pass {
*all_pass_clone.lock().unwrap() = false;
}
}
Ok(())
});
source_fn(wrapped.to_object(py));
let result = if *all_pass.lock().unwrap() { true.to_object(py) } else { false.to_object(py) };
let _ = downstream_observer.call1(py, (result,));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn collect(&self) -> PyObject {
let source_fn = self.subscribe_fn.clone();
Python::with_gil(|py| {
let items: Arc<Mutex<Vec<PyObject>>> = Arc::new(Mutex::new(Vec::new()));
let items_clone = items.clone();
let wrapped = RustObserver::new(move |py, value| {
items_clone.lock().unwrap().push(value.clone_ref(py));
Ok(())
});
source_fn(wrapped.to_object(py));
let collected = items.lock().unwrap();
let list = PyList::empty_bound(py);
for item in collected.iter() {
let _ = list.append(item.clone_ref(py));
}
list.unbind().into_any()
})
}
fn merge(&self, other: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
Python::with_gil(|py| {
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped_self = RustObserver::new(move |py, value| {
let _ = downstream_clone.call1(py, (value,));
Ok(())
});
source_fn(wrapped_self.to_object(py));
if let Ok(other_inner) = other.getattr(py, "_inner") {
if let Ok(other_sub_fn) = other_inner.getattr(py, "subscribe_fn") {
let _ = other_sub_fn.call1(py, (RustObserver::new(move |py, value| {
let _ = downstream_observer.call1(py, (value,));
Ok(())
}).to_object(py),));
}
}
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn concat(&self, other: PyObject) -> Self {
self.merge(other)
}
fn start_with(&self, value: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
Python::with_gil(|py| {
let _ = downstream_observer.call1(py, (value.clone_ref(py),));
source_fn(downstream_observer);
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn default_if_empty(&self, default: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
let emitted = Arc::new(Mutex::new(false));
Python::with_gil(|py| {
let emitted_clone = emitted.clone();
let downstream_clone = downstream_observer.clone_ref(py);
let default_clone = default.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
*emitted_clone.lock().unwrap() = true;
let _ = downstream_clone.call1(py, (value,));
Ok(())
});
source_fn(wrapped.to_object(py));
if !*emitted.lock().unwrap() {
let _ = downstream_observer.call1(py, (default_clone,));
}
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn do_on_next(&self, action: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
Python::with_gil(|py| {
let action_clone = action.clone_ref(py);
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let _ = action_clone.call1(py, (value.clone_ref(py),));
let _ = downstream_clone.call1(py, (value,));
Ok(())
});
source_fn(wrapped.to_object(py));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn run(&self) -> Self {
Python::with_gil(|py| {
let noop = py.None();
let source_fn = self.subscribe_fn.clone();
source_fn(noop);
});
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |observer| source_fn(observer))
}
#[pyo3(signature = (name=None))]
fn debug(&self, name: Option<String>) -> Self {
let prefix = name.unwrap_or_else(|| String::from("[Observable]"));
println!("{} subscribed", prefix);
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
let prefix = prefix.clone();
Python::with_gil(|py| {
let prefix_clone = prefix.clone();
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let repr = value.call_method0(py, "__repr__").unwrap();
let repr_str: String = repr.extract(py).unwrap_or_default();
println!("{} on_next: {}", prefix_clone, repr_str);
let _ = downstream_clone.call1(py, (value,));
Ok(())
});
source_fn(wrapped.to_object(py));
println!("{} on_completed", prefix);
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn filter_by_event_type(&self, event_type: PyObject) -> Self {
let source_fn = self.subscribe_fn.clone();
Self::new_impl(move |downstream_observer| {
Python::with_gil(|py| {
let event_type_clone = event_type.clone_ref(py);
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
if let Ok(event_type_val) = value.call_method0(py, "event_type") {
let event_type_bound = event_type_clone.bind(py);
let event_type_val_bound = event_type_val.bind(py);
let eq: bool = event_type_val_bound.eq(event_type_bound).unwrap_or(false);
if eq {
let _ = downstream_clone.call1(py, (value,));
}
}
Ok(())
});
source_fn(wrapped.to_object(py));
Py::new(py, Subscription::new()).unwrap()
})
})
}
#[pyo3(signature = (duration, get_key=None))]
fn throttle_events(&self, duration: f64, get_key: Option<PyObject>) -> Self {
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
let source_fn = self.subscribe_fn.clone();
let last_emit: Arc<Mutex<HashMap<usize, f64>>> = Arc::new(Mutex::new(HashMap::new()));
Self::new_impl(move |downstream_observer| {
let last_emit_clone = last_emit.clone();
Python::with_gil(|py| {
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, value| {
let key = value.as_ptr() as usize;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs_f64();
let mut last_emit_map = last_emit_clone.lock().unwrap();
let last = last_emit_map.get(&key).copied().unwrap_or(0.0);
if now - last >= duration {
last_emit_map.insert(key, now);
let _ = downstream_clone.call1(py, (value,));
}
Ok(())
});
source_fn(wrapped.to_object(py));
Py::new(py, Subscription::new()).unwrap()
})
})
}
fn event_rate(&self, window_size: usize) -> Self {
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
let source_fn = self.subscribe_fn.clone();
let timestamps: Arc<Mutex<VecDeque<f64>>> = Arc::new(Mutex::new(VecDeque::with_capacity(window_size)));
Self::new_impl(move |downstream_observer| {
let timestamps_clone = timestamps.clone();
Python::with_gil(|py| {
let downstream_clone = downstream_observer.clone_ref(py);
let wrapped = RustObserver::new(move |py, _| {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs_f64();
let mut ts = timestamps_clone.lock().unwrap();
ts.push_back(now);
if ts.len() > window_size {
ts.pop_front();
}
if ts.len() >= 2 {
let elapsed = ts.back().unwrap() - ts.front().unwrap();
if elapsed > 0.0 {
let rate = (ts.len() - 1) as f64 / elapsed;
let _ = downstream_clone.call1(py, (rate,));
}
}
Ok(())
});
source_fn(wrapped.to_object(py));
Py::new(py, Subscription::new()).unwrap()
})
})
}
}
type ObserverClosure = Arc<dyn Fn(Python<'_>, PyObject) -> PyResult<()> + Send + Sync>;
#[pyclass]
pub struct RustObserver {
inner: ObserverClosure,
}
impl RustObserver {
fn new<F>(f: F) -> Self
where
F: Fn(Python<'_>, PyObject) -> PyResult<()> + Send + Sync + 'static,
{
Self {
inner: Arc::new(f),
}
}
fn to_object(&self, py: Python<'_>) -> PyObject {
let cloned = RustObserver {
inner: self.inner.clone(),
};
Py::new(py, cloned).unwrap().into_any()
}
}
#[pymethods]
impl RustObserver {
fn __call__(&self, value: PyObject) -> PyResult<()> {
Python::with_gil(|py| (self.inner)(py, value))
}
}
#[pyclass]
pub struct PublishSubject {
observers: Arc<Mutex<Vec<(usize, Arc<Mutex<bool>>, PyObject)>>>,
next_id: Arc<Mutex<usize>>,
}
#[pymethods]
impl PublishSubject {
#[new]
fn new() -> Self {
Self {
observers: Arc::new(Mutex::new(Vec::new())),
next_id: Arc::new(Mutex::new(0)),
}
}
fn on_next(&self, value: PyObject) {
Python::with_gil(|py| {
let snapshot: Vec<(usize, Arc<Mutex<bool>>, PyObject)> = self
.observers
.lock()
.unwrap()
.iter()
.map(|(id, flag, cb)| (*id, Arc::clone(flag), cb.clone_ref(py)))
.collect();
for (_, flag, cb) in snapshot.iter() {
if !*flag.lock().unwrap() {
let _ = cb.call1(py, (value.clone_ref(py),));
}
}
self.observers.lock().unwrap().retain(|obs| !*obs.1.lock().unwrap());
})
}
fn on_completed(&self) {
self.observers.lock().unwrap().clear();
}
#[pyo3(signature = (on_next, on_error=None, on_completed=None))]
fn subscribe(&self, on_next: PyObject, on_error: Option<PyObject>, on_completed: Option<PyObject>) -> Py<Subscription> {
Python::with_gil(|py| {
let mut id = self.next_id.lock().unwrap();
let my_id = *id;
*id += 1;
drop(id);
let disposed = Arc::new(Mutex::new(false));
let disposed_clone = disposed.clone();
let wrapped = RustObserver::new(move |py, value| {
if !*disposed_clone.lock().unwrap() {
let _ = on_next.call1(py, (value,));
}
Ok(())
});
self.observers.lock().unwrap().push((my_id, disposed.clone(), wrapped.to_object(py)));
Py::new(py, Subscription { disposed }).unwrap()
})
}
}
#[pyclass]
pub struct BehaviorSubject {
observers: Arc<Mutex<Vec<(usize, Arc<Mutex<bool>>, PyObject)>>>,
next_id: Arc<Mutex<usize>>,
current_value: Arc<Mutex<PyObject>>,
}
#[pymethods]
impl BehaviorSubject {
#[new]
fn new(initial_value: PyObject) -> Self {
Self {
observers: Arc::new(Mutex::new(Vec::new())),
next_id: Arc::new(Mutex::new(0)),
current_value: Arc::new(Mutex::new(initial_value)),
}
}
fn on_next(&self, value: PyObject) {
Python::with_gil(|py| {
*self.current_value.lock().unwrap() = value.clone_ref(py);
let snapshot: Vec<(usize, Arc<Mutex<bool>>, PyObject)> = self
.observers
.lock()
.unwrap()
.iter()
.map(|(id, flag, cb)| (*id, Arc::clone(flag), cb.clone_ref(py)))
.collect();
for (_, flag, cb) in snapshot.iter() {
if !*flag.lock().unwrap() {
let _ = cb.call1(py, (value.clone_ref(py),));
}
}
self.observers.lock().unwrap().retain(|obs| !*obs.1.lock().unwrap());
})
}
fn on_completed(&self) {
self.observers.lock().unwrap().clear();
}
#[pyo3(signature = (on_next, on_error=None, on_completed=None))]
fn subscribe(&self, on_next: PyObject, on_error: Option<PyObject>, on_completed: Option<PyObject>) -> Py<Subscription> {
Python::with_gil(|py| {
let mut id = self.next_id.lock().unwrap();
let my_id = *id;
*id += 1;
drop(id);
let disposed = Arc::new(Mutex::new(false));
let disposed_clone = disposed.clone();
let current = self.current_value.lock().unwrap().clone_ref(py);
let _ = on_next.call1(py, (current,));
let wrapped = RustObserver::new(move |py, value| {
if !*disposed_clone.lock().unwrap() {
let _ = on_next.call1(py, (value,));
}
Ok(())
});
self.observers.lock().unwrap().push((my_id, disposed.clone(), wrapped.to_object(py)));
Py::new(py, Subscription { disposed }).unwrap()
})
}
fn value(&self) -> PyObject {
Python::with_gil(|py| self.current_value.lock().unwrap().clone_ref(py))
}
}
#[pyclass]
pub struct ReplaySubject {
observers: Arc<Mutex<Vec<(usize, Arc<Mutex<bool>>, PyObject)>>>,
next_id: Arc<Mutex<usize>>,
buffer: Arc<Mutex<Vec<PyObject>>>,
buffer_size: usize,
}
#[pymethods]
impl ReplaySubject {
#[new]
fn new(buffer_size: usize) -> Self {
Self {
observers: Arc::new(Mutex::new(Vec::new())),
next_id: Arc::new(Mutex::new(0)),
buffer: Arc::new(Mutex::new(Vec::new())),
buffer_size,
}
}
fn on_next(&self, value: PyObject) {
Python::with_gil(|py| {
{
let mut buf = self.buffer.lock().unwrap();
buf.push(value.clone_ref(py));
if buf.len() > self.buffer_size {
buf.remove(0);
}
}
let snapshot: Vec<(usize, Arc<Mutex<bool>>, PyObject)> = self
.observers
.lock()
.unwrap()
.iter()
.map(|(id, flag, cb)| (*id, Arc::clone(flag), cb.clone_ref(py)))
.collect();
for (_, flag, cb) in snapshot.iter() {
if !*flag.lock().unwrap() {
let _ = cb.call1(py, (value.clone_ref(py),));
}
}
self.observers.lock().unwrap().retain(|obs| !*obs.1.lock().unwrap());
})
}
fn on_completed(&self) {
self.observers.lock().unwrap().clear();
}
#[pyo3(signature = (on_next, on_error=None, on_completed=None))]
fn subscribe(&self, on_next: PyObject, on_error: Option<PyObject>, on_completed: Option<PyObject>) -> Py<Subscription> {
Python::with_gil(|py| {
let mut id = self.next_id.lock().unwrap();
let my_id = *id;
*id += 1;
drop(id);
let disposed = Arc::new(Mutex::new(false));
let disposed_clone = disposed.clone();
let buffered: Vec<PyObject> = self
.buffer
.lock()
.unwrap()
.iter()
.map(|v| v.clone_ref(py))
.collect();
for item in buffered {
let _ = on_next.call1(py, (item,));
}
let wrapped = RustObserver::new(move |py, value| {
if !*disposed_clone.lock().unwrap() {
let _ = on_next.call1(py, (value,));
}
Ok(())
});
self.observers.lock().unwrap().push((my_id, disposed.clone(), wrapped.to_object(py)));
Py::new(py, Subscription { disposed }).unwrap()
})
}
}
#[pyclass]
pub struct CurrentThreadScheduler;
#[pymethods]
impl CurrentThreadScheduler {
#[new]
fn new() -> Self {
CurrentThreadScheduler
}
fn now(&self) -> f64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs_f64()
* 1000.0
}
}
#[pyclass]
pub struct ThreadPoolScheduler {
num_threads: usize,
}
#[pymethods]
impl ThreadPoolScheduler {
#[new]
#[pyo3(signature = (num_threads = 2))]
fn new(num_threads: usize) -> Self {
ThreadPoolScheduler { num_threads }
}
fn now(&self) -> f64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs_f64()
* 1000.0
}
fn get_num_threads(&self) -> usize {
self.num_threads
}
}
#[pyclass]
pub struct AsyncScheduler;
#[pymethods]
impl AsyncScheduler {
#[new]
fn new() -> Self {
AsyncScheduler
}
fn now(&self) -> f64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs_f64()
* 1000.0
}
}
#[pyclass]
pub struct ImmediateScheduler;
#[pymethods]
impl ImmediateScheduler {
#[new]
fn new() -> Self {
ImmediateScheduler
}
fn now(&self) -> f64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs_f64()
* 1000.0
}
}
#[pymodule]
fn rx_rust(m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<Subscription>()?;
m.add_class::<Observable>()?;
m.add_class::<PublishSubject>()?;
m.add_class::<BehaviorSubject>()?;
m.add_class::<ReplaySubject>()?;
m.add_class::<CurrentThreadScheduler>()?;
m.add_class::<ThreadPoolScheduler>()?;
m.add_class::<AsyncScheduler>()?;
m.add_class::<ImmediateScheduler>()?;
add_file_watcher_to_module(m)?;
crate::clipboard::toplevel::register_clipboard_module(m)?;
crate::keyboard_mouse::toplevel::register_keyboard_mouse_module(m)?;
add_folder_watcher_to_module(m)?;
Ok(())
}