odp_agent/
agent.rs

1use std::sync::Arc;
2
3use futures::{StreamExt, stream};
4use odp_core::{Offering, OfferingSearchRequest, Representation};
5use odp_directory::{
6    DirectoryClient, DirectoryService, Environment, IterationOptions, SearchRequest,
7};
8
9use crate::{AgentError, ServiceClient, TraversalOptions};
10
11pub trait ServiceClientFactory: Send + Sync {
12    fn create(&self, service: &DirectoryService) -> Result<ServiceClient, AgentError>;
13}
14
15struct DefaultServiceClientFactory;
16
17impl ServiceClientFactory for DefaultServiceClientFactory {
18    fn create(&self, service: &DirectoryService) -> Result<ServiceClient, AgentError> {
19        ServiceClient::new(&service.service_origin)
20    }
21}
22
23#[derive(Clone, Debug, Default, PartialEq)]
24pub struct FederatedSearchRequest {
25    pub concurrency: usize,
26    pub max_offerings_per_service: usize,
27    pub max_services: usize,
28    pub offerings: OfferingSearchRequest,
29    pub services: SearchRequest,
30}
31
32#[derive(Clone, Debug, PartialEq)]
33pub struct DiscoveryEvent {
34    pub issue: Option<String>,
35    pub offering: Option<Offering>,
36    pub service: DirectoryService,
37}
38
39pub struct Agent {
40    directory: DirectoryClient,
41    factory: Arc<dyn ServiceClientFactory>,
42}
43
44impl Agent {
45    pub fn new(environment: Environment) -> Result<Self, AgentError> {
46        let directory = DirectoryClient::new(environment)
47            .map_err(|error| AgentError::Directory(error.to_string()))?;
48        Ok(Self {
49            directory,
50            factory: Arc::new(DefaultServiceClientFactory),
51        })
52    }
53
54    pub fn with_clients(
55        directory: DirectoryClient,
56        factory: Arc<dyn ServiceClientFactory>,
57    ) -> Self {
58        Self { directory, factory }
59    }
60
61    pub const fn environment(&self) -> Environment {
62        self.directory.environment()
63    }
64
65    pub async fn search_offerings_across_services(
66        &self,
67        request: &FederatedSearchRequest,
68    ) -> Result<Vec<DiscoveryEvent>, AgentError> {
69        let maximum_services = bounded(request.max_services, 10, 100, "max_services")?;
70        let maximum_offerings = bounded(
71            request.max_offerings_per_service,
72            10,
73            100,
74            "max_offerings_per_service",
75        )?;
76        let concurrency = bounded(request.concurrency, 4, 16, "concurrency")?;
77        let services = self
78            .directory
79            .search_services(
80                &request.services,
81                IterationOptions {
82                    max_items: maximum_services,
83                    max_pages: 0,
84                },
85            )
86            .await
87            .map_err(|error| AgentError::Directory(error.to_string()))?;
88        let offerings = request.offerings.clone();
89        let factory = &self.factory;
90        let results = stream::iter(services.into_iter().map(|service| {
91            let offerings = offerings.clone();
92            async move {
93                let result =
94                    search_service(factory.as_ref(), &service, &offerings, maximum_offerings).await;
95                (service, result)
96            }
97        }))
98        .buffered(concurrency)
99        .collect::<Vec<_>>()
100        .await;
101        Ok(results
102            .into_iter()
103            .flat_map(|(service, result)| match result {
104                Ok(offerings) => offerings
105                    .into_iter()
106                    .map(|offering| DiscoveryEvent {
107                        issue: None,
108                        offering: Some(offering),
109                        service: service.clone(),
110                    })
111                    .collect(),
112                Err(error) => vec![DiscoveryEvent {
113                    issue: Some(error.to_string()),
114                    offering: None,
115                    service,
116                }],
117            })
118            .collect())
119    }
120}
121
122async fn search_service(
123    factory: &dyn ServiceClientFactory,
124    service: &DirectoryService,
125    request: &OfferingSearchRequest,
126    maximum: usize,
127) -> Result<Vec<Offering>, AgentError> {
128    let client = factory.create(service)?;
129    let traversal = TraversalOptions {
130        max_items: maximum,
131        max_pages: 0,
132    };
133    if has_search(request) {
134        client
135            .search_all_offerings(request, Representation::Terse, traversal)
136            .await
137    } else {
138        client
139            .list_all_offerings(Representation::Terse, 0, traversal)
140            .await
141    }
142}
143
144fn has_search(request: &OfferingSearchRequest) -> bool {
145    !request.query.is_empty()
146        || !request.filters.is_empty()
147        || request.include_descendants
148        || !request.sort.is_empty()
149        || !request.refinements.is_empty()
150        || !request.collection_id.is_empty()
151}
152
153fn bounded(value: usize, fallback: usize, maximum: usize, name: &str) -> Result<usize, AgentError> {
154    let value = if value == 0 { fallback } else { value };
155    if value > maximum {
156        return Err(AgentError::InvalidRequest(format!(
157            "{name} must be from 1 through {maximum}"
158        )));
159    }
160    Ok(value)
161}
162
163#[cfg(test)]
164mod tests {
165    use std::{collections::BTreeMap, sync::Arc};
166
167    use async_trait::async_trait;
168    use odp_directory::{HttpRequest, HttpResponse, Transport, TransportError};
169
170    use super::*;
171
172    struct DirectoryTransport;
173
174    #[async_trait]
175    impl Transport for DirectoryTransport {
176        async fn send(&self, _request: HttpRequest) -> Result<HttpResponse, TransportError> {
177            Ok(json_response(br#"{"items":[{"description":"One","indexed_at":"2026-08-25T00:00:00Z","language":"en","localizations":["en"],"name":"One","operations":[],"service_origin":"https://one.example"},{"description":"Two","indexed_at":"2026-08-25T00:00:00Z","language":"en","localizations":["en"],"name":"Two","operations":[],"service_origin":"https://two.example"}]}"#))
178        }
179    }
180
181    struct ServiceTransport;
182
183    #[async_trait]
184    impl Transport for ServiceTransport {
185        async fn send(&self, request: HttpRequest) -> Result<HttpResponse, TransportError> {
186            if request.url.ends_with("/.well-known/odp") {
187                return Ok(odp_response(br#"{"description":"Plants","http":{"endpoint_base":"/odp"},"language":"en","localizations":["en"],"name":"Plants","odp_version":"1.0","operations":[{"authentication":"not-required","name":"get-offering"},{"authentication":"not-required","name":"list-offerings"}]}"#));
188            }
189            let id = if request.url.starts_with("https://one.example") {
190                "one"
191            } else {
192                "two"
193            };
194            Ok(odp_response(
195                format!(
196                    r#"{{"items":[{{"id":"{id}","name":"Plant {id}","odp_version":"1.0"}}],"odp_version":"1.0"}}"#
197                )
198                .as_bytes(),
199            ))
200        }
201    }
202
203    struct Factory;
204
205    impl ServiceClientFactory for Factory {
206        fn create(&self, service: &DirectoryService) -> Result<ServiceClient, AgentError> {
207            ServiceClient::with_transport(&service.service_origin, Arc::new(ServiceTransport))
208        }
209    }
210
211    fn json_response(body: &[u8]) -> HttpResponse {
212        HttpResponse {
213            body: body.to_vec(),
214            headers: BTreeMap::from([("content-type".to_owned(), "application/json".to_owned())]),
215            status: 200,
216        }
217    }
218
219    fn odp_response(body: &[u8]) -> HttpResponse {
220        HttpResponse {
221            body: body.to_vec(),
222            headers: BTreeMap::from([(
223                "content-type".to_owned(),
224                "application/odp+json".to_owned(),
225            )]),
226            status: 200,
227        }
228    }
229
230    #[tokio::test]
231    async fn preserves_directory_order_across_concurrent_service_searches() {
232        let directory =
233            DirectoryClient::with_transport(Environment::Production, Arc::new(DirectoryTransport));
234        let agent = Agent::with_clients(directory, Arc::new(Factory));
235        let events = agent
236            .search_offerings_across_services(&FederatedSearchRequest {
237                concurrency: 2,
238                ..FederatedSearchRequest::default()
239            })
240            .await
241            .unwrap();
242        assert_eq!(events.len(), 2);
243        assert_eq!(events[0].service.name, "One");
244        assert_eq!(events[1].service.name, "Two");
245    }
246}
odp_agent/agent.rs

odp_agent/
agent.rs
