opcda_bridge/
client.rs

1//! The connected gRPC client and typed search stream.
2
3use crate::error::{Error, Result};
4use crate::types::{
5    BrowsePage, BrowsePageRequest, Capabilities, SearchEvent, SearchRequest, TagValue, Value,
6    WriteResult,
7};
8use opcda_bridge_proto::bridge::bridge_client::BridgeClient;
9use opcda_bridge_proto::bridge::write_request::TypedValue;
10use opcda_bridge_proto::bridge::{
11    CloseBrowseSessionRequest, GetCapabilitiesRequest, ListServersRequest, ReadRequest,
12    WriteRequest,
13};
14use tonic::Code;
15use tonic::codec::Streaming;
16use tonic::transport::Channel;
17
18/// A connected client for an opcda-bridge gateway's gRPC API.
19#[derive(Debug)]
20pub struct Client {
21    inner: BridgeClient<Channel>,
22}
23
24/// A cancellable stream of typed namespace-search events.
25///
26/// Dropping this value drops the underlying gRPC stream, allowing the gateway
27/// to stop scheduling further search work.
28#[derive(Debug)]
29pub struct SearchStream {
30    inner: Streaming<opcda_bridge_proto::bridge::SearchEvent>,
31}
32
33impl SearchStream {
34    /// Wait for the next event. `None` means the server closed the stream.
35    pub async fn message(&mut self) -> Result<Option<SearchEvent>> {
36        self.inner
37            .message()
38            .await?
39            .map(SearchEvent::try_from)
40            .transpose()
41    }
42}
43
44impl Client {
45    /// Connect to a plaintext gateway at `host` (for example, `localhost:7600`).
46    pub async fn connect(host: &str) -> Result<Self> {
47        let inner = BridgeClient::connect(format!("http://{host}")).await?;
48        Ok(Self { inner })
49    }
50
51    /// Report protocol, paging, browse-session, search, and namespace support.
52    pub async fn capabilities(&mut self, server: impl Into<String>) -> Result<Capabilities> {
53        self.inner
54            .get_capabilities(GetCapabilitiesRequest {
55                server: server.into(),
56            })
57            .await
58            .map_err(|status| feature_error("capability discovery", status))?
59            .into_inner()
60            .try_into()
61    }
62
63    /// List the OPC DA servers registered on the gateway's host.
64    pub async fn list_servers(&mut self) -> Result<Vec<String>> {
65        let response = self
66            .inner
67            .list_servers(ListServersRequest {
68                host: "localhost".to_string(),
69            })
70            .await?;
71        Ok(response.into_inner().servers)
72    }
73
74    /// Open a browse session and return only its first root page.
75    pub async fn browse(
76        &mut self,
77        server: impl Into<String>,
78        page_size: u32,
79    ) -> Result<BrowsePage> {
80        self.open_browse(server, page_size).await
81    }
82
83    /// Open a browse session and return only its first root page.
84    pub async fn open_browse(
85        &mut self,
86        server: impl Into<String>,
87        page_size: u32,
88    ) -> Result<BrowsePage> {
89        self.browse_page(BrowsePageRequest::root(server, page_size))
90            .await
91    }
92
93    /// Request exactly one root, child, or continuation page.
94    ///
95    /// This method never follows `next_page_token` automatically.
96    pub async fn browse_page(&mut self, request: BrowsePageRequest) -> Result<BrowsePage> {
97        self.inner
98            .browse(opcda_bridge_proto::bridge::BrowseRequest::from(request))
99            .await
100            .map_err(|status| feature_error("paged browse", status))?
101            .into_inner()
102            .try_into()
103    }
104
105    /// Explicitly release a gateway browse session.
106    pub async fn close_browse_session(&mut self, session_id: impl Into<String>) -> Result<()> {
107        self.inner
108            .close_browse_session(CloseBrowseSessionRequest {
109                session_id: session_id.into(),
110            })
111            .await
112            .map_err(|status| feature_error("browse-session close", status))?;
113        Ok(())
114    }
115
116    /// Start a bounded search and return its progressive event stream.
117    pub async fn search_stream(&mut self, request: SearchRequest) -> Result<SearchStream> {
118        let inner = self
119            .inner
120            .search(opcda_bridge_proto::bridge::SearchRequest::from(request))
121            .await
122            .map_err(|status| feature_error("namespace search", status))?
123            .into_inner();
124        Ok(SearchStream { inner })
125    }
126
127    /// Explicitly collect a complete search stream into memory.
128    pub async fn search(&mut self, request: SearchRequest) -> Result<Vec<SearchEvent>> {
129        let mut stream = self.search_stream(request).await?;
130        let mut events = Vec::new();
131        while let Some(event) = stream.message().await? {
132            events.push(event);
133        }
134        Ok(events)
135    }
136
137    /// Read one or more exact OPC DA ItemIDs from `server`.
138    pub async fn read(&mut self, server: String, tags: Vec<String>) -> Result<Vec<TagValue>> {
139        let response = self
140            .inner
141            .read(ReadRequest {
142                server,
143                tag_ids: tags,
144            })
145            .await?;
146        Ok(response
147            .into_inner()
148            .values
149            .into_iter()
150            .map(|v| TagValue {
151                tag_id: v.tag_id,
152                value: v.value,
153                quality: v.quality,
154                timestamp: v.timestamp,
155            })
156            .collect())
157    }
158
159    /// Write `value` to one exact OPC DA ItemID on `server`.
160    pub async fn write(
161        &mut self,
162        server: String,
163        tag: String,
164        value: Value,
165    ) -> Result<WriteResult> {
166        let typed_value = match value {
167            Value::String(s) => TypedValue::StringValue(s),
168            Value::Int(i) => TypedValue::IntValue(i),
169            Value::Float(f) => TypedValue::FloatValue(f),
170            Value::Bool(b) => TypedValue::BoolValue(b),
171        };
172        let response = self
173            .inner
174            .write(WriteRequest {
175                server,
176                tag_id: tag,
177                typed_value: Some(typed_value),
178            })
179            .await?;
180        let result = response.into_inner();
181        Ok(WriteResult {
182            tag_id: result.tag_id,
183            success: result.success,
184            error: result.error,
185        })
186    }
187}
188
189fn feature_error(operation: &'static str, status: tonic::Status) -> Error {
190    if status.code() == Code::Unimplemented {
191        Error::IncompatibleGateway { operation }
192    } else {
193        Error::Rpc(status)
194    }
195}
196
197#[cfg(test)]
198mod tests {
199    use super::*;
200    use crate::error::Error;
201    use crate::test_support::{MockBridgeService, start_mock_server};
202    use crate::{BrowseNodeKind, BrowseSource, NamespaceOrganization, SearchMatchMode};
203    use opcda_bridge_proto::bridge::search_event;
204    use opcda_bridge_proto::bridge::{
205        BrowseNode as ProtoBrowseNode, BrowsePage as ProtoBrowsePage,
206        BrowseSource as ProtoBrowseSource, GetCapabilitiesResponse, ListServersResponse,
207        NamespaceOrganization as ProtoOrganization, ReadResponse, SearchCompleted,
208        SearchEvent as ProtoSearchEvent, SearchProgress, TagValue as ProtoTagValue, WriteResponse,
209    };
210    use std::sync::Arc;
211    use std::time::Duration;
212    use tonic::Status;
213
214    fn item_node() -> ProtoBrowseNode {
215        ProtoBrowseNode {
216            node_key: "node".into(),
217            display_name: "PV".into(),
218            kind: opcda_bridge_proto::bridge::BrowseNodeKind::Item as i32,
219            item_id: Some("FCS!TAG.PV".into()),
220        }
221    }
222
223    #[tokio::test]
224    async fn connect_success_and_failure_are_typed() {
225        let host = start_mock_server(MockBridgeService::default()).await;
226        Client::connect(&host).await.unwrap();
227        assert!(matches!(
228            Client::connect("127.0.0.1:1").await.unwrap_err(),
229            Error::Connect(_)
230        ));
231    }
232
233    #[tokio::test]
234    async fn mock_server_shutdown_completes() {
235        let service = MockBridgeService::default();
236        let shutdown = Arc::clone(&service.server_shutdown);
237        let stopped = Arc::clone(&service.server_stopped);
238        let _host = start_mock_server(service).await;
239        shutdown.notify_one();
240        tokio::time::timeout(Duration::from_secs(1), stopped.notified())
241            .await
242            .unwrap();
243    }
244
245    #[tokio::test]
246    async fn capabilities_maps_fields_and_request() {
247        let service = MockBridgeService {
248            capabilities_response: GetCapabilitiesResponse {
249                application_version: "0.3.0".into(),
250                protocol_version: "0.3".into(),
251                max_page_size: 1000,
252                supports_browse_sessions: true,
253                supports_search: true,
254                organization: ProtoOrganization::Hierarchical as i32,
255                source: ProtoBrowseSource::Da2 as i32,
256            },
257            ..Default::default()
258        };
259        let requests = Arc::clone(&service.capabilities_requests);
260        let host = start_mock_server(service).await;
261        let mut client = Client::connect(&host).await.unwrap();
262        let capabilities = client.capabilities("S").await.unwrap();
263        assert_eq!(
264            capabilities.organization,
265            NamespaceOrganization::Hierarchical
266        );
267        assert_eq!(capabilities.source, BrowseSource::Da2);
268        assert_eq!(requests.lock().unwrap()[0].server, "S");
269    }
270
271    #[tokio::test]
272    async fn capabilities_rpc_error_is_typed() {
273        let host = start_mock_server(MockBridgeService {
274            capabilities_error: Some(Status::unimplemented("old gateway")),
275            ..Default::default()
276        })
277        .await;
278        let mut client = Client::connect(&host).await.unwrap();
279        assert!(matches!(
280            client.capabilities("S").await.unwrap_err(),
281            Error::IncompatibleGateway { .. }
282        ));
283    }
284
285    #[tokio::test]
286    async fn list_servers_maps_data_and_errors() {
287        let host = start_mock_server(MockBridgeService {
288            list_servers_response: ListServersResponse {
289                servers: vec!["S1".into(), "S2".into()],
290            },
291            ..Default::default()
292        })
293        .await;
294        let mut client = Client::connect(&host).await.unwrap();
295        assert_eq!(client.list_servers().await.unwrap(), ["S1", "S2"]);
296
297        let host = start_mock_server(MockBridgeService {
298            list_servers_error: Some(Status::internal("boom")),
299            ..Default::default()
300        })
301        .await;
302        let mut client = Client::connect(&host).await.unwrap();
303        assert!(matches!(
304            client.list_servers().await.unwrap_err(),
305            Error::Rpc(_)
306        ));
307    }
308
309    #[tokio::test]
310    async fn browse_returns_one_typed_page_without_draining() {
311        let service = MockBridgeService {
312            browse_response: ProtoBrowsePage {
313                session_id: "session".into(),
314                nodes: vec![item_node()],
315                next_page_token: Some("next".into()),
316                complete: false,
317                organization: ProtoOrganization::Hierarchical as i32,
318                source: ProtoBrowseSource::Da3 as i32,
319                warning: None,
320            },
321            ..Default::default()
322        };
323        let requests = Arc::clone(&service.browse_requests);
324        let host = start_mock_server(service).await;
325        let mut client = Client::connect(&host).await.unwrap();
326        let page = client.browse("S", 25).await.unwrap();
327        assert_eq!(page.nodes[0].kind, BrowseNodeKind::Item);
328        assert_eq!(page.next_page_token.as_deref(), Some("next"));
329        let request = &requests.lock().unwrap()[0];
330        assert_eq!(request.page_size, 25);
331        assert!(request.session_id.is_none());
332    }
333
334    #[tokio::test]
335    async fn browse_page_forwards_session_parent_token_and_refresh() {
336        let service = MockBridgeService::default();
337        let requests = Arc::clone(&service.browse_requests);
338        let host = start_mock_server(service).await;
339        let mut client = Client::connect(&host).await.unwrap();
340        client
341            .browse_page(
342                BrowsePageRequest::next("S", "session", Some("parent".into()), "token", 10)
343                    .with_refresh(true),
344            )
345            .await
346            .unwrap();
347        let request = &requests.lock().unwrap()[0];
348        assert_eq!(request.session_id.as_deref(), Some("session"));
349        assert_eq!(request.parent_node_key.as_deref(), Some("parent"));
350        assert_eq!(request.page_token.as_deref(), Some("token"));
351        assert!(request.refresh);
352    }
353
354    #[tokio::test]
355    async fn browse_rpc_and_protocol_errors_are_typed() {
356        let host = start_mock_server(MockBridgeService {
357            browse_error: Some(Status::failed_precondition("expired")),
358            ..Default::default()
359        })
360        .await;
361        let mut client = Client::connect(&host).await.unwrap();
362        assert!(matches!(
363            client.open_browse("S", 20).await.unwrap_err(),
364            Error::Rpc(_)
365        ));
366
367        let host = start_mock_server(MockBridgeService {
368            browse_response: ProtoBrowsePage::default(),
369            ..Default::default()
370        })
371        .await;
372        let mut client = Client::connect(&host).await.unwrap();
373        assert!(matches!(
374            client.open_browse("S", 20).await.unwrap_err(),
375            Error::Protocol(_)
376        ));
377
378        let host = start_mock_server(MockBridgeService {
379            browse_error: Some(Status::unimplemented("old gateway")),
380            ..Default::default()
381        })
382        .await;
383        let mut client = Client::connect(&host).await.unwrap();
384        assert!(matches!(
385            client.open_browse("S", 20).await.unwrap_err(),
386            Error::IncompatibleGateway { .. }
387        ));
388    }
389
390    #[tokio::test]
391    async fn close_browse_session_forwards_id_and_error() {
392        let service = MockBridgeService::default();
393        let requests = Arc::clone(&service.close_requests);
394        let host = start_mock_server(service).await;
395        let mut client = Client::connect(&host).await.unwrap();
396        client.close_browse_session("session").await.unwrap();
397        assert_eq!(requests.lock().unwrap()[0].session_id, "session");
398
399        let host = start_mock_server(MockBridgeService {
400            close_error: Some(Status::not_found("missing")),
401            ..Default::default()
402        })
403        .await;
404        let mut client = Client::connect(&host).await.unwrap();
405        assert!(matches!(
406            client.close_browse_session("missing").await.unwrap_err(),
407            Error::Rpc(_)
408        ));
409    }
410
411    #[tokio::test]
412    async fn search_stream_and_collect_map_events_and_request() {
413        let events = vec![
414            ProtoSearchEvent {
415                event: Some(search_event::Event::Progress(SearchProgress {
416                    visited_nodes: 5,
417                    matches: 0,
418                    partial: true,
419                })),
420            },
421            ProtoSearchEvent {
422                event: Some(search_event::Event::Completed(SearchCompleted {
423                    complete: true,
424                    cancelled: false,
425                    truncated: false,
426                    warning: None,
427                })),
428            },
429        ];
430        let service = MockBridgeService {
431            search_events: events,
432            ..Default::default()
433        };
434        let requests = Arc::clone(&service.search_requests);
435        let host = start_mock_server(service).await;
436        let mut client = Client::connect(&host).await.unwrap();
437        let mut request = SearchRequest::new("S", "PV", SearchMatchMode::Prefix);
438        request.session_id = Some("session".into());
439        request.scope_node_key = Some("scope".into());
440        request.max_results = 50;
441        request.include_branches = true;
442        request.refresh = true;
443        let found = client.search(request).await.unwrap();
444        assert_eq!(found.len(), 2);
445        let request = &requests.lock().unwrap()[0];
446        assert_eq!(request.query, "PV");
447        assert_eq!(
448            request.match_mode,
449            opcda_bridge_proto::bridge::SearchMatchMode::Prefix as i32
450        );
451        assert!(request.include_branches);
452        assert!(request.refresh);
453    }
454
455    #[tokio::test]
456    async fn search_initial_stream_and_protocol_errors_are_typed() {
457        let host = start_mock_server(MockBridgeService {
458            search_initial_error: Some(Status::unavailable("down")),
459            ..Default::default()
460        })
461        .await;
462        let mut client = Client::connect(&host).await.unwrap();
463        assert!(matches!(
464            client
465                .search_stream(SearchRequest::new("S", "PV", SearchMatchMode::Exact))
466                .await
467                .unwrap_err(),
468            Error::Rpc(_)
469        ));
470
471        let host = start_mock_server(MockBridgeService {
472            search_stream_error: Some(Status::deadline_exceeded("slow")),
473            ..Default::default()
474        })
475        .await;
476        let mut client = Client::connect(&host).await.unwrap();
477        assert!(matches!(
478            client
479                .search(SearchRequest::new("S", "PV", SearchMatchMode::Exact))
480                .await
481                .unwrap_err(),
482            Error::Rpc(_)
483        ));
484
485        let host = start_mock_server(MockBridgeService {
486            search_events: vec![ProtoSearchEvent::default()],
487            ..Default::default()
488        })
489        .await;
490        let mut client = Client::connect(&host).await.unwrap();
491        assert!(matches!(
492            client
493                .search(SearchRequest::new("S", "PV", SearchMatchMode::Exact))
494                .await
495                .unwrap_err(),
496            Error::Protocol(_)
497        ));
498
499        assert!(matches!(
500            feature_error("test", Status::unimplemented("old")),
501            Error::IncompatibleGateway { .. }
502        ));
503    }
504
505    #[tokio::test]
506    async fn read_maps_data_and_errors() {
507        let host = start_mock_server(MockBridgeService {
508            read_response: ReadResponse {
509                values: vec![ProtoTagValue {
510                    tag_id: "t1".into(),
511                    value: "42".into(),
512                    quality: "Good".into(),
513                    timestamp: "now".into(),
514                }],
515            },
516            ..Default::default()
517        })
518        .await;
519        let mut client = Client::connect(&host).await.unwrap();
520        assert_eq!(
521            client.read("S".into(), vec![]).await.unwrap()[0].value,
522            "42"
523        );
524
525        let host = start_mock_server(MockBridgeService {
526            read_error: Some(Status::internal("boom")),
527            ..Default::default()
528        })
529        .await;
530        let mut client = Client::connect(&host).await.unwrap();
531        assert!(matches!(
532            client.read("S".into(), vec![]).await.unwrap_err(),
533            Error::Rpc(_)
534        ));
535    }
536
537    #[tokio::test]
538    async fn write_maps_every_value_and_result_or_error() {
539        for value in [
540            Value::Bool(true),
541            Value::Int(42),
542            Value::Float(3.5),
543            Value::String("text".into()),
544        ] {
545            let host = start_mock_server(MockBridgeService {
546                write_response: WriteResponse {
547                    tag_id: "t".into(),
548                    success: true,
549                    error: None,
550                },
551                ..Default::default()
552            })
553            .await;
554            let mut client = Client::connect(&host).await.unwrap();
555            assert!(
556                client
557                    .write("S".into(), "t".into(), value)
558                    .await
559                    .unwrap()
560                    .success
561            );
562        }
563
564        let host = start_mock_server(MockBridgeService {
565            write_error: Some(Status::internal("boom")),
566            ..Default::default()
567        })
568        .await;
569        let mut client = Client::connect(&host).await.unwrap();
570        assert!(matches!(
571            client
572                .write("S".into(), "t".into(), Value::Int(1))
573                .await
574                .unwrap_err(),
575            Error::Rpc(_)
576        ));
577    }
578}
opcda_bridge/client.rs

opcda_bridge/
client.rs
