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

opc_da_client/backend/
connector.rs
