opcua_client/session/services/subscriptions/
callbacks.rs

1use opcua_types::{
2    match_extension_object_owned, DataChangeNotification, DataValue, EventNotificationList,
3    NotificationMessage, StatusChangeNotification, Variant,
4};
5
6use crate::{session::services::subscriptions::MonitoredItemMap, MonitoredItem};
7
8/// A trait for handling subscription notifications.
9/// Typically, you will want to use OnSubscriptionNotification instead,
10/// which has a blanket implementation for this trait.
11pub trait OnSubscriptionNotificationCore: Send + Sync {
12    /// Called when a notification is received on a subscription.
13    fn on_subscription_notification(
14        &mut self,
15        notification: NotificationMessage,
16        monitored_items: MonitoredItemMap<'_>,
17    );
18}
19
20impl<T> OnSubscriptionNotificationCore for T
21where
22    T: OnSubscriptionNotification + Send + Sync,
23{
24    fn on_subscription_notification(
25        &mut self,
26        notification: NotificationMessage,
27        monitored_items: MonitoredItemMap<'_>,
28    ) {
29        let Some(notifications) = notification.notification_data else {
30            return;
31        };
32
33        for obj in notifications {
34            match_extension_object_owned!(obj,
35                v: DataChangeNotification => {
36                    for notif in v.monitored_items.into_iter().flatten() {
37                        let item = monitored_items.get(notif.client_handle);
38
39                        if let Some(item) = item {
40                            self.on_data_value(notif.value, item);
41                        } else {
42                            tracing::warn!("Received notification for unknown monitored item {}", notif.client_handle);
43                        }
44                    }
45                },
46                v: EventNotificationList => {
47                    for notif in v.events.into_iter().flatten() {
48                        let item = monitored_items.get(notif.client_handle);
49
50                        if let Some(item) = item {
51                            self.on_event(notif.event_fields, item);
52                        } else {
53                            tracing::warn!("Received event for unknown monitored item {}", notif.client_handle);
54                        }
55                    }
56                },
57                v: StatusChangeNotification => {
58                    self.on_subscription_status_change(v);
59                }
60            )
61        }
62    }
63}
64
65/// A set of callbacks for notifications on a subscription.
66/// You may implement this on your own struct, or simply use [SubscriptionCallbacks]
67/// for a simple collection of closures.
68pub trait OnSubscriptionNotification: Send + Sync {
69    /// Called when a subscription changes state on the server.
70    #[allow(unused)]
71    fn on_subscription_status_change(&mut self, notification: StatusChangeNotification) {}
72
73    /// Called for each data value change.
74    #[allow(unused)]
75    fn on_data_value(&mut self, notification: DataValue, item: &MonitoredItem) {}
76
77    /// Called for each received event.
78    #[allow(unused)]
79    fn on_event(&mut self, event_fields: Option<Vec<Variant>>, item: &MonitoredItem) {}
80}
81
82type StatusChangeCallbackFun = dyn FnMut(StatusChangeNotification) + Send + Sync;
83type DataChangeCallbackFun = dyn FnMut(DataValue, &MonitoredItem) + Send + Sync;
84type EventCallbackFun = dyn FnMut(Option<Vec<Variant>>, &MonitoredItem) + Send + Sync;
85
86/// A convenient wrapper around a set of callback functions that implements [OnSubscriptionNotification]
87pub struct SubscriptionCallbacks {
88    status_change: Box<StatusChangeCallbackFun>,
89    data_value: Box<DataChangeCallbackFun>,
90    event: Box<EventCallbackFun>,
91}
92
93impl SubscriptionCallbacks {
94    /// Create a new subscription callback wrapper.
95    ///
96    /// # Arguments
97    ///
98    /// * `status_change` - Called when a subscription changes state on the server.
99    /// * `data_value` - Called for each received data value.
100    /// * `event` - Called for each received event.
101    pub fn new(
102        status_change: impl FnMut(StatusChangeNotification) + Send + Sync + 'static,
103        data_value: impl FnMut(DataValue, &MonitoredItem) + Send + Sync + 'static,
104        event: impl FnMut(Option<Vec<Variant>>, &MonitoredItem) + Send + Sync + 'static,
105    ) -> Self {
106        Self {
107            status_change: Box::new(status_change) as Box<StatusChangeCallbackFun>,
108            data_value: Box::new(data_value) as Box<DataChangeCallbackFun>,
109            event: Box::new(event) as Box<EventCallbackFun>,
110        }
111    }
112}
113
114impl OnSubscriptionNotification for SubscriptionCallbacks {
115    fn on_subscription_status_change(&mut self, notification: StatusChangeNotification) {
116        (self.status_change)(notification);
117    }
118
119    fn on_data_value(&mut self, notification: DataValue, item: &MonitoredItem) {
120        (self.data_value)(notification, item);
121    }
122
123    fn on_event(&mut self, event_fields: Option<Vec<Variant>>, item: &MonitoredItem) {
124        (self.event)(event_fields, item);
125    }
126}
127
128/// A wrapper around a data change callback that implements [OnSubscriptionNotification]
129pub struct DataChangeCallback {
130    data_value: Box<DataChangeCallbackFun>,
131}
132
133impl DataChangeCallback {
134    /// Create a new data change callback wrapper.
135    ///
136    /// # Arguments
137    ///
138    /// * `data_value` - Called for each received data value.
139    pub fn new(data_value: impl FnMut(DataValue, &MonitoredItem) + Send + Sync + 'static) -> Self {
140        Self {
141            data_value: Box::new(data_value)
142                as Box<dyn FnMut(DataValue, &MonitoredItem) + Send + Sync>,
143        }
144    }
145}
146
147impl OnSubscriptionNotification for DataChangeCallback {
148    fn on_data_value(&mut self, notification: DataValue, item: &MonitoredItem) {
149        (self.data_value)(notification, item);
150    }
151}
152
153/// A wrapper around an event callback that implements [OnSubscriptionNotification]
154pub struct EventCallback {
155    event: Box<EventCallbackFun>,
156}
157
158impl EventCallback {
159    /// Create a new event callback wrapper.
160    ///
161    /// # Arguments
162    ///
163    /// * `data_value` - Called for each received data value.
164    pub fn new(
165        event: impl FnMut(Option<Vec<Variant>>, &MonitoredItem) + Send + Sync + 'static,
166    ) -> Self {
167        Self {
168            event: Box::new(event)
169                as Box<dyn FnMut(Option<Vec<Variant>>, &MonitoredItem) + Send + Sync>,
170        }
171    }
172}
173
174impl OnSubscriptionNotification for EventCallback {
175    fn on_event(&mut self, event_fields: Option<Vec<Variant>>, item: &MonitoredItem) {
176        (self.event)(event_fields, item);
177    }
178}
opcua_client/session/services/subscriptions/callbacks.rs

opcua_client/session/services/subscriptions/
callbacks.rs
