omry-integration 0.12.0

Helpers for Omry client applications that run integration tests against the Omry backend.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

//! Module with APIs that make it more convenient to test the Omry RPC server.
use anyhow::Result;
use omry_service::archiving::{Record, RecordParams};
use omry_service::common::RecordId;
use omry_service::search::SearchResponse;
use omry_service::search_wrappers::SearchParametersWrapper;
use omry_service::service::RecordIds;
use omry_service::tarpc::context;
use omry_service::{FloraServiceClient, PONG};
use std::sync::Arc;
use tokio::task::JoinSet;
use tracing::instrument;

/// Trait to "bridge" the user-facing application with the flora backend
/// over [`omry_service::tarpc`].
#[allow(async_fn_in_trait)]
pub trait ServiceBridge {
    /// Returns a reference to the underlying tarpc client used by this bridge.
    fn client(&self) -> &FloraServiceClient;

    /// Pings the flora server.
    ///
    /// Returns the expected [`PONG`] response on success.
    #[instrument(skip(self))]
    async fn ping(&self) -> Result<String> {
        let response = self.client().ping(context::current()).await?;
        if response == PONG {
            Ok(response)
        } else {
            anyhow::bail!("Unexpected ping response: {response}")
        }
    }

    /// Adds a single record to the flora archive.
    #[instrument(skip(self))]
    async fn add_record(&self, record: RecordParams) -> Result<RecordId> {
        let id = self
            .client()
            .add_record(context::current(), record)
            .await?
            .map_err(|e| anyhow::anyhow!("{e}"))?;
        Ok(id)
    }

    /// Queries the backend for the number of records it currently stores.
    #[instrument(skip(self))]
    async fn count(&self) -> Result<i64> {
        let count = self
            .client()
            .count(context::current())
            .await?
            .map_err(|e| anyhow::anyhow!("{e}"))?;
        Ok(count)
    }

    /// Queries the flora backend for records with the given IDs.
    #[instrument(skip(self))]
    async fn get_records(&self, record_ids: RecordIds) -> Result<Arc<Vec<Record>>> {
        let records = self
            .client()
            .get_records(context::current(), record_ids)
            .await?
            .map_err(|e| anyhow::anyhow!("{e}"))?;
        Ok(records)
    }

    /// Sends a search request to the flora backend.
    #[instrument(skip(self))]
    async fn search(&self, search_parameters: SearchParametersWrapper) -> Result<SearchResponse> {
        let search_result = self
            .client()
            .search(context::current(), search_parameters)
            .await?
            .map_err(|e| anyhow::anyhow!("{e}"))?;

        Ok(search_result)
    }

    /// Sends a requests for the Omry API to delete record(s) with
    /// the given IDs.
    #[instrument(skip(self))]
    async fn delete(&self, ids: Vec<RecordId>) -> Result<usize> {
        tracing::debug!(
            "Will make {} RPCs to delete the given IDs (one per each)",
            ids.len()
        );

        let mut set = JoinSet::new();
        for id in ids {
            let client = self.client().clone();

            set.spawn(async move {
                client
                    .delete_record(context::current(), id)
                    .await
                    .map_err(|error| error.to_string())
                    .flatten()
            });
        }

        let mut n_deleted = 0usize;
        let mut n_errors = 0usize;
        let results = set.join_all().await;
        for result in results {
            match result {
                Ok(n) => n_deleted += n,
                Err(error) => {
                    tracing::error!("Error calling `delete_record`: {error}");
                    n_errors += 1;
                }
            }
        }

        if n_errors != 0 {
            anyhow::bail!("{n_errors} records couldn't be scheduled for deletion due to errors.");
        }

        Ok(n_deleted)
    }
}