omry_integration/
bridge.rs

1//! Module with APIs that make it more convenient to test the Omry RPC server.
2use anyhow::{Context, Result};
3use omry_service::archiving::{Record, RecordMeta, RecordParams};
4use omry_service::common::RecordId;
5use omry_service::search::SearchResponse;
6use omry_service::search_wrappers::SearchParametersWrapper;
7use omry_service::service::RecordIds;
8use omry_service::tarpc::context;
9use omry_service::{FloraServiceClient, PONG};
10use std::sync::Arc;
11use tokio::task::JoinSet;
12use tracing::instrument;
13
14/// Trait to "bridge" the user-facing application with the flora backend
15/// over [`omry_service::tarpc`].
16#[allow(async_fn_in_trait)]
17pub trait ServiceBridge {
18    /// Returns a reference to the underlying tarpc client used by this bridge.
19    fn client(&self) -> &FloraServiceClient;
20
21    /// Pings the flora server.
22    ///
23    /// Returns the expected [`PONG`] response on success.
24    #[instrument(skip(self))]
25    async fn ping(&self) -> Result<String> {
26        let response = self.client().ping(context::current()).await?;
27        if response == PONG {
28            Ok(response)
29        } else {
30            anyhow::bail!("Unexpected ping response: {response}")
31        }
32    }
33
34    /// Adds a single record to the flora archive.
35    #[instrument(skip(self))]
36    async fn add_record(&self, record: RecordParams) -> Result<RecordId> {
37        let id = self
38            .client()
39            .add_record(context::current(), record)
40            .await?
41            .map_err(|e| anyhow::anyhow!("{e}"))?;
42        Ok(id)
43    }
44
45    /// Queries the backend for the number of records it currently stores.
46    #[instrument(skip(self))]
47    async fn count(&self) -> Result<i64> {
48        let count = self
49            .client()
50            .count(context::current())
51            .await?
52            .map_err(|e| anyhow::anyhow!("{e}"))?;
53        Ok(count)
54    }
55
56    /// Queries the flora backend for records with the given IDs.
57    #[instrument(skip(self))]
58    async fn get_records(&self, record_ids: RecordIds) -> Result<Arc<Vec<Record>>> {
59        let records = self
60            .client()
61            .get_records(context::current(), record_ids)
62            .await?
63            .map_err(|e| anyhow::anyhow!("{e}"))?;
64        Ok(records)
65    }
66
67    /// Returns the metadata for all the records in the SQLite database.
68    #[instrument(skip(self))]
69    async fn get_all_records_meta(&self) -> Result<Vec<RecordMeta>> {
70        let records = self
71            .client()
72            .get_all_records_meta(context::current())
73            .await
74            .context("Failed to invoke 'get_all_records_meta' RPC")?
75            .map_err(|e| anyhow::anyhow!("{e}"))?;
76
77        Ok(records)
78    }
79
80    /// Sends a search request to the flora backend.
81    #[instrument(skip(self))]
82    async fn search(&self, search_parameters: SearchParametersWrapper) -> Result<SearchResponse> {
83        let search_result = self
84            .client()
85            .search(context::current(), search_parameters)
86            .await?
87            .map_err(|e| anyhow::anyhow!("{e}"))?;
88
89        Ok(search_result)
90    }
91
92    /// Sends a requests for the Omry API to delete record(s) with
93    /// the given IDs.
94    #[instrument(skip(self))]
95    async fn delete(&self, ids: Vec<RecordId>) -> Result<usize> {
96        tracing::debug!(
97            "Will make {} RPCs to delete the given IDs (one per each)",
98            ids.len()
99        );
100
101        let mut set = JoinSet::new();
102        for id in ids {
103            let client = self.client().clone();
104
105            set.spawn(async move {
106                client
107                    .delete_record(context::current(), id)
108                    .await
109                    .map_err(|error| error.to_string())
110                    .flatten()
111            });
112        }
113
114        let mut n_deleted = 0usize;
115        let mut n_errors = 0usize;
116        let results = set.join_all().await;
117        for result in results {
118            match result {
119                Ok(n) => n_deleted += n,
120                Err(error) => {
121                    tracing::error!("Error calling `delete_record`: {error}");
122                    n_errors += 1;
123                }
124            }
125        }
126
127        if n_errors != 0 {
128            anyhow::bail!("{n_errors} records couldn't be scheduled for deletion due to errors.");
129        }
130
131        Ok(n_deleted)
132    }
133
134    /// Returns the version information for the server application.
135    #[instrument(skip(self))]
136    async fn version(&self) -> Result<String> {
137        let version = self
138            .client()
139            .version(context::current())
140            .await
141            .context("Failed to invoke 'version' RPC")?
142            .map_err(|e| anyhow::anyhow!("{e}"))?;
143
144        Ok(version)
145    }
146}
omry_integration/bridge.rs

omry_integration/
bridge.rs
