omry_integration/
bridge.rs

1//! Module with APIs that make it more convenient to test the Omry RPC server.
2use anyhow::Result;
3use omry_service::archiving::{Record, 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    /// Sends a search request to the flora backend.
68    #[instrument(skip(self))]
69    async fn search(&self, search_parameters: SearchParametersWrapper) -> Result<SearchResponse> {
70        let search_result = self
71            .client()
72            .search(context::current(), search_parameters)
73            .await?
74            .map_err(|e| anyhow::anyhow!("{e}"))?;
75
76        Ok(search_result)
77    }
78
79    /// Sends a requests for the Omry API to delete record(s) with
80    /// the given IDs.
81    #[instrument(skip(self))]
82    async fn delete(&self, ids: Vec<RecordId>) -> Result<usize> {
83        tracing::debug!(
84            "Will make {} RPCs to delete the given IDs (one per each)",
85            ids.len()
86        );
87
88        let mut set = JoinSet::new();
89        for id in ids {
90            let client = self.client().clone();
91
92            set.spawn(async move {
93                client
94                    .delete_record(context::current(), id)
95                    .await
96                    .map_err(|error| error.to_string())
97                    .flatten()
98            });
99        }
100
101        let mut n_deleted = 0usize;
102        let mut n_errors = 0usize;
103        let results = set.join_all().await;
104        for result in results {
105            match result {
106                Ok(n) => n_deleted += n,
107                Err(error) => {
108                    tracing::error!("Error calling `delete_record`: {error}");
109                    n_errors += 1;
110                }
111            }
112        }
113
114        if n_errors != 0 {
115            anyhow::bail!("{n_errors} records couldn't be scheduled for deletion due to errors.");
116        }
117
118        Ok(n_deleted)
119    }
120}
omry_integration/bridge.rs

omry_integration/
bridge.rs
