opc_da_client/backend/
connector.rs

1//! Abstractions for OPC DA server connectivity.
2//!
3//! Defines the [`ServerConnector`], [`ConnectedServer`], and [`ConnectedGroup`]
4//! traits that decouple [`super::opc_da::OpcDaClient`] from concrete COM types.
5//! This enables mock implementations for unit testing without a live COM server.
6
7pub use crate::bindings::da::tagOPCITEMDEF;
8pub use crate::bindings::da::{tagOPCITEMRESULT, tagOPCITEMSTATE};
9pub use crate::opc_da::client::*;
10pub use crate::opc_da::com_utils::RemoteArray;
11use crate::opc_da::errors::{E_INVALIDARG_HRESULT, is_com_hresult};
12pub use crate::opc_da::errors::{OpcError, OpcResult};
13use crate::provider::BrowseNodeFilter;
14use anyhow::Context;
15pub use windows::Win32::System::Variant::VARIANT;
16use windows::core::Interface;
17
18/// Rust-native OPC DA 3.0 browse element used inside the backend boundary.
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub struct NativeBrowseElement {
21    pub(crate) name: String,
22    pub(crate) item_id: Option<String>,
23    pub(crate) has_children: bool,
24    pub(crate) is_item: bool,
25}
26
27/// Rust-native OPC DA 3.0 page used inside the backend boundary.
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub struct NativeBrowsePage {
30    pub(crate) elements: Vec<NativeBrowseElement>,
31    pub(crate) more_elements: bool,
32    pub(crate) continuation: Option<String>,
33}
34
35#[derive(Debug, Clone, Copy, PartialEq, Eq)]
36pub enum Da2BranchNavigation {
37    Navigable,
38    RejectedInvalidArgument,
39}
40
41#[derive(Debug, Clone, PartialEq, Eq)]
42pub struct Da2BranchClassification {
43    pub item_id: Option<String>,
44    pub navigation: Da2BranchNavigation,
45}
46
47pub fn classify_da2_branch<S: ConnectedServer>(
48    server: &S,
49    item_name: &str,
50) -> OpcResult<Da2BranchClassification> {
51    let item_id = match server.resolve_da2_item_id(item_name) {
52        Ok(item_id) => item_id,
53        Err(error) if is_com_hresult(&error, E_INVALIDARG_HRESULT) => None,
54        Err(error) => return Err(error),
55    };
56    let down = crate::bindings::da::OPC_BROWSE_DOWN.0.cast_unsigned();
57    let up = crate::bindings::da::OPC_BROWSE_UP.0.cast_unsigned();
58    let navigation = match server.change_browse_position(down, item_name) {
59        Ok(()) => {
60            server.change_browse_position(up, "")?;
61            Da2BranchNavigation::Navigable
62        }
63        Err(error) if is_com_hresult(&error, E_INVALIDARG_HRESULT) => {
64            Da2BranchNavigation::RejectedInvalidArgument
65        }
66        Err(error) => return Err(error),
67    };
68    Ok(Da2BranchClassification {
69        item_id,
70        navigation,
71    })
72}
73
74/// Object-safe string enumerator used to keep DA 2.x COM enumeration state on
75/// the worker while allowing tests to supply pure Rust iterators.
76pub trait BrowseStringIterator {
77    fn next_string(&mut self) -> Option<OpcResult<String>>;
78}
79
80impl<T> BrowseStringIterator for T
81where
82    T: Iterator<Item = OpcResult<String>>,
83{
84    fn next_string(&mut self) -> Option<OpcResult<String>> {
85        self.next()
86    }
87}
88
89/// Factory for connecting to OPC DA servers.
90///
91/// Abstracts the concrete COM client usage so that tests can inject mocks
92/// that return pre-configured server/group results without a live COM runtime.
93///
94/// # Errors
95///
96/// All methods return `OpcResult` — implementations should wrap COM errors
97/// with contextual messages.
98pub trait ServerConnector: Send + Sync {
99    /// The server facade type returned by [`Self::connect`].
100    type Server: ConnectedServer;
101
102    /// Enumerate all OPC DA server ProgIDs on the local machine.
103    ///
104    /// # Errors
105    ///
106    /// Returns an error if the COM registry enumeration fails.
107    fn enumerate_servers(&self) -> OpcResult<Vec<String>>;
108
109    /// Connect to the named OPC DA server and return a server facade.
110    ///
111    /// # Errors
112    ///
113    /// Returns an error if the COM server cannot be created or connected.
114    fn connect(&self, server_name: &str) -> OpcResult<Self::Server>;
115}
116
117/// Facade over a connected OPC DA server instance.
118///
119/// Wraps namespace browsing and group management operations in Rust-native types.
120///
121/// # Errors
122///
123/// All methods return `OpcResult` — COM errors are propagated with context.
124pub trait ConnectedServer {
125    /// The group facade type returned by [`Self::add_group`].
126    type Group: ConnectedGroup;
127
128    /// Query the server's namespace organization type.
129    ///
130    /// Returns `OPC_NS_FLAT` or `OPC_NS_HIERARCHICAL` as a `u32`.
131    ///
132    /// # Errors
133    ///
134    /// Returns an error if the COM call fails.
135    fn query_organization(&self) -> OpcResult<u32>;
136
137    /// Browse the server's address space for item IDs of the given type.
138    ///
139    /// # Errors
140    ///
141    /// Returns an error if the COM browse call fails.
142    fn browse_opc_item_ids(
143        &self,
144        browse_type: u32,
145        filter: Option<&str>,
146        data_type: u16,
147        access_rights: u32,
148    ) -> OpcResult<StringIterator>;
149
150    /// Change the current browse position (e.g., navigate into/out of branches).
151    ///
152    /// # Errors
153    ///
154    /// Returns an error if the position change is rejected by the server.
155    fn change_browse_position(&self, direction: u32, name: &str) -> OpcResult<()>;
156
157    /// Resolve a browse name to its fully-qualified item ID.
158    ///
159    /// # Errors
160    ///
161    /// Returns an error if the server cannot resolve the item name.
162    fn get_item_id(&self, item_name: &str) -> OpcResult<String>;
163
164    /// Resolve a DA 2.x browse name only when it is also an item.
165    ///
166    /// OPC DA servers commonly report `OPC_E_UNKNOWNITEMID` or
167    /// `OPC_E_INVALIDITEMID` when `GetItemID` is called for a branch-only
168    /// browse name. Other failures remain hard errors.
169    fn resolve_da2_item_id(&self, item_name: &str) -> OpcResult<Option<String>> {
170        match self.get_item_id(item_name) {
171            Ok(item_id) => Ok(Some(item_id)),
172            Err(OpcError::Com { source })
173                if matches!(source.code().0.cast_unsigned(), 0xC004_0007 | 0xC004_0008) =>
174            {
175                Ok(None)
176            }
177            Err(error) => Err(error),
178        }
179    }
180
181    /// Return whether a DA 2.x item name also identifies a child branch.
182    ///
183    /// Backends that can probe branch navigation should override this method.
184    fn da2_name_has_children(&self, _item_name: &str) -> OpcResult<bool> {
185        Ok(false)
186    }
187
188    /// Return whether OPC DA 2.x address-space browsing is available.
189    fn supports_da2_browse(&self) -> bool {
190        true
191    }
192
193    /// Return whether OPC DA 3.0 native browsing is available.
194    fn supports_da3_browse(&self) -> bool {
195        false
196    }
197
198    /// Start a stateful OPC DA 2.x string enumeration.
199    ///
200    /// The returned iterator remains on the COM worker and is never exposed
201    /// through the public API.
202    fn begin_da2_browse(
203        &self,
204        browse_type: u32,
205        filter: Option<&str>,
206        data_type: u16,
207        access_rights: u32,
208    ) -> OpcResult<Box<dyn BrowseStringIterator>> {
209        Ok(Box::new(self.browse_opc_item_ids(
210            browse_type,
211            filter,
212            data_type,
213            access_rights,
214        )?))
215    }
216
217    /// Return one native OPC DA 3.0 browse page.
218    ///
219    /// The continuation value is backend-private and is replaced with an
220    /// opaque random token before crossing the public API boundary.
221    fn browse_da3(
222        &self,
223        _item_id: Option<&str>,
224        _continuation: Option<&str>,
225        _max_elements: u32,
226        _filter: BrowseNodeFilter,
227    ) -> OpcResult<NativeBrowsePage> {
228        Err(OpcError::NotImplemented(
229            "IOPCBrowse is not supported".to_string(),
230        ))
231    }
232
233    /// Add a new OPC group to this server connection.
234    ///
235    /// # Errors
236    ///
237    /// Returns an error if the group creation fails.
238    #[allow(clippy::too_many_arguments)]
239    fn add_group(
240        &self,
241        name: &str,
242        active: bool,
243        update_rate: u32,
244        client_handle: GroupHandle,
245        time_bias: i32,
246        percent_deadband: f32,
247        locale_id: u32,
248        revised_update_rate: &mut u32,
249        server_handle: &mut GroupHandle,
250    ) -> OpcResult<Self::Group>;
251
252    /// Remove an OPC group by its server-assigned handle.
253    ///
254    /// # Errors
255    ///
256    /// Returns an error if the group removal fails.
257    fn remove_group(&self, server_group: GroupHandle, force: bool) -> OpcResult<()>;
258}
259
260/// Facade over an OPC DA group for item management and I/O.
261///
262/// # Errors
263///
264/// All methods return `OpcResult` — COM errors are propagated with context.
265pub trait ConnectedGroup {
266    /// Add items to this group for monitoring.
267    ///
268    /// # Errors
269    ///
270    /// Returns an error if the COM `AddItems` call fails.
271    fn add_items(
272        &self,
273        items: &[tagOPCITEMDEF],
274    ) -> OpcResult<(
275        RemoteArray<tagOPCITEMRESULT>,
276        RemoteArray<windows::core::HRESULT>,
277    )>;
278
279    /// Perform a synchronous read of the given server handles.
280    ///
281    /// # Errors
282    ///
283    /// Returns an error if the COM `Read` call fails.
284    fn read(
285        &self,
286        source: crate::bindings::da::tagOPCDATASOURCE,
287        server_handles: &[ItemHandle],
288    ) -> OpcResult<(
289        RemoteArray<tagOPCITEMSTATE>,
290        RemoteArray<windows::core::HRESULT>,
291    )>;
292
293    /// Write values to the given server handles.
294    ///
295    /// # Errors
296    ///
297    /// Returns an error if the COM `Write` call fails.
298    fn write(
299        &self,
300        server_handles: &[ItemHandle],
301        values: &[VARIANT],
302    ) -> OpcResult<RemoteArray<windows::core::HRESULT>>;
303}
304
305// ── COM-backed implementations ──────────────────────────────────────
306
307/// Real COM-backed server connector implementation.
308///
309/// Uses Windows COM to enumerate and connect to OPC DA servers.
310pub struct ComConnector;
311
312impl ServerConnector for ComConnector {
313    type Server = ComServer;
314
315    fn enumerate_servers(&self) -> OpcResult<Vec<String>> {
316        let client = crate::opc_da::client::v2::Client;
317        let guid_iter = client
318            .get_servers()
319            .context("Failed to enumerate OPC DA servers from registry")?;
320
321        let mut servers = Vec::new();
322        for guid in guid_iter.flatten() {
323            // SAFETY: `crate::opc_da::GUID` and `windows::core::GUID` are both `#[repr(C)]` structs with identical layout.
324            // SAFETY: Validated by a `const_assert_eq!` in `opc_da/client/iterator.rs`.
325            let win_guid: windows::core::GUID = unsafe { std::mem::transmute_copy(&guid) };
326            if win_guid == windows::core::GUID::zeroed() {
327                continue;
328            }
329
330            if let Ok(progid) = crate::helpers::guid_to_progid(&win_guid)
331                && !progid.is_empty()
332            {
333                servers.push(progid);
334            }
335        }
336        servers.sort();
337        servers.dedup();
338        Ok(servers)
339    }
340
341    fn connect(&self, server_name: &str) -> OpcResult<Self::Server> {
342        let opc_server = crate::helpers::connect_server(server_name)?;
343        let unknown: windows::core::IUnknown = opc_server.cast()?;
344
345        Ok(ComServer {
346            server: opc_server,
347            common: unknown.cast()?,
348            connection_point_container: unknown.cast()?,
349            item_properties: unknown.cast()?,
350            server_public_groups: unknown.cast().ok(),
351            browse_server_address_space: unknown.cast().ok(),
352            browse: unknown.cast().ok(),
353        })
354    }
355}
356
357/// COM-backed [`ConnectedServer`].
358pub struct ComServer {
359    pub(crate) server: crate::bindings::da::IOPCServer,
360    pub(crate) common: crate::bindings::comn::IOPCCommon,
361    pub(crate) connection_point_container: windows::Win32::System::Com::IConnectionPointContainer,
362    pub(crate) item_properties: crate::bindings::da::IOPCItemProperties,
363    pub(crate) server_public_groups: Option<crate::bindings::da::IOPCServerPublicGroups>,
364    pub(crate) browse_server_address_space:
365        Option<crate::bindings::da::IOPCBrowseServerAddressSpace>,
366    pub(crate) browse: Option<crate::bindings::da::IOPCBrowse>,
367}
368
369impl ServerTrait<ComGroup> for ComServer {
370    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCServer> {
371        Ok(&self.server)
372    }
373}
374
375impl CommonTrait for ComServer {
376    fn interface(&self) -> OpcResult<&crate::bindings::comn::IOPCCommon> {
377        Ok(&self.common)
378    }
379}
380
381impl ConnectionPointContainerTrait for ComServer {
382    fn interface(&self) -> OpcResult<&windows::Win32::System::Com::IConnectionPointContainer> {
383        Ok(&self.connection_point_container)
384    }
385}
386
387impl ItemPropertiesTrait for ComServer {
388    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCItemProperties> {
389        Ok(&self.item_properties)
390    }
391}
392
393impl ServerPublicGroupsTrait for ComServer {
394    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCServerPublicGroups> {
395        self.server_public_groups.as_ref().ok_or_else(|| {
396            OpcError::NotImplemented("IOPCServerPublicGroups not supported".to_string())
397        })
398    }
399}
400
401impl BrowseServerAddressSpaceTrait for ComServer {
402    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCBrowseServerAddressSpace> {
403        self.browse_server_address_space.as_ref().ok_or_else(|| {
404            OpcError::NotImplemented("IOPCBrowseServerAddressSpace not supported".to_string())
405        })
406    }
407}
408
409impl BrowseTrait for ComServer {
410    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCBrowse> {
411        self.browse
412            .as_ref()
413            .ok_or_else(|| OpcError::NotImplemented("IOPCBrowse not supported".to_string()))
414    }
415}
416
417impl ConnectedServer for ComServer {
418    type Group = ComGroup;
419
420    fn query_organization(&self) -> OpcResult<u32> {
421        let org = BrowseServerAddressSpaceTrait::query_organization(self)?;
422        Ok(org.0.cast_unsigned())
423    }
424
425    fn browse_opc_item_ids(
426        &self,
427        browse_type: u32,
428        filter: Option<&str>,
429        data_type: u16,
430        access_rights: u32,
431    ) -> OpcResult<StringIterator> {
432        BrowseServerAddressSpaceTrait::browse_opc_item_ids(
433            self,
434            crate::bindings::da::tagOPCBROWSETYPE(browse_type.cast_signed()),
435            filter,
436            data_type,
437            access_rights,
438        )
439    }
440
441    fn change_browse_position(&self, direction: u32, name: &str) -> OpcResult<()> {
442        BrowseServerAddressSpaceTrait::change_browse_position(
443            self,
444            crate::bindings::da::tagOPCBROWSEDIRECTION(direction.cast_signed()),
445            name,
446        )
447    }
448
449    fn get_item_id(&self, item_name: &str) -> OpcResult<String> {
450        BrowseServerAddressSpaceTrait::get_item_id(self, item_name)
451    }
452
453    fn da2_name_has_children(&self, item_name: &str) -> OpcResult<bool> {
454        let down = crate::bindings::da::OPC_BROWSE_DOWN.0.cast_unsigned();
455        let up = crate::bindings::da::OPC_BROWSE_UP.0.cast_unsigned();
456        match ConnectedServer::change_browse_position(self, down, item_name) {
457            Ok(()) => {
458                ConnectedServer::change_browse_position(self, up, "")?;
459                Ok(true)
460            }
461            Err(OpcError::Com { source })
462                if !matches!(
463                    source.code().0.cast_unsigned(),
464                    0x8007_06BA | 0x8007_06BF | 0x8007_06BE | 0x8008_0005
465                ) =>
466            {
467                Ok(false)
468            }
469            Err(error) => Err(error),
470        }
471    }
472
473    fn supports_da2_browse(&self) -> bool {
474        self.browse_server_address_space.is_some()
475    }
476
477    fn supports_da3_browse(&self) -> bool {
478        self.browse.is_some()
479    }
480
481    fn browse_da3(
482        &self,
483        item_id: Option<&str>,
484        continuation: Option<&str>,
485        max_elements: u32,
486        filter: BrowseNodeFilter,
487    ) -> OpcResult<NativeBrowsePage> {
488        use crate::bindings::da::{
489            OPC_BROWSE_FILTER_ALL, OPC_BROWSE_FILTER_BRANCHES, OPC_BROWSE_FILTER_ITEMS,
490            OPC_BROWSE_HASCHILDREN, OPC_BROWSE_ISITEM,
491        };
492        use crate::opc_da::com_utils::RemotePointer;
493
494        let native_filter = match filter {
495            BrowseNodeFilter::Branches => OPC_BROWSE_FILTER_BRANCHES,
496            BrowseNodeFilter::Items => OPC_BROWSE_FILTER_ITEMS,
497            BrowseNodeFilter::All => OPC_BROWSE_FILTER_ALL,
498        };
499        let (more_elements, continuation, elements) = BrowseTrait::browse(
500            self,
501            item_id,
502            continuation,
503            max_elements,
504            native_filter,
505            None::<&str>,
506            None::<&str>,
507            false,
508            false,
509            &[],
510        )?;
511
512        let owned_strings: Vec<_> = elements
513            .as_slice()
514            .iter()
515            .map(|element| {
516                (
517                    RemotePointer::from(element.szName),
518                    RemotePointer::from(element.szItemID),
519                    element.dwFlagValue,
520                )
521            })
522            .collect();
523
524        let mut mapped = Vec::with_capacity(owned_strings.len());
525        for (name, item_id, flags) in owned_strings {
526            mapped.push(NativeBrowseElement {
527                name: String::try_from(name)?,
528                item_id: Option::<String>::try_from(item_id)?.filter(|value| !value.is_empty()),
529                has_children: flags & OPC_BROWSE_HASCHILDREN != 0,
530                is_item: flags & OPC_BROWSE_ISITEM != 0,
531            });
532        }
533
534        Ok(NativeBrowsePage {
535            elements: mapped,
536            more_elements,
537            continuation: continuation.filter(|value| !value.is_empty()),
538        })
539    }
540
541    fn add_group(
542        &self,
543        name: &str,
544        active: bool,
545        update_rate: u32,
546        client_handle: GroupHandle,
547        time_bias: i32,
548        percent_deadband: f32,
549        locale_id: u32,
550        revised_update_rate: &mut u32,
551        server_handle: &mut GroupHandle,
552    ) -> OpcResult<Self::Group> {
553        ServerTrait::add_group(
554            self,
555            name,
556            active,
557            update_rate,
558            client_handle,
559            time_bias,
560            percent_deadband,
561            locale_id,
562            revised_update_rate,
563            server_handle,
564        )
565    }
566
567    fn remove_group(&self, server_group: GroupHandle, force: bool) -> OpcResult<()> {
568        ServerTrait::remove_group(self, server_group, force)
569    }
570}
571
572pub struct ComGroup {
573    pub(crate) item_mgt: crate::bindings::da::IOPCItemMgt,
574    pub(crate) group_state_mgt: crate::bindings::da::IOPCGroupStateMgt,
575    pub(crate) public_group_state_mgt: Option<crate::bindings::da::IOPCPublicGroupStateMgt>,
576    pub(crate) sync_io: crate::bindings::da::IOPCSyncIO,
577    pub(crate) async_io: Option<crate::bindings::da::IOPCAsyncIO>,
578    pub(crate) async_io2: crate::bindings::da::IOPCAsyncIO2,
579    pub(crate) connection_point_container: windows::Win32::System::Com::IConnectionPointContainer,
580    pub(crate) data_object: Option<windows::Win32::System::Com::IDataObject>,
581}
582
583impl ItemMgtTrait for ComGroup {
584    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCItemMgt> {
585        Ok(&self.item_mgt)
586    }
587}
588
589impl GroupStateMgtTrait for ComGroup {
590    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCGroupStateMgt> {
591        Ok(&self.group_state_mgt)
592    }
593}
594
595impl PublicGroupStateMgtTrait for ComGroup {
596    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCPublicGroupStateMgt> {
597        self.public_group_state_mgt.as_ref().ok_or_else(|| {
598            OpcError::NotImplemented("IOPCPublicGroupStateMgt not supported".to_string())
599        })
600    }
601}
602
603impl SyncIoTrait for ComGroup {
604    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCSyncIO> {
605        Ok(&self.sync_io)
606    }
607}
608
609impl AsyncIoTrait for ComGroup {
610    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCAsyncIO> {
611        self.async_io
612            .as_ref()
613            .ok_or_else(|| OpcError::NotImplemented("IOPCAsyncIO not supported".to_string()))
614    }
615}
616
617impl AsyncIo2Trait for ComGroup {
618    fn interface(&self) -> OpcResult<&crate::bindings::da::IOPCAsyncIO2> {
619        Ok(&self.async_io2)
620    }
621}
622
623impl ConnectionPointContainerTrait for ComGroup {
624    fn interface(&self) -> OpcResult<&windows::Win32::System::Com::IConnectionPointContainer> {
625        Ok(&self.connection_point_container)
626    }
627}
628
629impl DataObjectTrait for ComGroup {
630    fn interface(&self) -> OpcResult<&windows::Win32::System::Com::IDataObject> {
631        self.data_object
632            .as_ref()
633            .ok_or_else(|| OpcError::NotImplemented("IDataObject not supported".to_string()))
634    }
635}
636
637impl ConnectedGroup for ComGroup {
638    fn add_items(
639        &self,
640        items: &[tagOPCITEMDEF],
641    ) -> OpcResult<(
642        RemoteArray<tagOPCITEMRESULT>,
643        RemoteArray<windows::core::HRESULT>,
644    )> {
645        ItemMgtTrait::add_items(self, items)
646    }
647
648    fn read(
649        &self,
650        source: crate::bindings::da::tagOPCDATASOURCE,
651        server_handles: &[ItemHandle],
652    ) -> OpcResult<(
653        RemoteArray<tagOPCITEMSTATE>,
654        RemoteArray<windows::core::HRESULT>,
655    )> {
656        SyncIoTrait::read(self, source, server_handles)
657    }
658
659    fn write(
660        &self,
661        server_handles: &[ItemHandle],
662        values: &[VARIANT],
663    ) -> OpcResult<RemoteArray<windows::core::HRESULT>> {
664        SyncIoTrait::write(self, server_handles, values)
665    }
666}
667
668impl TryFrom<windows::core::IUnknown> for ComGroup {
669    type Error = windows::core::Error;
670
671    fn try_from(unknown: windows::core::IUnknown) -> Result<Self, Self::Error> {
672        Ok(Self {
673            item_mgt: unknown.cast()?,
674            group_state_mgt: unknown.cast()?,
675            public_group_state_mgt: unknown.cast().ok(),
676            sync_io: unknown.cast()?,
677            async_io: unknown.cast().ok(),
678            async_io2: unknown.cast()?,
679            connection_point_container: unknown.cast()?,
680            data_object: unknown.cast().ok(),
681        })
682    }
683}
opc_da_client/backend/connector.rs

opc_da_client/backend/
connector.rs
