ferroui_xaml_interactions/core/
observable_stream_behavior.rs

1//! Port of `Core/ObservableStreamBehavior.cs`.
2
3use super::{attach_actions_to_logical_tree, detach_actions_from_logical_tree};
4use ferroui_base::logical_tree::LogicalTreeAttachmentEventArgs;
5use ferroui_base::reactive::{AnonymousObserver, IDisposable, IObservable, ObservableError};
6use ferroui_base::threading::{Dispatcher, DispatcherPriority};
7use ferroui_base::{
8    ferro_class, ferro_class_info, ferro_properties, instantiate, BoxedValue, FerroObjectImpl, FerroObjectImplExt,
9    FerroProperty, FerroPropertyChangedEventArgs, Ref, StyledElementImpl, StyledElementImplExt, StyledProperty,
10    WeakRef,
11};
12use ferroui_xaml_interactivity::{ActionCollection, Interaction, StyledElementBehavior, StyledElementBehaviorImpl};
13use std::cell::RefCell;
14use std::rc::Rc;
15
16/// A behavior that subscribes to an observable and executes actions on its values, its error and
17/// its completion.
18///
19/// The [`source`](Self::source) is an observable of untyped values
20/// (`Rc<dyn IObservable<Option<BoxedValue>>>` or `Rc<dyn IObservable<BoxedValue>>`, the
21/// `IObservable<object>` of the reference). The subscription lives while the behavior is in the
22/// logical tree; the actions run in jobs of the dispatcher.
23#[repr(C)]
24pub struct ObservableStreamBehavior {
25    base: StyledElementBehavior,
26    subscription: RefCell<Option<Rc<dyn IDisposable>>>,
27}
28
29ferro_class!(ObservableStreamBehavior: StyledElementBehavior);
30ferro_class_info!(ObservableStreamBehavior {
31    new: ObservableStreamBehavior::new,
32    markup: {
33        content: Actions,
34    },
35});
36ferroui_base::ferro_impl_classes!(ObservableStreamBehavior: StyledElementBehaviorImpl);
37
38ferro_properties! {
39    impl ObservableStreamBehavior {
40        /// Identifies the `Source` property.
41        pub fn source_property() -> StyledProperty<Option<BoxedValue>> {
42            FerroProperty::register::<ObservableStreamBehavior, _>("Source", None)
43        }
44
45        /// Identifies the `Actions` property.
46        pub fn actions_property() -> StyledProperty<Option<ActionCollection>> {
47            FerroProperty::register::<ObservableStreamBehavior, _>("Actions", None)
48        }
49
50        /// Identifies the `ErrorActions` property.
51        pub fn error_actions_property() -> StyledProperty<Option<ActionCollection>> {
52            FerroProperty::register::<ObservableStreamBehavior, _>("ErrorActions", None)
53        }
54
55        /// Identifies the `CompletedActions` property.
56        pub fn completed_actions_property() -> StyledProperty<Option<ActionCollection>> {
57            FerroProperty::register::<ObservableStreamBehavior, _>("CompletedActions", None)
58        }
59    }
60}
61
62ferroui_base::ferro_overrides! {
63    impl FerroObjectImpl for ObservableStreamBehavior {
64        /// The constructor body: the behavior starts with empty collections of actions.
65        fn constructed(this: &Self) {
66            Self::parent_constructed(this);
67            this.set_current_value(Self::actions_property(), Some(ActionCollection::new()));
68            this.set_current_value(Self::error_actions_property(), Some(ActionCollection::new()));
69            this.set_current_value(Self::completed_actions_property(), Some(ActionCollection::new()));
70        }
71
72        fn on_property_changed(this: &Self, change: &FerroPropertyChangedEventArgs<'_>) {
73            Self::parent_on_property_changed(this, change);
74
75            let property = change.property();
76            if property == Self::source_property().as_property() {
77                this.update_subscription();
78            } else if property == Self::actions_property().as_property()
79                || property == Self::error_actions_property().as_property()
80                || property == Self::completed_actions_property().as_property()
81            {
82                this.handle_action_collection_changed(change);
83            }
84        }
85    }
86}
87
88ferroui_base::ferro_overrides! {
89    impl StyledElementImpl for ObservableStreamBehavior {
90        fn on_attached_to_logical_tree(this: &Self, e: &LogicalTreeAttachmentEventArgs) {
91            Self::parent_on_attached_to_logical_tree(this, e);
92            attach_actions_to_logical_tree(this.actions().as_ref(), this);
93            attach_actions_to_logical_tree(this.error_actions().as_ref(), this);
94            attach_actions_to_logical_tree(this.completed_actions().as_ref(), this);
95            this.update_subscription();
96        }
97
98        fn on_detached_from_logical_tree(this: &Self, e: &LogicalTreeAttachmentEventArgs) {
99            let subscription = this.subscription.borrow_mut().take();
100            if let Some(subscription) = subscription {
101                subscription.dispose();
102            }
103            detach_actions_from_logical_tree(this.actions().as_ref(), this);
104            detach_actions_from_logical_tree(this.error_actions().as_ref(), this);
105            detach_actions_from_logical_tree(this.completed_actions().as_ref(), this);
106            Self::parent_on_detached_from_logical_tree(this, e);
107        }
108    }
109}
110
111/// The observable an untyped source is (`Source is IObservable<object> observable`): its values
112/// are given to the actions as untyped values.
113fn observable_of(source: &BoxedValue) -> Option<Rc<dyn IObservable<Option<BoxedValue>>>> {
114    if let Some(observable) = source.downcast_ref::<Rc<dyn IObservable<Option<BoxedValue>>>>() {
115        return Some(observable.clone());
116    }
117
118    if let Some(observable) = source.downcast_ref::<Rc<dyn IObservable<BoxedValue>>>() {
119        let observable = observable.clone();
120        return Some(ferroui_base::reactive::Observable::create(move |observer| {
121            observable.subscribe(Rc::new(ForwardingObserver { target: observer }))
122        }));
123    }
124
125    None
126}
127
128/// Gives the values of an observable of untyped values to an observer of optional untyped values.
129struct ForwardingObserver {
130    target: Rc<dyn ferroui_base::reactive::IObserver<Option<BoxedValue>>>,
131}
132
133impl ferroui_base::reactive::IObserver<BoxedValue> for ForwardingObserver {
134    fn on_next(&self, value: BoxedValue) {
135        self.target.on_next(Some(value));
136    }
137
138    fn on_error(&self, error: ObservableError) {
139        self.target.on_error(error);
140    }
141
142    fn on_completed(&self) {
143        self.target.on_completed();
144    }
145}
146
147impl ObservableStreamBehavior {
148    /// Creates the class data; see [`ferroui_base::FerroObject::construct`].
149    pub fn construct() -> Self {
150        Self { base: StyledElementBehavior::construct(), subscription: RefCell::new(None) }
151    }
152
153    /// Creates a behavior.
154    pub fn new() -> Ref<Self> {
155        instantiate(Self::construct())
156    }
157
158    /// Gets or sets the source observable. This is a property of the object model.
159    pub fn source(&self) -> Option<BoxedValue> {
160        self.get_value(Self::source_property())
161    }
162
163    pub fn set_source(&self, value: Option<BoxedValue>) {
164        self.set_value(Self::source_property(), value)
165    }
166
167    /// Gets or sets the collection of actions to execute when the observable emits a value. This
168    /// is a property of the object model.
169    pub fn actions(&self) -> Option<ActionCollection> {
170        self.get_value(Self::actions_property())
171    }
172
173    pub fn set_actions(&self, value: Option<ActionCollection>) {
174        self.set_value(Self::actions_property(), value)
175    }
176
177    /// Gets or sets the collection of actions to execute when the observable emits an error. This
178    /// is a property of the object model.
179    pub fn error_actions(&self) -> Option<ActionCollection> {
180        self.get_value(Self::error_actions_property())
181    }
182
183    pub fn set_error_actions(&self, value: Option<ActionCollection>) {
184        self.set_value(Self::error_actions_property(), value)
185    }
186
187    /// Gets or sets the collection of actions to execute when the observable completes. This is a
188    /// property of the object model.
189    pub fn completed_actions(&self) -> Option<ActionCollection> {
190        self.get_value(Self::completed_actions_property())
191    }
192
193    pub fn set_completed_actions(&self, value: Option<ActionCollection>) {
194        self.set_value(Self::completed_actions_property(), value)
195    }
196
197    fn handle_action_collection_changed(&self, change: &FerroPropertyChangedEventArgs<'_>) {
198        let old_actions = change.get_old_value::<Option<ActionCollection>>().flatten();
199        let new_actions = change.get_new_value::<Option<ActionCollection>>();
200
201        if let Some(old_actions) = &old_actions {
202            if self.is_attached_to_logical_tree() {
203                detach_actions_from_logical_tree(Some(old_actions), self);
204            }
205        }
206
207        if let Some(new_actions) = &new_actions {
208            if self.is_attached_to_logical_tree() {
209                attach_actions_to_logical_tree(Some(new_actions), self);
210            }
211        }
212    }
213
214    fn update_subscription(&self) {
215        let subscription = self.subscription.borrow_mut().take();
216        if let Some(subscription) = subscription {
217            subscription.dispose();
218        }
219
220        let Some(observable) = self.source().as_ref().and_then(observable_of) else {
221            return;
222        };
223
224        let on_next = {
225            let behavior: WeakRef<ObservableStreamBehavior> = self.to_ref().downgrade();
226            move |value: Option<BoxedValue>| {
227                let behavior = behavior.clone();
228                Dispatcher::ui_thread().post_local(
229                    move || {
230                        if let Some(behavior) = behavior.upgrade() {
231                            let sender = behavior.associated_sender();
232                            Interaction::execute_actions(sender.as_ref(), behavior.actions().as_ref(), value.as_ref());
233                        }
234                    },
235                    DispatcherPriority::DEFAULT,
236                );
237            }
238        };
239
240        let on_error = {
241            let behavior: WeakRef<ObservableStreamBehavior> = self.to_ref().downgrade();
242            move |error: ObservableError| {
243                let behavior = behavior.clone();
244                // Deviation (DEVIATIONS.md, Behaviors): the error actions are given the message of
245                // the error (an error of the observable is not an untyped value).
246                let error: BoxedValue = Rc::new(error.to_string());
247                Dispatcher::ui_thread().post_local(
248                    move || {
249                        if let Some(behavior) = behavior.upgrade() {
250                            let sender = behavior.associated_sender();
251                            Interaction::execute_actions(sender.as_ref(), behavior.error_actions().as_ref(), Some(&error));
252                        }
253                    },
254                    DispatcherPriority::DEFAULT,
255                );
256            }
257        };
258
259        let on_completed = {
260            let behavior: WeakRef<ObservableStreamBehavior> = self.to_ref().downgrade();
261            move || {
262                let behavior = behavior.clone();
263                Dispatcher::ui_thread().post_local(
264                    move || {
265                        if let Some(behavior) = behavior.upgrade() {
266                            let sender = behavior.associated_sender();
267                            Interaction::execute_actions(sender.as_ref(), behavior.completed_actions().as_ref(), None);
268                        }
269                    },
270                    DispatcherPriority::DEFAULT,
271                );
272            }
273        };
274
275        let subscription =
276            observable.subscribe(Rc::new(AnonymousObserver::new_with_error_and_completed(on_next, on_error, on_completed)));
277        *self.subscription.borrow_mut() = Some(subscription);
278    }
279
280    /// The associated object as the sender of the actions.
281    fn associated_sender(&self) -> Option<BoxedValue> {
282        self.associated_object().map(|associated_object| Rc::new(associated_object) as BoxedValue)
283    }
284}
ferroui_xaml_interactions/core/observable_stream_behavior.rs

ferroui_xaml_interactions/core/
observable_stream_behavior.rs
