launchdarkly_server_sdk/
data_system_builders.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use launchdarkly_sdk_transport::{HttpTransport, HyperTransport};
5use thiserror::Error;
6
7use crate::data_source_builders::{DataSourceFactory, PollingDataSourceBuilder};
8use crate::data_system::DataSystem;
9use crate::fdv2::data_system::{FDv2DataSystem, InitializerFactory, SynchronizerFactory};
10use crate::fdv2::fdv1_adapter::FDv1AdapterFactory;
11use crate::fdv2::polling::{PollingInitializerFactory, PollingSynchronizerFactory};
12use crate::fdv2::request_headers::RequestHeaders;
13use crate::fdv2::streaming::StreamingSynchronizerFactory;
14use crate::service_endpoints::ServiceEndpoints;
15
16const DEFAULT_INITIAL_RECONNECT_DELAY: Duration = Duration::from_secs(1);
17const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(30);
18const DEFAULT_FALLBACK_TIMEOUT: Duration = Duration::from_secs(120);
19const DEFAULT_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300);
20
21/// Error returned when a data system configuration cannot be built.
22#[non_exhaustive]
23#[derive(Debug, Error)]
24pub enum BuildError {
25    /// The data system configuration was invalid.
26    #[error("data system config failed to build: {0}")]
27    InvalidConfig(String),
28}
29
30/// The inputs a data source needs to build itself.
31#[non_exhaustive]
32pub struct DataSourceBuildContext<'a> {
33    /// The configured service endpoints.
34    pub endpoints: &'a ServiceEndpoints,
35    /// The HTTP headers to attach to every request.
36    pub headers: &'a RequestHeaders,
37}
38
39/// A configured FDv2 source usable as a synchronizer.
40pub trait FDv2SynchronizerConfig {
41    /// Builds a synchronizer factory from this configuration.
42    fn build_synchronizer(
43        &self,
44        context: &DataSourceBuildContext,
45    ) -> Result<Box<dyn SynchronizerFactory>, BuildError>;
46
47    /// Clones this configuration into a new box.
48    fn to_owned(&self) -> Box<dyn FDv2SynchronizerConfig>;
49}
50
51/// A configured FDv2 source usable as an initializer.
52pub trait FDv2InitializerConfig {
53    /// Builds an initializer factory from this configuration.
54    fn build_initializer(
55        &self,
56        context: &DataSourceBuildContext,
57    ) -> Result<Box<dyn InitializerFactory>, BuildError>;
58
59    /// Clones this configuration into a new box.
60    fn to_owned(&self) -> Box<dyn FDv2InitializerConfig>;
61}
62
63/// Builds the default HTTPS transport, or errors if no TLS feature is enabled.
64fn default_https_transport() -> Result<impl HttpTransport + 'static, BuildError> {
65    #[cfg(any(
66        feature = "hyper-rustls-native-roots",
67        feature = "hyper-rustls-webpki-roots",
68        feature = "native-tls"
69    ))]
70    {
71        HyperTransport::new_https().map_err(|e| {
72            BuildError::InvalidConfig(format!("failed to create default https transport: {e:?}"))
73        })
74    }
75    #[cfg(not(any(
76        feature = "hyper-rustls-native-roots",
77        feature = "hyper-rustls-webpki-roots",
78        feature = "native-tls"
79    )))]
80    {
81        Err::<HyperTransport, _>(BuildError::InvalidConfig(
82            "https connector required when hyper-rustls-native-roots, hyper-rustls-webpki-roots, or native-tls features are disabled".into(),
83        ))
84    }
85}
86
87/// Configures an FDv2 streaming source, which can only act as a synchronizer.
88#[derive(Clone)]
89pub struct FDv2StreamingBuilder<T: HttpTransport = HyperTransport> {
90    initial_reconnect_delay: Duration,
91    base_url: Option<String>,
92    transport: Option<T>,
93}
94
95impl<T: HttpTransport + Clone + Send + Sync + 'static> FDv2StreamingBuilder<T> {
96    /// Creates a builder with default values.
97    pub fn new() -> Self {
98        Self {
99            initial_reconnect_delay: DEFAULT_INITIAL_RECONNECT_DELAY,
100            base_url: None,
101            transport: None,
102        }
103    }
104
105    /// Sets the initial reconnect delay for the streaming connection.
106    ///
107    /// # Examples
108    /// ```
109    /// # use launchdarkly_server_sdk::FDv2StreamingBuilder;
110    /// # use launchdarkly_sdk_transport::HyperTransport;
111    /// # use std::time::Duration;
112    /// # fn main() {
113    ///     let mut source = FDv2StreamingBuilder::<HyperTransport>::new();
114    ///     source.initial_reconnect_delay(Duration::from_secs(10));
115    /// # }
116    /// ```
117    pub fn initial_reconnect_delay(&mut self, duration: Duration) -> &mut Self {
118        self.initial_reconnect_delay = duration;
119        self
120    }
121
122    /// Sets the streaming base URL, overriding the configured service endpoints.
123    ///
124    /// # Examples
125    /// ```
126    /// # use launchdarkly_server_sdk::FDv2StreamingBuilder;
127    /// # use launchdarkly_sdk_transport::HyperTransport;
128    /// # fn main() {
129    ///     let mut source = FDv2StreamingBuilder::<HyperTransport>::new();
130    ///     source.base_url("https://stream.example.com");
131    /// # }
132    /// ```
133    pub fn base_url(&mut self, url: &str) -> &mut Self {
134        self.base_url = Some(url.to_string());
135        self
136    }
137
138    /// Sets the transport to use, instead of the default HTTPS transport.
139    pub fn transport(&mut self, transport: T) -> &mut Self {
140        self.transport = Some(transport);
141        self
142    }
143}
144
145impl<T: HttpTransport + Clone + Send + Sync + 'static> FDv2SynchronizerConfig
146    for FDv2StreamingBuilder<T>
147{
148    fn build_synchronizer(
149        &self,
150        context: &DataSourceBuildContext,
151    ) -> Result<Box<dyn SynchronizerFactory>, BuildError> {
152        let base_url = self
153            .base_url
154            .clone()
155            .unwrap_or_else(|| context.endpoints.streaming_base_url().to_string());
156        let factory: Box<dyn SynchronizerFactory> = match &self.transport {
157            Some(transport) => Box::new(StreamingSynchronizerFactory::new(
158                transport.clone(),
159                base_url,
160                context.headers.clone(),
161                self.initial_reconnect_delay,
162            )),
163            None => Box::new(StreamingSynchronizerFactory::new(
164                default_https_transport()?,
165                base_url,
166                context.headers.clone(),
167                self.initial_reconnect_delay,
168            )),
169        };
170        Ok(factory)
171    }
172
173    fn to_owned(&self) -> Box<dyn FDv2SynchronizerConfig> {
174        Box::new(self.clone())
175    }
176}
177
178impl<T: HttpTransport + Clone + Send + Sync + 'static> Default for FDv2StreamingBuilder<T> {
179    fn default() -> Self {
180        Self::new()
181    }
182}
183
184/// Configures an FDv2 polling source, which can act as an initializer or a synchronizer.
185#[derive(Clone)]
186pub struct FDv2PollingBuilder<T: HttpTransport = HyperTransport> {
187    poll_interval: Duration,
188    base_url: Option<String>,
189    transport: Option<T>,
190}
191
192impl<T: HttpTransport + Clone + Send + Sync + 'static> FDv2PollingBuilder<T> {
193    /// Creates a builder with default values.
194    pub fn new() -> Self {
195        Self {
196            poll_interval: DEFAULT_POLL_INTERVAL,
197            base_url: None,
198            transport: None,
199        }
200    }
201
202    /// Sets the interval between polling requests, with an effective minimum of 30 seconds.
203    ///
204    /// # Examples
205    /// ```
206    /// # use launchdarkly_server_sdk::FDv2PollingBuilder;
207    /// # use launchdarkly_sdk_transport::HyperTransport;
208    /// # use std::time::Duration;
209    /// # fn main() {
210    ///     let mut source = FDv2PollingBuilder::<HyperTransport>::new();
211    ///     source.poll_interval(Duration::from_secs(60));
212    /// # }
213    /// ```
214    pub fn poll_interval(&mut self, poll_interval: Duration) -> &mut Self {
215        self.poll_interval = poll_interval;
216        self
217    }
218
219    /// Sets the polling base URL, overriding the configured service endpoints.
220    ///
221    /// # Examples
222    /// ```
223    /// # use launchdarkly_server_sdk::FDv2PollingBuilder;
224    /// # use launchdarkly_sdk_transport::HyperTransport;
225    /// # fn main() {
226    ///     let mut source = FDv2PollingBuilder::<HyperTransport>::new();
227    ///     source.base_url("https://polling.example.com");
228    /// # }
229    /// ```
230    pub fn base_url(&mut self, url: &str) -> &mut Self {
231        self.base_url = Some(url.to_string());
232        self
233    }
234
235    /// Sets the transport to use, instead of the default HTTPS transport.
236    pub fn transport(&mut self, transport: T) -> &mut Self {
237        self.transport = Some(transport);
238        self
239    }
240}
241
242impl<T: HttpTransport + Clone + Send + Sync + 'static> FDv2SynchronizerConfig
243    for FDv2PollingBuilder<T>
244{
245    fn build_synchronizer(
246        &self,
247        context: &DataSourceBuildContext,
248    ) -> Result<Box<dyn SynchronizerFactory>, BuildError> {
249        let base_url = self
250            .base_url
251            .clone()
252            .unwrap_or_else(|| context.endpoints.polling_base_url().to_string());
253        let factory: Box<dyn SynchronizerFactory> = match &self.transport {
254            Some(transport) => Box::new(PollingSynchronizerFactory::new(
255                transport.clone(),
256                base_url,
257                context.headers.clone(),
258                self.poll_interval,
259            )),
260            None => Box::new(PollingSynchronizerFactory::new(
261                default_https_transport()?,
262                base_url,
263                context.headers.clone(),
264                self.poll_interval,
265            )),
266        };
267        Ok(factory)
268    }
269
270    fn to_owned(&self) -> Box<dyn FDv2SynchronizerConfig> {
271        Box::new(self.clone())
272    }
273}
274
275impl<T: HttpTransport + Clone + Send + Sync + 'static> FDv2InitializerConfig
276    for FDv2PollingBuilder<T>
277{
278    fn build_initializer(
279        &self,
280        context: &DataSourceBuildContext,
281    ) -> Result<Box<dyn InitializerFactory>, BuildError> {
282        let base_url = self
283            .base_url
284            .clone()
285            .unwrap_or_else(|| context.endpoints.polling_base_url().to_string());
286        let factory: Box<dyn InitializerFactory> = match &self.transport {
287            Some(transport) => Box::new(PollingInitializerFactory::new(
288                transport.clone(),
289                base_url,
290                context.headers.clone(),
291            )),
292            None => Box::new(PollingInitializerFactory::new(
293                default_https_transport()?,
294                base_url,
295                context.headers.clone(),
296            )),
297        };
298        Ok(factory)
299    }
300
301    fn to_owned(&self) -> Box<dyn FDv2InitializerConfig> {
302        Box::new(self.clone())
303    }
304}
305
306impl<T: HttpTransport + Clone + Send + Sync + 'static> Default for FDv2PollingBuilder<T> {
307    fn default() -> Self {
308        Self::new()
309    }
310}
311
312/// Configures the FDv2 data system.
313///
314/// # Examples
315///
316/// Use the recommended data system.
317/// ```
318/// # use launchdarkly_server_sdk::{ConfigBuilder, DataSystemBuilder};
319/// # fn main() {
320///     ConfigBuilder::new("sdk-key").data_system(&DataSystemBuilder::default());
321/// # }
322/// ```
323///
324/// Assemble a custom data system from individual sources.
325/// ```
326/// # use launchdarkly_server_sdk::{
327/// #     ConfigBuilder, DataSystemBuilder, FDv2PollingBuilder, FDv2StreamingBuilder,
328/// # };
329/// # use launchdarkly_sdk_transport::HyperTransport;
330/// # fn main() {
331///     let mut data_system = DataSystemBuilder::custom();
332///     data_system.initializer(FDv2PollingBuilder::<HyperTransport>::new());
333///     data_system.synchronizer(FDv2StreamingBuilder::<HyperTransport>::new());
334///     ConfigBuilder::new("sdk-key").data_system(&data_system);
335/// # }
336/// ```
337pub struct DataSystemBuilder {
338    initializers: Vec<Box<dyn FDv2InitializerConfig>>,
339    synchronizers: Vec<Box<dyn FDv2SynchronizerConfig>>,
340    fdv1_fallback: Option<Box<dyn DataSourceFactory>>,
341}
342
343impl Clone for DataSystemBuilder {
344    fn clone(&self) -> Self {
345        Self {
346            initializers: self.initializers.iter().map(|c| (**c).to_owned()).collect(),
347            synchronizers: self
348                .synchronizers
349                .iter()
350                .map(|c| (**c).to_owned())
351                .collect(),
352            fdv1_fallback: self.fdv1_fallback.as_ref().map(|f| (**f).to_owned()),
353        }
354    }
355}
356
357impl DataSystemBuilder {
358    /// Creates an empty builder; the caller adds sources explicitly.
359    pub fn custom() -> Self {
360        Self {
361            initializers: Vec::new(),
362            synchronizers: Vec::new(),
363            fdv1_fallback: None,
364        }
365    }
366
367    /// Appends an initializer, called before the synchronizers in configuration order.
368    pub fn initializer(&mut self, source: impl FDv2InitializerConfig + 'static) -> &mut Self {
369        self.initializers.push(Box::new(source));
370        self
371    }
372
373    /// Appends a synchronizer, ordered after any already added.
374    pub fn synchronizer(&mut self, source: impl FDv2SynchronizerConfig + 'static) -> &mut Self {
375        self.synchronizers.push(Box::new(source));
376        self
377    }
378
379    /// Sets the FDv1 source used as a last-resort fallback.
380    pub fn fdv1_fallback(&mut self, factory: &dyn DataSourceFactory) -> &mut Self {
381        self.fdv1_fallback = Some(factory.to_owned());
382        self
383    }
384
385    /// Disables the FDv1 fallback.
386    pub fn disable_fdv1_fallback(&mut self) -> &mut Self {
387        self.fdv1_fallback = None;
388        self
389    }
390}
391
392impl Default for DataSystemBuilder {
393    /// The recommended data system setup.
394    fn default() -> Self {
395        let mut builder = Self::custom();
396        builder.initializer(FDv2PollingBuilder::<HyperTransport>::new());
397        builder.synchronizer(FDv2StreamingBuilder::<HyperTransport>::new());
398        builder.synchronizer(FDv2PollingBuilder::<HyperTransport>::new());
399        builder.fdv1_fallback(&PollingDataSourceBuilder::<HyperTransport>::new());
400        builder
401    }
402}
403
404/// Builds the internal FDv2 data system from a configured source set.
405pub(crate) trait DataSystemFactory {
406    fn build(
407        &self,
408        endpoints: &ServiceEndpoints,
409        sdk_key: &str,
410        tags: Option<&str>,
411        instance_id: &str,
412    ) -> Result<Arc<dyn DataSystem>, BuildError>;
413}
414
415impl DataSystemFactory for DataSystemBuilder {
416    fn build(
417        &self,
418        endpoints: &ServiceEndpoints,
419        sdk_key: &str,
420        tags: Option<&str>,
421        instance_id: &str,
422    ) -> Result<Arc<dyn DataSystem>, BuildError> {
423        let headers = RequestHeaders::new(sdk_key, tags, instance_id);
424        let context = DataSourceBuildContext {
425            endpoints,
426            headers: &headers,
427        };
428
429        let initializer_factories: Vec<Arc<dyn InitializerFactory>> = self
430            .initializers
431            .iter()
432            .map(|c| c.build_initializer(&context).map(Arc::from))
433            .collect::<Result<_, _>>()?;
434
435        let mut synchronizer_factories: Vec<Arc<dyn SynchronizerFactory>> = self
436            .synchronizers
437            .iter()
438            .map(|c| c.build_synchronizer(&context).map(Arc::from))
439            .collect::<Result<_, _>>()?;
440
441        // Build the FDv1 fallback source once and wrap it as a synchronizer; the
442        // adapter re-subscribes it whenever the fallback activates.
443        if let Some(fdv1_factory) = &self.fdv1_fallback {
444            let mut fdv1_factory = (**fdv1_factory).to_owned();
445            fdv1_factory.set_instance_id(instance_id.to_string());
446            let source = fdv1_factory
447                .build(endpoints, sdk_key, tags.map(|t| t.to_string()))
448                .map_err(|e| {
449                    BuildError::InvalidConfig(format!("failed to build FDv1 fallback source: {e}"))
450                })?;
451            let adapter = FDv1AdapterFactory::new(Box::new(move || source.clone()));
452            synchronizer_factories.push(Arc::new(adapter));
453        }
454
455        let system: Arc<dyn DataSystem> = Arc::new(FDv2DataSystem::new(
456            initializer_factories,
457            synchronizer_factories,
458            DEFAULT_FALLBACK_TIMEOUT,
459            DEFAULT_RECOVERY_TIMEOUT,
460        ));
461        Ok(system)
462    }
463}
464
465#[cfg(test)]
466mod tests {
467    use bytes::Bytes;
468    use launchdarkly_sdk_transport::{Request, ResponseFuture};
469
470    use super::*;
471
472    #[test]
473    fn custom_starts_empty() {
474        let builder = DataSystemBuilder::custom();
475
476        assert!(builder.initializers.is_empty());
477        assert!(builder.synchronizers.is_empty());
478        assert!(builder.fdv1_fallback.is_none());
479    }
480
481    #[test]
482    fn default_has_recommended_sources() {
483        let builder = DataSystemBuilder::default();
484
485        assert_eq!(builder.initializers.len(), 1);
486        assert_eq!(builder.synchronizers.len(), 2);
487        assert!(builder.fdv1_fallback.is_some());
488    }
489
490    #[test]
491    fn disable_fdv1_fallback_clears_it() {
492        let mut builder = DataSystemBuilder::default();
493        assert!(builder.fdv1_fallback.is_some());
494
495        builder.disable_fdv1_fallback();
496
497        assert!(builder.fdv1_fallback.is_none());
498    }
499
500    #[derive(Debug, Clone)]
501    struct TestTransport;
502
503    impl HttpTransport for TestTransport {
504        fn request(&self, _request: Request<Option<Bytes>>) -> ResponseFuture {
505            unreachable!();
506        }
507    }
508
509    #[test]
510    fn builders_build_factories_with_injected_transport() {
511        let endpoints = crate::ServiceEndpointsBuilder::new().build().unwrap();
512        let headers = RequestHeaders::new("sdk-key", None, "test-instance");
513        let context = DataSourceBuildContext {
514            endpoints: &endpoints,
515            headers: &headers,
516        };
517
518        // Each source builds a factory from a configured transport.
519        assert!(FDv2StreamingBuilder::<TestTransport>::new()
520            .transport(TestTransport)
521            .build_synchronizer(&context)
522            .is_ok());
523        assert!(FDv2PollingBuilder::<TestTransport>::new()
524            .transport(TestTransport)
525            .build_synchronizer(&context)
526            .is_ok());
527        assert!(FDv2PollingBuilder::<TestTransport>::new()
528            .transport(TestTransport)
529            .build_initializer(&context)
530            .is_ok());
531    }
532
533    // The default path builds a real HTTPS transport, which needs a TLS backend,
534    // so this only runs where one of those features is enabled.
535    #[test]
536    #[cfg(any(
537        feature = "hyper-rustls-native-roots",
538        feature = "hyper-rustls-webpki-roots",
539        feature = "native-tls"
540    ))]
541    fn builders_build_factories_with_default_transport() {
542        let endpoints = crate::ServiceEndpointsBuilder::new().build().unwrap();
543        let headers = RequestHeaders::new("sdk-key", None, "test-instance");
544        let context = DataSourceBuildContext {
545            endpoints: &endpoints,
546            headers: &headers,
547        };
548
549        // Each source builds a factory off the default HTTPS transport.
550        assert!(FDv2StreamingBuilder::<HyperTransport>::new()
551            .build_synchronizer(&context)
552            .is_ok());
553        assert!(FDv2PollingBuilder::<HyperTransport>::new()
554            .build_synchronizer(&context)
555            .is_ok());
556        assert!(FDv2PollingBuilder::<HyperTransport>::new()
557            .build_initializer(&context)
558            .is_ok());
559    }
560}
launchdarkly_server_sdk/data_system_builders.rs

launchdarkly_server_sdk/
data_system_builders.rs
