opcua_client/session/services/
attributes.rs

1use std::time::Duration;
2
3use crate::{
4    session::{
5        process_service_result, process_unexpected_response,
6        request_builder::{
7            builder_base, builder_debug, builder_error, builder_trace, RequestHeaderBuilder,
8        },
9        UARequest,
10    },
11    AsyncSecureChannel, Session,
12};
13use opcua_core::ResponseMessage;
14use opcua_types::{
15    DataValue, DeleteAtTimeDetails, DeleteEventDetails, DeleteRawModifiedDetails, Error,
16    ExtensionObject, HistoryReadRequest, HistoryReadResponse, HistoryReadResult,
17    HistoryReadValueId, HistoryUpdateRequest, HistoryUpdateResponse, HistoryUpdateResult,
18    IntegerId, NodeId, ReadAtTimeDetails, ReadEventDetails, ReadProcessedDetails,
19    ReadRawModifiedDetails, ReadRequest, ReadResponse, ReadValueId, StatusCode, TimestampsToReturn,
20    UpdateDataDetails, UpdateEventDetails, UpdateStructureDataDetails, WriteRequest, WriteResponse,
21    WriteValue,
22};
23use tracing::{debug_span, Instrument};
24
25/// Enumeration used with Session::history_read()
26#[derive(Debug, Clone)]
27pub enum HistoryReadAction {
28    /// Read historical events.
29    ReadEventDetails(ReadEventDetails),
30    /// Read raw data values.
31    ReadRawModifiedDetails(ReadRawModifiedDetails),
32    /// Read data values with processing.
33    ReadProcessedDetails(ReadProcessedDetails),
34    /// Read data values at specific timestamps.
35    ReadAtTimeDetails(ReadAtTimeDetails),
36}
37
38impl From<HistoryReadAction> for ExtensionObject {
39    fn from(action: HistoryReadAction) -> Self {
40        match action {
41            HistoryReadAction::ReadEventDetails(v) => Self::from_message(v),
42            HistoryReadAction::ReadRawModifiedDetails(v) => Self::from_message(v),
43            HistoryReadAction::ReadProcessedDetails(v) => Self::from_message(v),
44            HistoryReadAction::ReadAtTimeDetails(v) => Self::from_message(v),
45        }
46    }
47}
48
49/// Enumeration used with Session::history_update()
50#[derive(Debug, Clone)]
51pub enum HistoryUpdateAction {
52    /// Update historical data values.
53    UpdateDataDetails(UpdateDataDetails),
54    /// Update historical structures.
55    UpdateStructureDataDetails(UpdateStructureDataDetails),
56    /// Update historical events.
57    UpdateEventDetails(UpdateEventDetails),
58    /// Delete raw data values.
59    DeleteRawModifiedDetails(DeleteRawModifiedDetails),
60    /// Delete data values at specific timestamps.
61    DeleteAtTimeDetails(DeleteAtTimeDetails),
62    /// Delete historical events.
63    DeleteEventDetails(DeleteEventDetails),
64}
65
66impl From<UpdateDataDetails> for HistoryUpdateAction {
67    fn from(value: UpdateDataDetails) -> Self {
68        Self::UpdateDataDetails(value)
69    }
70}
71impl From<UpdateStructureDataDetails> for HistoryUpdateAction {
72    fn from(value: UpdateStructureDataDetails) -> Self {
73        Self::UpdateStructureDataDetails(value)
74    }
75}
76impl From<UpdateEventDetails> for HistoryUpdateAction {
77    fn from(value: UpdateEventDetails) -> Self {
78        Self::UpdateEventDetails(value)
79    }
80}
81impl From<DeleteRawModifiedDetails> for HistoryUpdateAction {
82    fn from(value: DeleteRawModifiedDetails) -> Self {
83        Self::DeleteRawModifiedDetails(value)
84    }
85}
86impl From<DeleteAtTimeDetails> for HistoryUpdateAction {
87    fn from(value: DeleteAtTimeDetails) -> Self {
88        Self::DeleteAtTimeDetails(value)
89    }
90}
91impl From<DeleteEventDetails> for HistoryUpdateAction {
92    fn from(value: DeleteEventDetails) -> Self {
93        Self::DeleteEventDetails(value)
94    }
95}
96
97impl From<HistoryUpdateAction> for ExtensionObject {
98    fn from(action: HistoryUpdateAction) -> Self {
99        match action {
100            HistoryUpdateAction::UpdateDataDetails(v) => Self::from_message(v),
101            HistoryUpdateAction::UpdateStructureDataDetails(v) => Self::from_message(v),
102            HistoryUpdateAction::UpdateEventDetails(v) => Self::from_message(v),
103            HistoryUpdateAction::DeleteRawModifiedDetails(v) => Self::from_message(v),
104            HistoryUpdateAction::DeleteAtTimeDetails(v) => Self::from_message(v),
105            HistoryUpdateAction::DeleteEventDetails(v) => Self::from_message(v),
106        }
107    }
108}
109
110/// Builder for a call to the `Read` service.
111///
112/// See OPC UA Part 4 - Services 5.10.2 for complete description of the service and error responses.
113#[derive(Debug, Clone)]
114pub struct Read {
115    nodes_to_read: Vec<ReadValueId>,
116    timestamps_to_return: TimestampsToReturn,
117    max_age: f64,
118
119    header: RequestHeaderBuilder,
120}
121
122impl Read {
123    /// Construct a new call to the `Read` service.
124    pub fn new(session: &Session) -> Self {
125        Self {
126            nodes_to_read: Vec::new(),
127            timestamps_to_return: TimestampsToReturn::Neither,
128            max_age: 0.0,
129            header: RequestHeaderBuilder::new_from_session(session),
130        }
131    }
132
133    /// Construct a new call to the `Read` service, setting header parameters manually.
134    pub fn new_manual(
135        session_id: u32,
136        timeout: Duration,
137        auth_token: NodeId,
138        request_handle: IntegerId,
139    ) -> Self {
140        Self {
141            nodes_to_read: Vec::new(),
142            timestamps_to_return: TimestampsToReturn::Neither,
143            max_age: 0.0,
144            header: RequestHeaderBuilder::new(session_id, timeout, auth_token, request_handle),
145        }
146    }
147
148    /// Set timestamps to return.
149    pub fn timestamps_to_return(mut self, timestamps: TimestampsToReturn) -> Self {
150        self.timestamps_to_return = timestamps;
151        self
152    }
153
154    /// Set max age.
155    pub fn max_age(mut self, max_age: f64) -> Self {
156        self.max_age = max_age;
157        self
158    }
159
160    /// Set nodes to read, overwriting any that were set previously.
161    pub fn nodes_to_read(mut self, nodes_to_read: Vec<ReadValueId>) -> Self {
162        self.nodes_to_read = nodes_to_read;
163        self
164    }
165
166    /// Add a node to read.
167    pub fn node(mut self, node: ReadValueId) -> Self {
168        self.nodes_to_read.push(node);
169        self
170    }
171}
172
173builder_base!(Read);
174
175impl UARequest for Read {
176    type Out = ReadResponse;
177
178    async fn send<'b>(self, channel: &'b AsyncSecureChannel) -> Result<Self::Out, Error>
179    where
180        Self: 'b,
181    {
182        let span = debug_span!(
183            "Sending Read request",
184            nodes_to_read = self.nodes_to_read.len(),
185            timestamps_to_return = ?self.timestamps_to_return,
186            max_age = self.max_age
187        );
188        let request = {
189            let _h = span.enter();
190            if self.nodes_to_read.is_empty() {
191                builder_error!(self, "read(), was not supplied with any nodes to read");
192                return Err(Error::new(
193                    StatusCode::BadNothingToDo,
194                    "read was not supplied with any nodes to read",
195                ));
196            }
197            ReadRequest {
198                request_header: self.header.header,
199                max_age: self.max_age,
200                timestamps_to_return: self.timestamps_to_return,
201                nodes_to_read: Some(self.nodes_to_read),
202            }
203        };
204
205        let response = channel
206            .send(request, self.header.timeout)
207            .instrument(span.clone())
208            .await?;
209
210        let _h = span.enter();
211        if let ResponseMessage::Read(response) = response {
212            builder_debug!(self, "read(), success");
213            process_service_result(&response.response_header)?;
214            Ok(*response)
215        } else {
216            builder_error!(self, "read() value failed");
217            Err(process_unexpected_response(response))
218        }
219    }
220}
221
222#[derive(Debug, Clone)]
223/// Reads historical values or events of one or more nodes. The caller is expected to provide
224/// a HistoryReadAction enum which must be one of the following:
225///
226/// * [`ReadEventDetails`]
227/// * [`ReadRawModifiedDetails`]
228/// * [`ReadProcessedDetails`]
229/// * [`ReadAtTimeDetails`]
230///
231/// See OPC UA Part 4 - Services 5.10.3 for complete description of the service and error responses.
232pub struct HistoryRead {
233    details: HistoryReadAction,
234    timestamps_to_return: TimestampsToReturn,
235    release_continuation_points: bool,
236    nodes_to_read: Vec<HistoryReadValueId>,
237
238    header: RequestHeaderBuilder,
239}
240
241builder_base!(HistoryRead);
242
243impl HistoryRead {
244    /// Create a new `HistoryRead` request.
245    pub fn new(details: HistoryReadAction, session: &Session) -> Self {
246        Self {
247            details,
248            timestamps_to_return: TimestampsToReturn::Neither,
249            release_continuation_points: false,
250            nodes_to_read: Vec::new(),
251
252            header: RequestHeaderBuilder::new_from_session(session),
253        }
254    }
255
256    /// Construct a new call to the `HistoryRead` service, setting header parameters manually.
257    pub fn new_manual(
258        details: HistoryReadAction,
259        session_id: u32,
260        timeout: Duration,
261        auth_token: NodeId,
262        request_handle: IntegerId,
263    ) -> Self {
264        Self {
265            details,
266            timestamps_to_return: TimestampsToReturn::Neither,
267            release_continuation_points: false,
268            nodes_to_read: Vec::new(),
269            header: RequestHeaderBuilder::new(session_id, timeout, auth_token, request_handle),
270        }
271    }
272
273    /// Set timestamps to return.
274    pub fn timestamps_to_return(mut self, timestamps: TimestampsToReturn) -> Self {
275        self.timestamps_to_return = timestamps;
276        self
277    }
278
279    /// Set release continuation points. Default is false, if this is true,
280    /// continuation points will be freed and the request will return without reading
281    /// any history.
282    pub fn release_continuation_points(mut self, release_continuation_points: bool) -> Self {
283        self.release_continuation_points = release_continuation_points;
284        self
285    }
286
287    /// Set nodes to read, overwriting any that were set previously.
288    pub fn nodes_to_read(mut self, nodes_to_read: Vec<HistoryReadValueId>) -> Self {
289        self.nodes_to_read = nodes_to_read;
290        self
291    }
292
293    /// Add a node to read.
294    pub fn node(mut self, node: HistoryReadValueId) -> Self {
295        self.nodes_to_read.push(node);
296        self
297    }
298}
299
300impl UARequest for HistoryRead {
301    type Out = HistoryReadResponse;
302
303    async fn send<'b>(self, channel: &'b AsyncSecureChannel) -> Result<Self::Out, Error>
304    where
305        Self: 'b,
306    {
307        let span = debug_span!(
308            "Sending HistoryRead request",
309            details = ?self.details,
310            timestamps_to_return = ?self.timestamps_to_return,
311            release_continuation_points = self.release_continuation_points,
312            num_nodes_to_read = self.nodes_to_read.len()
313        );
314        let request = {
315            let _h = span.enter();
316            let history_read_details = ExtensionObject::from(self.details);
317            builder_trace!(
318                self,
319                "history_read() requested to read nodes {:?}",
320                self.nodes_to_read
321            );
322            HistoryReadRequest {
323                request_header: self.header.header,
324                history_read_details,
325                timestamps_to_return: self.timestamps_to_return,
326                release_continuation_points: self.release_continuation_points,
327                nodes_to_read: if self.nodes_to_read.is_empty() {
328                    None
329                } else {
330                    Some(self.nodes_to_read)
331                },
332            }
333        };
334
335        let response = channel
336            .send(request, self.header.timeout)
337            .instrument(span.clone())
338            .await?;
339        let _h = span.enter();
340        if let ResponseMessage::HistoryRead(response) = response {
341            builder_debug!(self, "history_read(), success");
342            process_service_result(&response.response_header)?;
343            Ok(*response)
344        } else {
345            builder_error!(self, "history_read() value failed");
346            Err(process_unexpected_response(response))
347        }
348    }
349}
350
351#[derive(Debug, Clone)]
352/// Writes values to nodes by sending a [`WriteRequest`] to the server. Note that some servers may reject DataValues
353/// containing source or server timestamps.
354///
355/// See OPC UA Part 4 - Services 5.10.4 for complete description of the service and error responses.
356pub struct Write {
357    nodes_to_write: Vec<WriteValue>,
358
359    header: RequestHeaderBuilder,
360}
361
362builder_base!(Write);
363
364impl Write {
365    /// Construct a new call to the `Write` service.
366    pub fn new(session: &Session) -> Self {
367        Self {
368            nodes_to_write: Vec::new(),
369            header: RequestHeaderBuilder::new_from_session(session),
370        }
371    }
372
373    /// Construct a new call to the `Write` service, setting header parameters manually.
374    pub fn new_manual(
375        session_id: u32,
376        timeout: Duration,
377        auth_token: NodeId,
378        request_handle: IntegerId,
379    ) -> Self {
380        Self {
381            nodes_to_write: Vec::new(),
382            header: RequestHeaderBuilder::new(session_id, timeout, auth_token, request_handle),
383        }
384    }
385
386    /// Set nodes to write, overwriting any that were set previously.
387    pub fn nodes_to_write(mut self, nodes_to_write: Vec<WriteValue>) -> Self {
388        self.nodes_to_write = nodes_to_write;
389        self
390    }
391
392    /// Add a write value.
393    pub fn node(mut self, node: impl Into<WriteValue>) -> Self {
394        self.nodes_to_write.push(node.into());
395        self
396    }
397}
398
399impl UARequest for Write {
400    type Out = WriteResponse;
401
402    async fn send<'a>(self, channel: &'a AsyncSecureChannel) -> Result<Self::Out, Error>
403    where
404        Self: 'a,
405    {
406        let span = debug_span!(
407            "Sending Write request",
408            num_nodes_to_write = self.nodes_to_write.len()
409        );
410        let request = {
411            let _h = span.enter();
412            if self.nodes_to_write.is_empty() {
413                builder_error!(self, "write(), was not supplied with any nodes to write");
414                return Err(Error::new(
415                    StatusCode::BadNothingToDo,
416                    "write was not supplied with any nodes to write",
417                ));
418            }
419            WriteRequest {
420                request_header: self.header.header,
421                nodes_to_write: Some(self.nodes_to_write),
422            }
423        };
424        let response = channel
425            .send(request, self.header.timeout)
426            .instrument(span.clone())
427            .await?;
428        let _h = span.enter();
429        if let ResponseMessage::Write(response) = response {
430            builder_debug!(self, "write(), success");
431            process_service_result(&response.response_header)?;
432            Ok(*response)
433        } else {
434            builder_error!(self, "write() failed {:?}", response);
435            Err(process_unexpected_response(response))
436        }
437    }
438}
439
440#[derive(Debug, Clone)]
441/// Updates historical values. The caller is expected to provide one or more history update operations
442/// in a slice of HistoryUpdateAction enums which are one of the following:
443///
444/// * [`UpdateDataDetails`]
445/// * [`UpdateStructureDataDetails`]
446/// * [`UpdateEventDetails`]
447/// * [`DeleteRawModifiedDetails`]
448/// * [`DeleteAtTimeDetails`]
449/// * [`DeleteEventDetails`]
450///
451/// See OPC UA Part 4 - Services 5.10.5 for complete description of the service and error responses.
452pub struct HistoryUpdate {
453    details: Vec<HistoryUpdateAction>,
454
455    header: RequestHeaderBuilder,
456}
457
458builder_base!(HistoryUpdate);
459
460impl HistoryUpdate {
461    /// Construct a new call to the `HistoryUpdate` service.
462    pub fn new(session: &Session) -> Self {
463        Self {
464            details: Vec::new(),
465
466            header: RequestHeaderBuilder::new_from_session(session),
467        }
468    }
469
470    /// Construct a new call to the `HistoryUpdate` service, setting header parameters manually.
471    pub fn new_manual(
472        session_id: u32,
473        timeout: Duration,
474        auth_token: NodeId,
475        request_handle: IntegerId,
476    ) -> Self {
477        Self {
478            details: Vec::new(),
479            header: RequestHeaderBuilder::new(session_id, timeout, auth_token, request_handle),
480        }
481    }
482
483    /// Set the history update actions to perform.
484    pub fn details(mut self, details: Vec<HistoryUpdateAction>) -> Self {
485        self.details = details;
486        self
487    }
488
489    /// Add a history update action to the list.
490    pub fn action(mut self, action: impl Into<HistoryUpdateAction>) -> Self {
491        self.details.push(action.into());
492        self
493    }
494}
495
496impl UARequest for HistoryUpdate {
497    type Out = HistoryUpdateResponse;
498
499    async fn send<'a>(self, channel: &'a AsyncSecureChannel) -> Result<Self::Out, Error>
500    where
501        Self: 'a,
502    {
503        let span = debug_span!(
504            "Sending HistoryUpdate request",
505            num_details = self.details.len()
506        );
507        let request = {
508            let _h = span.enter();
509            if self.details.is_empty() {
510                builder_error!(
511                    self,
512                    "history_update(), was not supplied with any detail to update"
513                );
514                return Err(Error::new(
515                    StatusCode::BadNothingToDo,
516                    "history update was not supplied with any update details",
517                ));
518            }
519            let details = self
520                .details
521                .into_iter()
522                .map(ExtensionObject::from)
523                .collect();
524            HistoryUpdateRequest {
525                request_header: self.header.header,
526                history_update_details: Some(details),
527            }
528        };
529
530        let response = channel
531            .send(request, self.header.timeout)
532            .instrument(span.clone())
533            .await?;
534        let _h = span.enter();
535        if let ResponseMessage::HistoryUpdate(response) = response {
536            builder_error!(self, "history_update(), success");
537            process_service_result(&response.response_header)?;
538            Ok(*response)
539        } else {
540            builder_error!(self, "history_update() failed {:?}", response);
541            Err(process_unexpected_response(response))
542        }
543    }
544}
545
546impl Session {
547    /// Reads the value of nodes by sending a [`ReadRequest`] to the server.
548    ///
549    /// See OPC UA Part 4 - Services 5.10.2 for complete description of the service and error responses.
550    ///
551    /// # Arguments
552    ///
553    /// * `nodes_to_read` - A list of [`ReadValueId`] to be read by the server.
554    /// * `timestamps_to_return` - The [`TimestampsToReturn`] for each node, Both, Server, Source or None
555    /// * `max_age` - The maximum age of value to read in milliseconds. Read the service description
556    ///   for details. Basically it will attempt to read a value within the age range or
557    ///   attempt to read a new value. If 0 the server will attempt to read a new value from the datasource.
558    ///   If set to `i32::MAX` or greater, the server shall attempt to get a cached value.
559    ///
560    /// # Returns
561    ///
562    /// * `Ok(Vec<DataValue>)` - A list of [`DataValue`] corresponding to each read operation.
563    /// * `Err(Error)` - Request failed, [Status code](StatusCode) is the reason for failure.
564    ///
565    pub async fn read(
566        &self,
567        nodes_to_read: &[ReadValueId],
568        timestamps_to_return: TimestampsToReturn,
569        max_age: f64,
570    ) -> Result<Vec<DataValue>, Error> {
571        Ok(Read::new(self)
572            .nodes_to_read(nodes_to_read.to_vec())
573            .timestamps_to_return(timestamps_to_return)
574            .max_age(max_age)
575            .send(&self.channel)
576            .await?
577            .results
578            .unwrap_or_default())
579    }
580
581    /// Reads historical values or events of one or more nodes. The caller is expected to provide
582    /// a HistoryReadAction enum which must be one of the following:
583    ///
584    /// * [`ReadEventDetails`]
585    /// * [`ReadRawModifiedDetails`]
586    /// * [`ReadProcessedDetails`]
587    /// * [`ReadAtTimeDetails`]
588    ///
589    /// See OPC UA Part 4 - Services 5.10.3 for complete description of the service and error responses.
590    ///
591    /// # Arguments
592    ///
593    /// * `history_read_details` - A history read operation.
594    /// * `timestamps_to_return` - Enumeration of which timestamps to return.
595    /// * `release_continuation_points` - Flag indicating whether to release the continuation point for the operation.
596    /// * `nodes_to_read` - The list of [`HistoryReadValueId`] of the nodes to apply the history read operation to.
597    ///
598    /// # Returns
599    ///
600    /// * `Ok(Vec<HistoryReadResult>)` - A list of [`HistoryReadResult`] results corresponding to history read operation.
601    /// * `Err(Error)` - Request failed, [Status code](StatusCode) is the reason for failure.
602    ///
603    pub async fn history_read(
604        &self,
605        history_read_details: HistoryReadAction,
606        timestamps_to_return: TimestampsToReturn,
607        release_continuation_points: bool,
608        nodes_to_read: &[HistoryReadValueId],
609    ) -> Result<Vec<HistoryReadResult>, Error> {
610        Ok(HistoryRead::new(history_read_details, self)
611            .timestamps_to_return(timestamps_to_return)
612            .release_continuation_points(release_continuation_points)
613            .nodes_to_read(nodes_to_read.to_vec())
614            .send(&self.channel)
615            .await?
616            .results
617            .unwrap_or_default())
618    }
619
620    /// Writes values to nodes by sending a [`WriteRequest`] to the server. Note that some servers may reject DataValues
621    /// containing source or server timestamps.
622    ///
623    /// See OPC UA Part 4 - Services 5.10.4 for complete description of the service and error responses.
624    ///
625    /// # Arguments
626    ///
627    /// * `nodes_to_write` - A list of [`WriteValue`] to be sent to the server.
628    ///
629    /// # Returns
630    ///
631    /// * `Ok(Vec<StatusCode>)` - A list of [`StatusCode`] results corresponding to each write operation.
632    /// * `Err(Error)` - Request failed, [Status code](StatusCode) is the reason for failure.
633    ///
634    pub async fn write(&self, nodes_to_write: &[WriteValue]) -> Result<Vec<StatusCode>, Error> {
635        Ok(Write::new(self)
636            .nodes_to_write(nodes_to_write.to_vec())
637            .send(&self.channel)
638            .await?
639            .results
640            .unwrap_or_default())
641    }
642
643    /// Updates historical values. The caller is expected to provide one or more history update operations
644    /// in a slice of HistoryUpdateAction enums which are one of the following:
645    ///
646    /// * [`UpdateDataDetails`]
647    /// * [`UpdateStructureDataDetails`]
648    /// * [`UpdateEventDetails`]
649    /// * [`DeleteRawModifiedDetails`]
650    /// * [`DeleteAtTimeDetails`]
651    /// * [`DeleteEventDetails`]
652    ///
653    /// See OPC UA Part 4 - Services 5.10.5 for complete description of the service and error responses.
654    ///
655    /// # Arguments
656    ///
657    /// * `history_update_details` - A list of history update operations.
658    ///
659    /// # Returns
660    ///
661    /// * `Ok(Vec<HistoryUpdateResult>)` - A list of [`HistoryUpdateResult`] results corresponding to history update operation.
662    /// * `Err(Error)` - Request failed, [Status code](StatusCode) is the reason for failure.
663    ///
664    pub async fn history_update(
665        &self,
666        history_update_details: &[HistoryUpdateAction],
667    ) -> Result<Vec<HistoryUpdateResult>, Error> {
668        Ok(HistoryUpdate::new(self)
669            .details(history_update_details.to_vec())
670            .send(&self.channel)
671            .await?
672            .results
673            .unwrap_or_default())
674    }
675}
opcua_client/session/services/attributes.rs

opcua_client/session/services/
attributes.rs
