ferroui_base/data/core/expression_nodes/
stream_node.rs

1use super::{ExpressionNode, NodeState};
2use crate::data::core::plugins::{BindingPlugins, IStreamPlugin};
3use crate::data::core::WeakValue;
4use crate::data::BindingError;
5use crate::reactive::{IDisposable, IObserver};
6use crate::BoxedValue;
7use std::cell::RefCell;
8use std::rc::{Rc, Weak};
9
10/// A node that produces the values of an observable source (`^`): through a
11/// fixed stream plugin (compiled binding paths) or, when none is given, the
12/// first registered plugin that matches the source.
13pub struct StreamNode {
14    this: Weak<StreamNode>,
15    state: NodeState,
16    plugin: Option<Rc<dyn IStreamPlugin>>,
17    subscription: RefCell<Option<Rc<dyn IDisposable>>>,
18}
19
20impl StreamNode {
21    pub fn new(plugin: Rc<dyn IStreamPlugin>) -> Rc<Self> {
22        Self::create(Some(plugin))
23    }
24
25    /// Creates a node that selects the plugin from the registered plugins.
26    pub fn new_dynamic() -> Rc<Self> {
27        Self::create(None)
28    }
29
30    fn create(plugin: Option<Rc<dyn IStreamPlugin>>) -> Rc<Self> {
31        Rc::new_cyclic(|this| Self {
32            this: this.clone(),
33            state: NodeState::new(),
34            plugin,
35            subscription: RefCell::new(None),
36        })
37    }
38}
39
40struct WeakObserver(Weak<StreamNode>);
41
42impl IObserver<Option<BoxedValue>> for WeakObserver {
43    fn on_next(&self, value: Option<BoxedValue>) {
44        if let Some(node) = self.0.upgrade() {
45            node.state.set_value_or_notification(value);
46        }
47    }
48}
49
50impl ExpressionNode for StreamNode {
51    fn state(&self) -> &NodeState {
52        &self.state
53    }
54
55    fn build_string(&self, builder: &mut String) {
56        builder.push('^');
57    }
58
59    fn on_source_changed(&self, source: Option<&BoxedValue>, _data_validation_error: Option<&BindingError>) {
60        if !self.state.validate_non_null_source(source) {
61            return;
62        }
63        let reference = WeakValue::new(source.expect("validated"));
64        let dynamic = self.plugin.is_none();
65        let plugin = match &self.plugin {
66            Some(plugin) => Some(plugin.clone()),
67            None => BindingPlugins::stream_handlers().into_iter().find(|p| p.match_(&reference)),
68        };
69        match plugin.and_then(|p| p.start(&reference)) {
70            Some(observable) => {
71                let subscription = observable.subscribe(Rc::new(WeakObserver(self.this.clone())));
72                self.subscription.replace(Some(subscription));
73            }
74            None if dynamic => self.state.set_value(None, None),
75            None => self.state.clear_value(),
76        }
77    }
78
79    fn unsubscribe(&self, _old_source: &BoxedValue) {
80        let subscription = self.subscription.borrow_mut().take();
81        if let Some(s) = subscription {
82            s.dispose();
83        }
84    }
85}
86
87impl Drop for StreamNode {
88    fn drop(&mut self) {
89        if let Some(s) = self.subscription.get_mut().take() {
90            s.dispose();
91        }
92    }
93}
ferroui_base/data/core/expression_nodes/stream_node.rs

ferroui_base/data/core/expression_nodes/
stream_node.rs
