opc_da_client/
provider.rs

1use crate::opc_da::errors::{OpcError, OpcResult};
2use async_trait::async_trait;
3use std::sync::Arc;
4use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
5use tokio::sync::mpsc;
6use uuid::Uuid;
7
8#[cfg(feature = "test-support")]
9use mockall::automock;
10
11/// A single tag's read result.
12///
13/// Returned by [`OpcProvider::read_tag_values`] and
14/// [`OpcProvider::read_tag_values_for_display`].
15///
16/// # Examples
17///
18/// ```
19/// use opc_da_client::TagValue;
20///
21/// let tv = TagValue {
22///     tag_id: "Simulation.Random.1".to_string(),
23///     value: "42.5".to_string(),
24///     quality: "Good".to_string(),
25///     timestamp: "2026-01-01 00:00:00".to_string(),
26/// };
27/// assert_eq!(tv.tag_id, "Simulation.Random.1");
28/// ```
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct TagValue {
31    /// The fully qualified tag identifier (e.g., `"Channel1.Device1.Tag1"`).
32    pub tag_id: String,
33    /// The current value as a string representation.
34    ///
35    /// [`OpcProvider::read_tag_values`] preserves `VT_BSTR` contents exactly.
36    /// [`OpcProvider::read_tag_values_for_display`] may add presentation quotes
37    /// around BSTR values.
38    pub value: String,
39    /// OPC quality indicator (e.g., `"Good"`, `"Bad"`, or `"Uncertain"`).
40    pub quality: String,
41    /// Timestamp of the last value change, formatted as a local time string.
42    pub timestamp: String,
43}
44
45/// Typed value to write to an OPC DA tag.
46///
47/// # Examples
48///
49/// ```
50/// use opc_da_client::OpcValue;
51///
52/// let v = OpcValue::Float(3.14);
53/// assert_eq!(v, OpcValue::Float(3.14));
54/// ```
55#[derive(Debug, Clone, PartialEq)]
56pub enum OpcValue {
57    /// String value (`VT_BSTR`) — server may coerce to target type.
58    String(String),
59    /// 32-bit integer (`VT_I4`).
60    Int(i32),
61    /// 64-bit float (`VT_R8`).
62    Float(f64),
63    /// Boolean (`VT_BOOL`).
64    Bool(bool),
65}
66
67/// Result of a single write operation.
68///
69/// # Examples
70///
71/// ```
72/// use opc_da_client::WriteResult;
73///
74/// let wr = WriteResult {
75///     tag_id: "Tag1".to_string(),
76///     success: true,
77///     error: None,
78/// };
79/// assert!(wr.success);
80/// ```
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct WriteResult {
83    /// The tag that was written to.
84    pub tag_id: String,
85    /// Whether the write succeeded.
86    pub success: bool,
87    /// Error message if the write failed, `None` on success.
88    pub error: Option<String>,
89}
90
91/// OPC DA address-space organization reported by a server.
92#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93pub enum BrowseNamespace {
94    /// All item IDs live in one flat namespace.
95    Flat,
96    /// Items are organized under browsable branches.
97    Hierarchical,
98    /// The server supports DA 3.0 browsing but does not expose the DA 2.x
99    /// namespace query needed to classify its organization.
100    Unknown,
101}
102
103/// Native browse features available from an OPC DA server.
104#[derive(Debug, Clone, Copy, PartialEq, Eq)]
105pub struct BrowseCapabilities {
106    /// Address-space organization reported by the server.
107    pub namespace: BrowseNamespace,
108    /// Whether native OPC DA 3.0 `IOPCBrowse` is available.
109    pub supports_da3: bool,
110    /// Whether OPC DA 2.x `IOPCBrowseServerAddressSpace` is available.
111    pub supports_da2: bool,
112    /// Largest page size accepted by [`OpcProvider::browse_page`].
113    pub max_page_size: u32,
114}
115
116macro_rules! opaque_browse_token {
117    ($name:ident, $doc:literal) => {
118        #[doc = $doc]
119        #[derive(Clone, Copy, PartialEq, Eq, Hash)]
120        pub struct $name(Uuid);
121
122        impl $name {
123            pub(crate) fn new() -> Self {
124                Self(Uuid::new_v4())
125            }
126
127            /// Parse a token previously encoded with [`ToString::to_string`].
128            ///
129            /// This allows transport adapters to round-trip opaque tokens
130            /// without accessing their internal representation.
131            pub fn parse(value: &str) -> Result<Self, uuid::Error> {
132                value.parse()
133            }
134        }
135
136        impl std::fmt::Debug for $name {
137            fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
138                formatter
139                    .debug_tuple(stringify!($name))
140                    .field(&self.0)
141                    .finish()
142            }
143        }
144
145        impl std::fmt::Display for $name {
146            fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
147                self.0.fmt(formatter)
148            }
149        }
150
151        impl std::str::FromStr for $name {
152            type Err = uuid::Error;
153
154            fn from_str(value: &str) -> Result<Self, Self::Err> {
155                value.parse().map(Self)
156            }
157        }
158    };
159}
160
161opaque_browse_token!(
162    BrowseSessionToken,
163    "Opaque identifier for a browse session owned by the COM worker."
164);
165opaque_browse_token!(
166    BrowseNodeToken,
167    "Opaque identifier for a node returned by a browse session."
168);
169opaque_browse_token!(
170    BrowsePageToken,
171    "Opaque continuation token for the next bounded browse page."
172);
173
174/// Kinds of nodes returned by a native browse.
175#[derive(Debug, Clone, Copy, PartialEq, Eq)]
176pub enum BrowseNodeKind {
177    /// A branch that can have children but is not itself an item.
178    Branch,
179    /// An item that has no browsable children.
180    Item,
181    /// A node that is both an item and a branch.
182    BranchAndItem,
183}
184
185impl BrowseNodeKind {
186    /// Returns whether the node can be used as the parent of another browse.
187    pub fn has_children(self) -> bool {
188        matches!(self, Self::Branch | Self::BranchAndItem)
189    }
190
191    /// Returns whether the node identifies a selectable OPC item.
192    pub fn is_item(self) -> bool {
193        matches!(self, Self::Item | Self::BranchAndItem)
194    }
195}
196
197/// Node-kind filter for a one-level browse request.
198#[derive(Debug, Clone, Copy, PartialEq, Eq)]
199pub enum BrowseNodeFilter {
200    /// Return only branches.
201    Branches,
202    /// Return only items.
203    Items,
204    /// Return both branches and items.
205    All,
206}
207
208/// One address-space node returned by [`OpcProvider::browse_page`].
209#[derive(Debug, Clone, PartialEq, Eq)]
210pub struct BrowseNode {
211    /// Opaque token used to browse this node's immediate children.
212    pub token: BrowseNodeToken,
213    /// Display name relative to the requested parent.
214    pub name: String,
215    /// Exact fully-qualified item ID when the server supplies one.
216    pub item_id: Option<String>,
217    /// Whether this node is a branch, item, or both.
218    pub kind: BrowseNodeKind,
219}
220
221/// Parameters for one bounded, non-recursive browse operation.
222#[derive(Debug, Clone, Copy, PartialEq, Eq)]
223pub struct BrowsePageRequest {
224    /// Parent node, or `None` to browse the root.
225    pub parent: Option<BrowseNodeToken>,
226    /// Node kinds to return.
227    pub filter: BrowseNodeFilter,
228    /// Maximum number of nodes to return.
229    pub max_elements: u32,
230    /// Opaque continuation returned by the preceding page.
231    pub continuation: Option<BrowsePageToken>,
232}
233
234/// One bounded page of immediate address-space children.
235#[derive(Debug, Clone, PartialEq, Eq)]
236pub struct BrowsePage {
237    /// Nodes returned in this page.
238    pub nodes: Vec<BrowseNode>,
239    /// Opaque token for the next page, or `None` when enumeration is complete.
240    pub continuation: Option<BrowsePageToken>,
241}
242
243/// Options controlling one bounded namespace inventory.
244#[derive(Debug, Clone, Copy, PartialEq, Eq)]
245pub struct InventoryOptions {
246    /// Maximum number of native entries requested per browse operation.
247    pub batch_size: u32,
248    /// Optional safety cap for a deliberately bounded inventory.
249    pub max_entries: Option<u64>,
250}
251
252impl Default for InventoryOptions {
253    fn default() -> Self {
254        Self {
255            batch_size: 100,
256            max_entries: None,
257        }
258    }
259}
260
261/// One selectable OPC DA item discovered during inventory.
262#[derive(Debug, Clone, PartialEq, Eq)]
263pub struct InventoryEntry {
264    /// Local display name returned by the server.
265    pub display_name: String,
266    /// Exact ItemID returned by the server.
267    pub item_id: String,
268    /// Whether the item is also a browsable branch.
269    pub kind: BrowseNodeKind,
270    /// Stable display labels for the item's ancestors.
271    pub breadcrumbs: Vec<String>,
272}
273
274/// Progress emitted between bounded inventory operations.
275#[derive(Debug, Clone, PartialEq)]
276pub struct InventoryProgress {
277    pub branches_visited: u64,
278    pub entries_seen: u64,
279    pub unique_items: u64,
280    pub active_time_ms: u64,
281    pub paused_time_ms: u64,
282    pub items_per_second: f64,
283    pub estimated_remaining_ms: Option<u64>,
284}
285
286/// Terminal result for one inventory operation.
287#[derive(Debug, Clone, PartialEq, Eq)]
288pub struct InventoryCompleted {
289    pub complete: bool,
290    pub cancelled: bool,
291    pub truncated: bool,
292    pub warning: Option<String>,
293    pub capabilities: BrowseCapabilities,
294}
295
296/// Event emitted by [`InventoryStream`].
297#[derive(Debug, Clone, PartialEq)]
298pub enum InventoryEvent {
299    Entry(InventoryEntry),
300    Progress(InventoryProgress),
301    Completed(InventoryCompleted),
302}
303
304#[derive(Debug)]
305struct InventoryControlState {
306    cancelled: AtomicBool,
307    paused: AtomicBool,
308}
309
310/// Control handle for a running inventory.
311#[derive(Clone, Debug)]
312pub struct InventoryControl {
313    state: Arc<InventoryControlState>,
314}
315
316impl InventoryControl {
317    pub(crate) fn new() -> Self {
318        Self {
319            state: Arc::new(InventoryControlState {
320                cancelled: AtomicBool::new(false),
321                paused: AtomicBool::new(false),
322            }),
323        }
324    }
325
326    /// Request cancellation at the next bounded COM boundary.
327    pub fn cancel(&self) {
328        self.state.cancelled.store(true, Ordering::Release);
329    }
330
331    /// Pause before the next bounded COM operation.
332    pub fn pause(&self) {
333        self.state.paused.store(true, Ordering::Release);
334    }
335
336    /// Resume a paused inventory.
337    pub fn resume(&self) {
338        self.state.paused.store(false, Ordering::Release);
339    }
340
341    /// Return whether cancellation has been requested.
342    pub fn is_cancelled(&self) -> bool {
343        self.state.cancelled.load(Ordering::Acquire)
344    }
345
346    pub(crate) fn is_paused(&self) -> bool {
347        self.state.paused.load(Ordering::Acquire)
348    }
349}
350
351/// Cancellable stream of bounded inventory events.
352pub struct InventoryStream {
353    receiver: mpsc::Receiver<OpcResult<InventoryEvent>>,
354    control: InventoryControl,
355    worker: Option<std::thread::JoinHandle<()>>,
356}
357
358impl InventoryStream {
359    pub(crate) fn new(
360        receiver: mpsc::Receiver<OpcResult<InventoryEvent>>,
361        control: InventoryControl,
362        worker: std::thread::JoinHandle<()>,
363    ) -> Self {
364        Self {
365            receiver,
366            control,
367            worker: Some(worker),
368        }
369    }
370
371    /// Wait for the next inventory event.
372    ///
373    /// A failed inventory is delivered as `Some(Err(_))`, which is the
374    /// terminal event for the stream. A successful or cancelled inventory
375    /// ends with an [`InventoryEvent::Completed`] message.
376    pub async fn message(&mut self) -> Option<OpcResult<InventoryEvent>> {
377        self.receiver.recv().await
378    }
379
380    /// Return a control handle for this inventory.
381    pub fn control(&self) -> InventoryControl {
382        self.control.clone()
383    }
384
385    /// Request cancellation of this inventory.
386    pub fn cancel(&self) {
387        self.control.cancel();
388    }
389
390    /// Pause this inventory before its next bounded operation.
391    pub fn pause(&self) {
392        self.control.pause();
393    }
394
395    /// Resume this inventory.
396    pub fn resume(&self) {
397        self.control.resume();
398    }
399}
400
401impl Drop for InventoryStream {
402    fn drop(&mut self) {
403        // Close the receiver before joining so a worker blocked on a full
404        // event channel can observe the disconnect and finish.
405        self.receiver.close();
406        self.control.cancel();
407        if let Some(worker) = self.worker.take() {
408            let _ = worker.join();
409        }
410    }
411}
412
413#[cfg(test)]
414mod inventory_stream_tests {
415    use super::*;
416
417    #[test]
418    fn dropping_inventory_stream_cancels_and_joins_worker() {
419        let control = InventoryControl::new();
420        let worker_control = control.clone();
421        let finished = Arc::new(AtomicBool::new(false));
422        let worker_finished = Arc::clone(&finished);
423        let (_sender, receiver) = mpsc::channel(1);
424        let worker = std::thread::spawn(move || {
425            while !worker_control.is_cancelled() {
426                std::thread::yield_now();
427            }
428            worker_finished.store(true, Ordering::Release);
429        });
430
431        drop(InventoryStream::new(receiver, control, worker));
432        assert!(finished.load(Ordering::Acquire));
433    }
434}
435
436#[cfg(test)]
437mod read_display_fallback_tests {
438    use super::*;
439
440    struct FallbackProvider;
441
442    #[async_trait]
443    impl OpcProvider for FallbackProvider {
444        async fn list_servers(&self, _host: &str) -> OpcResult<Vec<String>> {
445            Ok(Vec::new())
446        }
447
448        async fn browse_tags(
449            &self,
450            _server: &str,
451            _max_tags: usize,
452            _progress: Arc<AtomicUsize>,
453            _tags_sink: Arc<std::sync::Mutex<Vec<String>>>,
454        ) -> OpcResult<Vec<String>> {
455            Ok(Vec::new())
456        }
457
458        async fn read_tag_values(
459            &self,
460            _server: &str,
461            tag_ids: Vec<String>,
462        ) -> OpcResult<Vec<TagValue>> {
463            Ok(tag_ids
464                .into_iter()
465                .map(|tag_id| TagValue {
466                    tag_id,
467                    value: "AUT".to_string(),
468                    quality: "Good".to_string(),
469                    timestamp: String::new(),
470                })
471                .collect())
472        }
473
474        async fn write_tag_value(
475            &self,
476            _server: &str,
477            tag_id: &str,
478            _value: OpcValue,
479        ) -> OpcResult<WriteResult> {
480            Ok(WriteResult {
481                tag_id: tag_id.to_string(),
482                success: true,
483                error: None,
484            })
485        }
486    }
487
488    #[tokio::test]
489    async fn display_read_defaults_to_semantic_read() {
490        let values = FallbackProvider
491            .read_tag_values_for_display("Server", vec!["Tag".to_string()])
492            .await
493            .unwrap();
494
495        assert_eq!(values[0].value, "AUT");
496    }
497}
498
499/// Async trait for OPC DA operations.
500///
501/// This is the stable public API. Backend implementations provide
502/// the actual COM/DCOM interaction.
503#[cfg_attr(feature = "test-support", automock)]
504#[async_trait]
505pub trait OpcProvider: Send + Sync {
506    /// List available OPC DA servers on the given host.
507    ///
508    /// # Errors
509    /// Returns `Err` if COM initialization fails or the server registry
510    /// cannot be enumerated.
511    async fn list_servers(&self, host: &str) -> OpcResult<Vec<String>>;
512
513    /// Browse tags recursively, pushing discoveries to `tags_sink`.
514    ///
515    /// # Errors
516    /// Returns `Err` if the server connection fails, the `ProgID` cannot be
517    /// resolved, or the namespace walk encounters an unrecoverable error.
518    async fn browse_tags(
519        &self,
520        server: &str,
521        max_tags: usize,
522        progress: Arc<AtomicUsize>,
523        tags_sink: Arc<std::sync::Mutex<Vec<String>>>,
524    ) -> OpcResult<Vec<String>>;
525
526    /// Return the native browse capabilities of an OPC DA server.
527    ///
528    /// # Errors
529    /// Returns `Err` if the server cannot be connected or exposes no supported
530    /// OPC DA browse interface.
531    async fn browse_capabilities(&self, server: &str) -> OpcResult<BrowseCapabilities> {
532        let _ = server;
533        Err(OpcError::NotImplemented(
534            "Native browsing is not implemented by this provider".to_string(),
535        ))
536    }
537
538    /// Open an isolated native browse session with its own server connection.
539    ///
540    /// # Errors
541    /// Returns `Err` if the server cannot be connected, browsing is unsupported,
542    /// or the worker's bounded session capacity has been reached.
543    async fn open_browse_session(&self, server: &str) -> OpcResult<BrowseSessionToken> {
544        let _ = server;
545        Err(OpcError::NotImplemented(
546            "Native browsing is not implemented by this provider".to_string(),
547        ))
548    }
549
550    /// Browse one bounded level of an open native browse session.
551    ///
552    /// # Errors
553    /// Returns `Err` for invalid, closed, or expired tokens; invalid page sizes;
554    /// unsupported requests; or underlying OPC browse failures.
555    async fn browse_page(
556        &self,
557        session: &BrowseSessionToken,
558        request: BrowsePageRequest,
559    ) -> OpcResult<BrowsePage> {
560        let _ = (session, request);
561        Err(OpcError::NotImplemented(
562            "Native browsing is not implemented by this provider".to_string(),
563        ))
564    }
565
566    /// Explicitly close a native browse session and release its server connection.
567    ///
568    /// # Errors
569    /// Returns `Err` if the session token is invalid, expired, or already closed.
570    async fn close_browse_session(&self, session: &BrowseSessionToken) -> OpcResult<()> {
571        let _ = session;
572        Err(OpcError::NotImplemented(
573            "Native browsing is not implemented by this provider".to_string(),
574        ))
575    }
576
577    /// Start a cancellable, bounded namespace inventory on a separate
578    /// connection from foreground operations.
579    ///
580    /// The returned stream never exposes interactive browse-session tokens;
581    /// all traversal state remains private to the inventory worker.
582    async fn start_inventory(
583        &self,
584        server: &str,
585        options: InventoryOptions,
586    ) -> OpcResult<InventoryStream> {
587        let _ = (server, options);
588        Err(OpcError::NotImplemented(
589            "Namespace inventory is not implemented by this provider".to_string(),
590        ))
591    }
592
593    /// Read current values for the given tag IDs.
594    ///
595    /// `VT_BSTR` values preserve their exact COM contents. No quote characters
596    /// are added or removed, so this method is suitable for machine consumers.
597    ///
598    /// # Errors
599    /// Returns `Err` if the server connection fails, no items can be added
600    /// to the OPC group, or the synchronous read operation fails.
601    async fn read_tag_values(&self, server: &str, tag_ids: Vec<String>)
602    -> OpcResult<Vec<TagValue>>;
603
604    /// Read current values formatted for human-readable display.
605    ///
606    /// The native provider wraps `VT_BSTR` contents in quote characters while
607    /// leaving all other value formatting unchanged. The default implementation
608    /// delegates to [`Self::read_tag_values`] so third-party providers remain
609    /// source-compatible.
610    ///
611    /// # Errors
612    /// Returns the same errors as [`Self::read_tag_values`].
613    async fn read_tag_values_for_display(
614        &self,
615        server: &str,
616        tag_ids: Vec<String>,
617    ) -> OpcResult<Vec<TagValue>> {
618        self.read_tag_values(server, tag_ids).await
619    }
620
621    /// Write a value to a single OPC DA tag.
622    ///
623    /// # Errors
624    /// Returns `Err` if the server connection fails, the tag cannot be added
625    /// to the OPC group, or the synchronous write operation fails.
626    async fn write_tag_value(
627        &self,
628        server: &str,
629        tag_id: &str,
630        value: OpcValue,
631    ) -> OpcResult<WriteResult>;
632}
opc_da_client/provider.rs

opc_da_client/
provider.rs
