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

opc_da_client/
provider.rs
