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

omry_integration/
bridge.rs
