launchdarkly_server_sdk/
client.rs

1use eval::Context;
2use parking_lot::RwLock;
3use std::io;
4use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
5use std::sync::{Arc, Mutex};
6use std::time::Duration;
7use tokio::runtime::Runtime;
8
9use launchdarkly_server_sdk_evaluation::{self as eval, Detail, FlagValue, PrerequisiteEvent};
10use serde::Serialize;
11use thiserror::Error;
12use tokio::sync::{broadcast, Semaphore};
13
14use super::config::Config;
15use super::data_source_builders::BuildError as DataSourceError;
16use super::data_system::{DataSystem, FDv1DataSystem};
17use super::data_system_builders::BuildError as DataSystemError;
18use super::evaluation::{FlagDetail, FlagDetailConfig};
19use super::stores::store::DataStore;
20use super::stores::store_builders::BuildError as DataStoreError;
21use crate::config::BuildError as ConfigBuildError;
22use crate::events::event::EventFactory;
23use crate::events::event::InputEvent;
24use crate::events::processor::EventProcessor;
25use crate::events::processor_builders::BuildError as EventProcessorError;
26use crate::{MigrationOpTracker, Stage};
27
28struct EventsScope {
29    disabled: bool,
30    event_factory: EventFactory,
31    prerequisite_event_recorder: Box<dyn eval::PrerequisiteEventRecorder + Send + Sync>,
32}
33
34struct PrerequisiteEventRecorder {
35    event_factory: EventFactory,
36    event_processor: Arc<dyn EventProcessor>,
37}
38
39impl eval::PrerequisiteEventRecorder for PrerequisiteEventRecorder {
40    fn record(&self, event: PrerequisiteEvent) {
41        let evt = self.event_factory.new_eval_event(
42            &event.prerequisite_flag.key,
43            event.context.clone(),
44            &event.prerequisite_flag,
45            event.prerequisite_result,
46            FlagValue::Json(serde_json::Value::Null),
47            Some(event.target_flag_key),
48        );
49
50        self.event_processor.send(evt);
51    }
52}
53
54/// Error type used to represent failures when building a [Client] instance.
55#[non_exhaustive]
56#[derive(Debug, Error)]
57pub enum BuildError {
58    /// Error used when a configuration setting is invalid. This typically indicates an invalid URL.
59    #[error("invalid client config: {0}")]
60    InvalidConfig(String),
61}
62
63impl From<DataSourceError> for BuildError {
64    fn from(error: DataSourceError) -> Self {
65        Self::InvalidConfig(error.to_string())
66    }
67}
68
69impl From<DataSystemError> for BuildError {
70    fn from(error: DataSystemError) -> Self {
71        Self::InvalidConfig(error.to_string())
72    }
73}
74
75impl From<DataStoreError> for BuildError {
76    fn from(error: DataStoreError) -> Self {
77        Self::InvalidConfig(error.to_string())
78    }
79}
80
81impl From<EventProcessorError> for BuildError {
82    fn from(error: EventProcessorError) -> Self {
83        Self::InvalidConfig(error.to_string())
84    }
85}
86
87impl From<ConfigBuildError> for BuildError {
88    fn from(error: ConfigBuildError) -> Self {
89        Self::InvalidConfig(error.to_string())
90    }
91}
92
93/// Error type used to represent failures when starting the [Client].
94#[non_exhaustive]
95#[derive(Debug, Error)]
96pub enum StartError {
97    /// Error used when spawning a background there fails.
98    #[error("couldn't spawn background thread for client: {0}")]
99    SpawnFailed(io::Error),
100}
101
102#[derive(PartialEq, Copy, Clone, Debug)]
103enum ClientInitState {
104    Initializing = 0,
105    Initialized = 1,
106    InitializationFailed = 2,
107}
108
109impl PartialEq<usize> for ClientInitState {
110    fn eq(&self, other: &usize) -> bool {
111        *self as usize == *other
112    }
113}
114
115impl From<usize> for ClientInitState {
116    fn from(val: usize) -> Self {
117        match val {
118            0 => ClientInitState::Initializing,
119            1 => ClientInitState::Initialized,
120            2 => ClientInitState::InitializationFailed,
121            _ => unreachable!(),
122        }
123    }
124}
125
126/// A client for the LaunchDarkly API.
127///
128/// In order to create a client instance, first create a config using [crate::ConfigBuilder].
129///
130/// # Examples
131///
132/// Creating a client, with default configuration.
133/// ```
134/// # use launchdarkly_server_sdk::{Client, ConfigBuilder, BuildError};
135/// # fn main() -> Result<(), BuildError> {
136///     let ld_client = Client::build(ConfigBuilder::new("sdk-key").build()?)?;
137/// #   Ok(())
138/// # }
139/// ```
140///
141/// Creating an instance which connects to a relay proxy.
142/// ```
143/// # use launchdarkly_server_sdk::{Client, ConfigBuilder, ServiceEndpointsBuilder, BuildError};
144/// # fn main() -> Result<(), BuildError> {
145///     let ld_client = Client::build(ConfigBuilder::new("sdk-key")
146///         .service_endpoints(ServiceEndpointsBuilder::new()
147///             .relay_proxy("http://my-relay-hostname:8080")
148///         ).build()?
149///     )?;
150/// #   Ok(())
151/// # }
152/// ```
153///
154/// Each builder type includes usage examples for the builder.
155pub struct Client {
156    event_processor: Arc<dyn EventProcessor>,
157    data_system: Arc<dyn DataSystem>,
158    data_store: Arc<RwLock<dyn DataStore>>,
159    events_default: EventsScope,
160    events_with_reasons: EventsScope,
161    init_notify: Arc<Semaphore>,
162    init_state: Arc<AtomicUsize>,
163    started: AtomicBool,
164    offline: bool,
165    daemon_mode: bool,
166    #[cfg_attr(
167        not(any(feature = "crypto-openssl", feature = "crypto-aws-lc-rs")),
168        allow(dead_code)
169    )]
170    sdk_key: String,
171    shutdown_broadcast: broadcast::Sender<()>,
172    runtime: RwLock<Option<Runtime>>,
173}
174
175impl Client {
176    /// Create a new instance of a [Client] based on the provided [Config] parameter.
177    pub fn build(config: Config) -> Result<Self, BuildError> {
178        if config.offline() {
179            info!("Started LaunchDarkly Client in offline mode");
180        } else if config.daemon_mode() {
181            info!("Started LaunchDarkly Client in daemon mode");
182        }
183
184        let tags = config.application_tag();
185        let instance_id = config.instance_id().to_string();
186
187        let endpoints = config.service_endpoints_builder().build()?;
188
189        let mut event_processor_builder = config.event_processor_builder().to_owned();
190        event_processor_builder.set_instance_id(instance_id.clone());
191        let event_processor =
192            event_processor_builder.build(&endpoints, config.sdk_key(), tags.clone())?;
193
194        let data_system: Arc<dyn DataSystem> = match config.data_system_builder() {
195            Some(data_system_builder) => data_system_builder.build(
196                &endpoints,
197                config.sdk_key(),
198                tags.as_deref(),
199                &instance_id,
200            )?,
201            None => {
202                let mut data_source_builder = config.data_source_builder().to_owned();
203                data_source_builder.set_instance_id(instance_id);
204                let data_source =
205                    data_source_builder.build(&endpoints, config.sdk_key(), tags.clone())?;
206                Arc::new(FDv1DataSystem::new(
207                    data_source,
208                    config.data_store_builder(),
209                )?)
210            }
211        };
212        let data_store = data_system.store();
213
214        let events_default = EventsScope {
215            disabled: config.offline(),
216            event_factory: EventFactory::new(false),
217            prerequisite_event_recorder: Box::new(PrerequisiteEventRecorder {
218                event_factory: EventFactory::new(false),
219                event_processor: event_processor.clone(),
220            }),
221        };
222
223        let events_with_reasons = EventsScope {
224            disabled: config.offline(),
225            event_factory: EventFactory::new(true),
226            prerequisite_event_recorder: Box::new(PrerequisiteEventRecorder {
227                event_factory: EventFactory::new(true),
228                event_processor: event_processor.clone(),
229            }),
230        };
231
232        let (shutdown_tx, _) = broadcast::channel(1);
233
234        Ok(Client {
235            event_processor,
236            data_system,
237            data_store,
238            events_default,
239            events_with_reasons,
240            init_notify: Arc::new(Semaphore::new(0)),
241            init_state: Arc::new(AtomicUsize::new(ClientInitState::Initializing as usize)),
242            started: AtomicBool::new(false),
243            offline: config.offline(),
244            daemon_mode: config.daemon_mode(),
245            sdk_key: config.sdk_key().into(),
246            shutdown_broadcast: shutdown_tx,
247            runtime: RwLock::new(None),
248        })
249    }
250
251    /// Starts a client in the current thread, which must have a default tokio runtime.
252    pub fn start_with_default_executor(&self) {
253        if self.started.load(Ordering::SeqCst) {
254            return;
255        }
256        self.started.store(true, Ordering::SeqCst);
257        self.start_with_default_executor_internal();
258    }
259
260    fn start_with_default_executor_internal(&self) {
261        // These clones are going to move into the closure, we
262        // do not want to move or reference `self`, because
263        // then lifetimes will get involved.
264        let notify = self.init_notify.clone();
265        let init_state = self.init_state.clone();
266
267        self.data_system.start(
268            Arc::new(move |success| {
269                init_state.store(
270                    (if success {
271                        ClientInitState::Initialized
272                    } else {
273                        ClientInitState::InitializationFailed
274                    }) as usize,
275                    Ordering::SeqCst,
276                );
277                notify.add_permits(1);
278            }),
279            self.shutdown_broadcast.subscribe(),
280        );
281    }
282
283    /// Creates a new tokio runtime and then starts the client. Tasks from the client will
284    /// be executed on created runtime.
285    /// If your application already has a tokio runtime, then you can use
286    /// [crate::Client::start_with_default_executor] and the client will dispatch tasks to
287    /// your existing runtime.
288    pub fn start_with_runtime(&self) -> Result<bool, StartError> {
289        if self.started.load(Ordering::SeqCst) {
290            return Ok(true);
291        }
292        self.started.store(true, Ordering::SeqCst);
293
294        let runtime = Runtime::new().map_err(StartError::SpawnFailed)?;
295        let _guard = runtime.enter();
296        self.runtime.write().replace(runtime);
297
298        self.start_with_default_executor_internal();
299
300        Ok(true)
301    }
302
303    /// This is an async method that will resolve once initialization is complete or the specified
304    /// timeout has occurred.
305    ///
306    /// If the timeout is triggered, this method will return `None`. Otherwise, the method will
307    /// return a boolean indicating whether or not the SDK has successfully initialized.
308    pub async fn wait_for_initialization(&self, timeout: Duration) -> Option<bool> {
309        if timeout > Duration::from_secs(60) {
310            warn!("wait_for_initialization was configured to block for up to {} seconds. We recommend blocking no longer than 60 seconds.", timeout.as_secs());
311        }
312
313        let initialized = tokio::time::timeout(timeout, self.initialized_async_internal()).await;
314        initialized.ok()
315    }
316
317    async fn initialized_async_internal(&self) -> bool {
318        if self.offline || self.daemon_mode {
319            return true;
320        }
321
322        // If the client is not initialized, then we need to wait for it to be initialized.
323        // Because we are using atomic types, and not a lock, then there is still the possibility
324        // that the value will change between the read and when we wait. We use a semaphore to wait,
325        // and we do not forget the permit, therefore if the permit has been added, then we will get
326        // it very quickly and reduce blocking.
327        if ClientInitState::Initialized != self.init_state.load(Ordering::SeqCst) {
328            let _permit = self.init_notify.acquire().await;
329        }
330        ClientInitState::Initialized == self.init_state.load(Ordering::SeqCst)
331    }
332
333    /// This function synchronously returns if the SDK is initialized.
334    /// In the case of unrecoverable errors in establishing a connection it is possible for the
335    /// SDK to never become initialized.
336    pub fn initialized(&self) -> bool {
337        self.offline
338            || self.daemon_mode
339            || ClientInitState::Initialized == self.init_state.load(Ordering::SeqCst)
340    }
341
342    /// Close shuts down the LaunchDarkly client. After calling this, the LaunchDarkly client
343    /// should no longer be used. The method will block until all pending analytics events (if any)
344    /// been sent.
345    pub fn close(&self) {
346        self.event_processor.close();
347
348        // If the system is in offline mode or daemon mode, no receiver will be listening to this
349        // broadcast channel, so sending on it would always result in an error.
350        if !self.offline && !self.daemon_mode {
351            if let Err(e) = self.shutdown_broadcast.send(()) {
352                error!("Failed to shutdown client appropriately: {e}");
353            }
354        }
355
356        // Potentially take the runtime we created when starting the client and do nothing with it
357        // so it drops, closing out all spawned tasks.
358        self.runtime.write().take();
359    }
360
361    /// Flush tells the client that all pending analytics events (if any) should be delivered as
362    /// soon as possible. Flushing is asynchronous, so this method will return before it is
363    /// complete. However, if you call [Client::close], events are guaranteed to be sent before
364    /// that method returns.
365    ///
366    /// For more information, see the Reference Guide:
367    /// <https://docs.launchdarkly.com/sdk/features/flush#rust>.
368    pub fn flush(&self) {
369        self.event_processor.flush();
370    }
371
372    /// Flush tells the client that all pending analytics events should be delivered as
373    /// soon as possible, and blocks until delivery is complete or the timeout expires.
374    ///
375    /// This method is particularly useful in short-lived execution environments like AWS Lambda
376    /// where you need to ensure events are sent before the function terminates.
377    ///
378    /// This method triggers a flush of events currently buffered and waits for that specific
379    /// flush to complete. Note that if periodic flushes or other flush operations are in-flight
380    /// when this is called, those may still be completing after this method returns.
381    ///
382    /// # Arguments
383    ///
384    /// * `timeout` - Maximum time to wait for flush to complete. Use `Duration::ZERO` to wait indefinitely.
385    ///
386    /// # Returns
387    ///
388    /// Returns `true` if flush completed successfully, `false` if timeout occurred.
389    ///
390    /// # Examples
391    ///
392    /// ```no_run
393    /// # use launchdarkly_server_sdk::{Client, ConfigBuilder};
394    /// # use std::time::Duration;
395    /// # async fn example() {
396    /// # let client = Client::build(ConfigBuilder::new("sdk-key").build().unwrap()).unwrap();
397    /// // Wait up to 5 seconds for flush to complete
398    /// let success = client.flush_blocking(Duration::from_secs(5)).await;
399    /// if !success {
400    ///     eprintln!("Warning: flush timed out");
401    /// }
402    /// # }
403    /// ```
404    ///
405    /// For more information, see the Reference Guide:
406    /// <https://docs.launchdarkly.com/sdk/features/flush#rust>.
407    pub async fn flush_blocking(&self, timeout: Duration) -> bool {
408        let event_processor = self.event_processor.clone();
409
410        let flush_future =
411            tokio::task::spawn_blocking(move || event_processor.flush_blocking(timeout));
412
413        if timeout == Duration::ZERO {
414            // Wait indefinitely
415            flush_future.await.unwrap_or(false)
416        } else {
417            // Apply timeout at async level too
418            match tokio::time::timeout(timeout, flush_future).await {
419                Ok(Ok(result)) => result,
420                Ok(Err(_)) => false, // spawn_blocking panicked
421                Err(_) => false,     // Timeout
422            }
423        }
424    }
425
426    /// Identify reports details about a context.
427    ///
428    /// For more information, see the Reference Guide:
429    /// <https://docs.launchdarkly.com/sdk/features/identify#rust>
430    pub fn identify(&self, context: Context) {
431        if self.events_default.disabled {
432            return;
433        }
434
435        self.send_internal(self.events_default.event_factory.new_identify(context));
436    }
437
438    /// Returns the value of a boolean feature flag for a given context.
439    ///
440    /// Returns `default` if there is an error, if the flag doesn't exist, or the feature is turned
441    /// off and has no off variation.
442    ///
443    /// For more information, see the Reference Guide:
444    /// <https://docs.launchdarkly.com/sdk/features/evaluating#rust>.
445    pub fn bool_variation(&self, context: &Context, flag_key: &str, default: bool) -> bool {
446        let val = self.variation(context, flag_key, default);
447        if let Some(b) = val.as_bool() {
448            b
449        } else {
450            warn!("bool_variation called for a non-bool flag {flag_key:?} (got {val:?})");
451            default
452        }
453    }
454
455    /// Returns the value of a string feature flag for a given context.
456    ///
457    /// Returns `default` if there is an error, if the flag doesn't exist, or the feature is turned
458    /// off and has no off variation.
459    ///
460    /// For more information, see the Reference Guide:
461    /// <https://docs.launchdarkly.com/sdk/features/evaluating#rust>.
462    pub fn str_variation(&self, context: &Context, flag_key: &str, default: String) -> String {
463        let val = self.variation(context, flag_key, default.clone());
464        if let Some(s) = val.as_string() {
465            s
466        } else {
467            warn!("str_variation called for a non-string flag {flag_key:?} (got {val:?})");
468            default
469        }
470    }
471
472    /// Returns the value of a float feature flag for a given context.
473    ///
474    /// Returns `default` if there is an error, if the flag doesn't exist, or the feature is turned
475    /// off and has no off variation.
476    ///
477    /// For more information, see the Reference Guide:
478    /// <https://docs.launchdarkly.com/sdk/features/evaluating#rust>.
479    pub fn float_variation(&self, context: &Context, flag_key: &str, default: f64) -> f64 {
480        let val = self.variation(context, flag_key, default);
481        if let Some(f) = val.as_float() {
482            f
483        } else {
484            warn!("float_variation called for a non-float flag {flag_key:?} (got {val:?})");
485            default
486        }
487    }
488
489    /// Returns the value of a integer feature flag for a given context.
490    ///
491    /// Returns `default` if there is an error, if the flag doesn't exist, or the feature is turned
492    /// off and has no off variation.
493    ///
494    /// For more information, see the Reference Guide:
495    /// <https://docs.launchdarkly.com/sdk/features/evaluating#rust>.
496    pub fn int_variation(&self, context: &Context, flag_key: &str, default: i64) -> i64 {
497        let val = self.variation(context, flag_key, default);
498        if let Some(f) = val.as_int() {
499            f
500        } else {
501            warn!("int_variation called for a non-int flag {flag_key:?} (got {val:?})");
502            default
503        }
504    }
505
506    /// Returns the value of a feature flag for the given context, allowing the value to be
507    /// of any JSON type.
508    ///
509    /// The value is returned as an [serde_json::Value].
510    ///
511    /// Returns `default` if there is an error, if the flag doesn't exist, or the feature is turned off.
512    ///
513    /// For more information, see the Reference Guide:
514    /// <https://docs.launchdarkly.com/sdk/features/evaluating#rust>.
515    pub fn json_variation(
516        &self,
517        context: &Context,
518        flag_key: &str,
519        default: serde_json::Value,
520    ) -> serde_json::Value {
521        self.variation(context, flag_key, default.clone())
522            .as_json()
523            .unwrap_or(default)
524    }
525
526    /// This method is the same as [Client::bool_variation], but also returns further information
527    /// about how the value was calculated. The "reason" data will also be included in analytics
528    /// events.
529    ///
530    /// For more information, see the Reference Guide:
531    /// <https://docs.launchdarkly.com/sdk/features/evaluation-reasons#rust>.
532    pub fn bool_variation_detail(
533        &self,
534        context: &Context,
535        flag_key: &str,
536        default: bool,
537    ) -> Detail<bool> {
538        self.variation_detail(context, flag_key, default).try_map(
539            |val| val.as_bool(),
540            default,
541            eval::Error::WrongType,
542        )
543    }
544
545    /// This method is the same as [Client::str_variation], but also returns further information
546    /// about how the value was calculated. The "reason" data will also be included in analytics
547    /// events.
548    ///
549    /// For more information, see the Reference Guide:
550    /// <https://docs.launchdarkly.com/sdk/features/evaluation-reasons#rust>.
551    pub fn str_variation_detail(
552        &self,
553        context: &Context,
554        flag_key: &str,
555        default: String,
556    ) -> Detail<String> {
557        self.variation_detail(context, flag_key, default.clone())
558            .try_map(|val| val.as_string(), default, eval::Error::WrongType)
559    }
560
561    /// This method is the same as [Client::float_variation], but also returns further information
562    /// about how the value was calculated. The "reason" data will also be included in analytics
563    /// events.
564    ///
565    /// For more information, see the Reference Guide:
566    /// <https://docs.launchdarkly.com/sdk/features/evaluation-reasons#rust>.
567    pub fn float_variation_detail(
568        &self,
569        context: &Context,
570        flag_key: &str,
571        default: f64,
572    ) -> Detail<f64> {
573        self.variation_detail(context, flag_key, default).try_map(
574            |val| val.as_float(),
575            default,
576            eval::Error::WrongType,
577        )
578    }
579
580    /// This method is the same as [Client::int_variation], but also returns further information
581    /// about how the value was calculated. The "reason" data will also be included in analytics
582    /// events.
583    ///
584    /// For more information, see the Reference Guide:
585    /// <https://docs.launchdarkly.com/sdk/features/evaluation-reasons#rust>.
586    pub fn int_variation_detail(
587        &self,
588        context: &Context,
589        flag_key: &str,
590        default: i64,
591    ) -> Detail<i64> {
592        self.variation_detail(context, flag_key, default).try_map(
593            |val| val.as_int(),
594            default,
595            eval::Error::WrongType,
596        )
597    }
598
599    /// This method is the same as [Client::json_variation], but also returns further information
600    /// about how the value was calculated. The "reason" data will also be included in analytics
601    /// events.
602    ///
603    /// For more information, see the Reference Guide:
604    /// <https://docs.launchdarkly.com/sdk/features/evaluation-reasons#rust>.
605    pub fn json_variation_detail(
606        &self,
607        context: &Context,
608        flag_key: &str,
609        default: serde_json::Value,
610    ) -> Detail<serde_json::Value> {
611        self.variation_detail(context, flag_key, default.clone())
612            .try_map(|val| val.as_json(), default, eval::Error::WrongType)
613    }
614
615    #[cfg(any(feature = "crypto-aws-lc-rs", feature = "crypto-openssl"))]
616    /// Generates the secure mode hash value for a context.
617    ///
618    /// For more information, see the Reference Guide:
619    /// <https://docs.launchdarkly.com/sdk/features/secure-mode#rust>.
620    pub fn secure_mode_hash(&self, context: &Context) -> Result<String, String> {
621        #[cfg(feature = "crypto-aws-lc-rs")]
622        {
623            let key =
624                aws_lc_rs::hmac::Key::new(aws_lc_rs::hmac::HMAC_SHA256, self.sdk_key.as_bytes());
625            let tag = aws_lc_rs::hmac::sign(&key, context.canonical_key().as_bytes());
626
627            Ok(data_encoding::HEXLOWER.encode(tag.as_ref()))
628        }
629        #[cfg(feature = "crypto-openssl")]
630        {
631            use openssl::hash::MessageDigest;
632            use openssl::pkey::PKey;
633            use openssl::sign::Signer;
634
635            let key = PKey::hmac(self.sdk_key.as_bytes())
636                .map_err(|e| format!("Failed to create HMAC key: {e}"))?;
637            let mut signer = Signer::new(MessageDigest::sha256(), &key)
638                .map_err(|e| format!("Failed to create signer: {e}"))?;
639            signer
640                .update(context.canonical_key().as_bytes())
641                .map_err(|e| format!("Failed to update signer: {e}"))?;
642            let hmac = signer
643                .sign_to_vec()
644                .map_err(|e| format!("Failed to sign: {e}"))?;
645
646            Ok(data_encoding::HEXLOWER.encode(&hmac))
647        }
648    }
649
650    /// Returns an object that encapsulates the state of all feature flags for a given context. This
651    /// includes the flag values, and also metadata that can be used on the front end.
652    ///
653    /// The most common use case for this method is to bootstrap a set of client-side feature flags
654    /// from a back-end service.
655    ///
656    /// You may pass any configuration of [FlagDetailConfig] to control what data is included.
657    ///
658    /// For more information, see the Reference Guide:
659    /// <https://docs.launchdarkly.com/sdk/features/all-flags#rust>
660    pub fn all_flags_detail(
661        &self,
662        context: &Context,
663        flag_state_config: FlagDetailConfig,
664    ) -> FlagDetail {
665        if self.offline {
666            warn!(
667                "all_flags_detail() called, but client is in offline mode. Returning empty state"
668            );
669            return FlagDetail::new(false);
670        }
671
672        if !self.initialized() {
673            warn!("all_flags_detail() called before client has finished initializing! Feature store unavailable - returning empty state");
674            return FlagDetail::new(false);
675        }
676
677        let data_store = self.data_store.read();
678
679        let mut flag_detail = FlagDetail::new(true);
680        flag_detail.populate(&*data_store, context, flag_state_config);
681
682        flag_detail
683    }
684
685    /// This method is the same as [Client::variation], but also returns further information about
686    /// how the value was calculated. The "reason" data will also be included in analytics events.
687    ///
688    /// For more information, see the Reference Guide:
689    /// <https://docs.launchdarkly.com/sdk/features/evaluation-reasons#rust>.
690    pub fn variation_detail<T: Into<FlagValue> + Clone>(
691        &self,
692        context: &Context,
693        flag_key: &str,
694        default: T,
695    ) -> Detail<FlagValue> {
696        let (detail, _) =
697            self.variation_internal(context, flag_key, default, &self.events_with_reasons);
698        detail
699    }
700
701    /// This is a generic function which returns the value of a feature flag for a given context.
702    ///
703    /// This method is an alternatively to the type specified methods (e.g.
704    /// [Client::bool_variation], [Client::int_variation], etc.).
705    ///
706    /// Returns `default` if there is an error, if the flag doesn't exist, or the feature is turned
707    /// off and has no off variation.
708    ///
709    /// For more information, see the Reference Guide:
710    /// <https://docs.launchdarkly.com/sdk/features/evaluating#rust>.
711    pub fn variation<T: Into<FlagValue> + Clone>(
712        &self,
713        context: &Context,
714        flag_key: &str,
715        default: T,
716    ) -> FlagValue {
717        let (detail, _) = self.variation_internal(context, flag_key, default, &self.events_default);
718        detail.value.unwrap()
719    }
720
721    /// This method returns the migration stage of the migration feature flag for the given
722    /// evaluation context.
723    ///
724    /// This method returns the default stage if there is an error or the flag does not exist.
725    pub fn migration_variation(
726        &self,
727        context: &Context,
728        flag_key: &str,
729        default_stage: Stage,
730    ) -> (Stage, Arc<Mutex<MigrationOpTracker>>) {
731        let (detail, flag) =
732            self.variation_internal(context, flag_key, default_stage, &self.events_default);
733
734        let migration_detail =
735            detail.try_map(|v| v.try_into().ok(), default_stage, eval::Error::WrongType);
736        let tracker = MigrationOpTracker::new(
737            flag_key.into(),
738            flag,
739            context.clone(),
740            migration_detail.clone(),
741            default_stage,
742        );
743
744        (
745            migration_detail.value.unwrap_or(default_stage),
746            Arc::new(Mutex::new(tracker)),
747        )
748    }
749
750    /// Reports that a context has performed an event.
751    ///
752    /// The `key` parameter is defined by the application and will be shown in analytics reports;
753    /// it normally corresponds to the event name of a metric that you have created through the
754    /// LaunchDarkly dashboard. If you want to associate additional data with this event, use
755    /// [Client::track_data] or [Client::track_metric].
756    ///
757    /// For more information, see the Reference Guide:
758    /// <https://docs.launchdarkly.com/sdk/features/events#rust>.
759    pub fn track_event(&self, context: Context, key: impl Into<String>) {
760        let _ = self.track(context, key, None, serde_json::Value::Null);
761    }
762
763    /// Reports that a context has performed an event, and associates it with custom data.
764    ///
765    /// The `key` parameter is defined by the application and will be shown in analytics reports;
766    /// it normally corresponds to the event name of a metric that you have created through the
767    /// LaunchDarkly dashboard.
768    ///
769    /// `data` parameter is any type that implements [Serialize]. If no such value is needed, use
770    /// [serde_json::Value::Null] (or call [Client::track_event] instead). To send a numeric value
771    /// for experimentation, use [Client::track_metric].
772    ///
773    /// For more information, see the Reference Guide:
774    /// <https://docs.launchdarkly.com/sdk/features/events#rust>.
775    pub fn track_data(
776        &self,
777        context: Context,
778        key: impl Into<String>,
779        data: impl Serialize,
780    ) -> serde_json::Result<()> {
781        self.track(context, key, None, data)
782    }
783
784    /// Reports that a context has performed an event, and associates it with a numeric value. This
785    /// value is used by the LaunchDarkly experimentation feature in numeric custom metrics, and
786    /// will also be returned as part of the custom event for Data Export.
787    ///
788    /// The `key` parameter is defined by the application and will be shown in analytics reports;
789    /// it normally corresponds to the event name of a metric that you have created through the
790    /// LaunchDarkly dashboard.
791    ///
792    /// For more information, see the Reference Guide:
793    /// <https://docs.launchdarkly.com/sdk/features/events#rust>.
794    pub fn track_metric(
795        &self,
796        context: Context,
797        key: impl Into<String>,
798        value: f64,
799        data: impl Serialize,
800    ) {
801        let _ = self.track(context, key, Some(value), data);
802    }
803
804    fn track(
805        &self,
806        context: Context,
807        key: impl Into<String>,
808        metric_value: Option<f64>,
809        data: impl Serialize,
810    ) -> serde_json::Result<()> {
811        if !self.events_default.disabled {
812            let event =
813                self.events_default
814                    .event_factory
815                    .new_custom(context, key, metric_value, data)?;
816
817            self.send_internal(event);
818        }
819
820        Ok(())
821    }
822
823    /// Tracks the results of a migrations operation. This event includes measurements which can be
824    /// used to enhance the observability of a migration within the LaunchDarkly UI.
825    ///
826    /// This event should be generated through [crate::MigrationOpTracker]. If you are using the
827    /// [crate::Migrator] to handle migrations, this event will be created and emitted
828    /// automatically.
829    pub fn track_migration_op(&self, tracker: Arc<Mutex<MigrationOpTracker>>) {
830        if self.events_default.disabled {
831            return;
832        }
833
834        match tracker.lock() {
835            Ok(tracker) => {
836                let event = tracker.build();
837                match event {
838                    Ok(event) => {
839                        self.send_internal(
840                            self.events_default.event_factory.new_migration_op(event),
841                        );
842                    }
843                    Err(e) => error!("Failed to build migration event, no event will be sent: {e}"),
844                }
845            }
846            Err(e) => error!("Failed to lock migration tracker, no event will be sent: {e}"),
847        }
848    }
849
850    fn variation_internal<T: Into<FlagValue> + Clone>(
851        &self,
852        context: &Context,
853        flag_key: &str,
854        default: T,
855        events_scope: &EventsScope,
856    ) -> (Detail<FlagValue>, Option<eval::Flag>) {
857        if self.offline {
858            return (
859                Detail::err_default(eval::Error::ClientNotReady, default.into()),
860                None,
861            );
862        }
863
864        let (flag, result) = match self.initialized() {
865            false => (
866                None,
867                Detail::err_default(eval::Error::ClientNotReady, default.clone().into()),
868            ),
869            true => {
870                let data_store = self.data_store.read();
871                match data_store.flag(flag_key) {
872                    Some(flag) => {
873                        let result = eval::evaluate(
874                            data_store.to_store(),
875                            &flag,
876                            context,
877                            Some(&*events_scope.prerequisite_event_recorder),
878                        )
879                        .map(|v| v.clone())
880                        .or(default.clone().into());
881
882                        (Some(flag), result)
883                    }
884                    None => (
885                        None,
886                        Detail::err_default(eval::Error::FlagNotFound, default.clone().into()),
887                    ),
888                }
889            }
890        };
891
892        if !events_scope.disabled {
893            let event = match &flag {
894                Some(f) => events_scope.event_factory.new_eval_event(
895                    flag_key,
896                    context.clone(),
897                    f,
898                    result.clone(),
899                    default.into(),
900                    None,
901                ),
902                None => events_scope.event_factory.new_unknown_flag_event(
903                    flag_key,
904                    context.clone(),
905                    result.clone(),
906                    default.into(),
907                ),
908            };
909            self.send_internal(event);
910        }
911
912        (result, flag)
913    }
914
915    fn send_internal(&self, event: InputEvent) {
916        self.event_processor.send(event);
917    }
918}
919
920#[cfg(test)]
921mod tests {
922    use assert_json_diff::assert_json_eq;
923    use crossbeam_channel::Receiver;
924    use eval::{ContextBuilder, MultiContextBuilder};
925    use futures::FutureExt;
926    use launchdarkly_server_sdk_evaluation::{Flag, Reason, Segment};
927    use maplit::hashmap;
928    use std::collections::HashMap;
929    use tokio::time::Instant;
930
931    use crate::data_source::MockDataSource;
932    use crate::data_source_builders::MockDataSourceBuilder;
933    use crate::evaluation::FlagFilter;
934    use crate::events::create_event_sender;
935    use crate::events::event::{OutputEvent, VariationKey};
936    use crate::events::processor_builders::EventProcessorBuilder;
937    use crate::stores::persistent_store::tests::InMemoryPersistentDataStore;
938    use crate::stores::store_types::{PatchTarget, StorageItem};
939    use crate::test_common::{
940        self, basic_flag, basic_flag_with_prereq, basic_flag_with_prereqs_and_visibility,
941        basic_flag_with_visibility, basic_int_flag, basic_migration_flag, basic_off_flag,
942    };
943    use crate::test_data::TestData;
944    use crate::{
945        AllData, ConfigBuilder, DataSystemBuilder, FlagBuilder, MigratorBuilder,
946        NullEventProcessorBuilder, Operation, Origin, PersistentDataStore,
947        PersistentDataStoreBuilder, PersistentDataStoreFactory, SerializedItem,
948    };
949    use test_case::test_case;
950
951    use super::*;
952
953    fn is_send_and_sync<T: Send + Sync>() {}
954
955    #[test]
956    fn ensure_client_is_send_and_sync() {
957        is_send_and_sync::<Client>()
958    }
959
960    #[tokio::test]
961    async fn client_asynchronously_initializes_within_timeout() {
962        let (client, _event_rx) = make_mocked_client_with_delay(1000, false, false);
963        client.start_with_default_executor();
964
965        let now = Instant::now();
966        let initialized = client
967            .wait_for_initialization(Duration::from_millis(1500))
968            .await;
969        let elapsed_time = now.elapsed();
970        // Give ourself a good margin for thread scheduling.
971        assert!(elapsed_time.as_millis() > 500);
972        assert_eq!(initialized, Some(true));
973    }
974
975    #[tokio::test]
976    async fn client_asynchronously_initializes_slower_than_timeout() {
977        let (client, _event_rx) = make_mocked_client_with_delay(2000, false, false);
978        client.start_with_default_executor();
979
980        let now = Instant::now();
981        let initialized = client
982            .wait_for_initialization(Duration::from_millis(500))
983            .await;
984        let elapsed_time = now.elapsed();
985        // Give ourself a good margin for thread scheduling.
986        assert!(elapsed_time.as_millis() < 750);
987        assert!(initialized.is_none());
988    }
989
990    #[tokio::test]
991    async fn client_initializes_immediately_in_offline_mode() {
992        let (client, _event_rx) = make_mocked_client_with_delay(1000, true, false);
993        client.start_with_default_executor();
994
995        assert!(client.initialized());
996
997        let now = Instant::now();
998        let initialized = client
999            .wait_for_initialization(Duration::from_millis(2000))
1000            .await;
1001        let elapsed_time = now.elapsed();
1002        assert_eq!(initialized, Some(true));
1003        assert!(elapsed_time.as_millis() < 500)
1004    }
1005
1006    #[tokio::test]
1007    async fn client_initializes_immediately_in_daemon_mode() {
1008        let (client, _event_rx) = make_mocked_client_with_delay(1000, false, true);
1009        client.start_with_default_executor();
1010
1011        assert!(client.initialized());
1012
1013        let now = Instant::now();
1014        let initialized = client
1015            .wait_for_initialization(Duration::from_millis(2000))
1016            .await;
1017        let elapsed_time = now.elapsed();
1018        assert_eq!(initialized, Some(true));
1019        assert!(elapsed_time.as_millis() < 500)
1020    }
1021
1022    #[test_case(basic_flag("myFlag"), false.into(), true.into())]
1023    #[test_case(basic_int_flag("myFlag"), 0.into(), test_common::FLOAT_TO_INT_MAX.into())]
1024    fn client_updates_changes_evaluation_results(
1025        flag: eval::Flag,
1026        default: FlagValue,
1027        expected: FlagValue,
1028    ) {
1029        let context = ContextBuilder::new("foo")
1030            .build()
1031            .expect("Failed to create context");
1032
1033        let (client, _event_rx) = make_mocked_client();
1034
1035        let result = client.variation_detail(&context, "myFlag", default.clone());
1036        assert_eq!(result.value.unwrap(), default);
1037
1038        client.start_with_default_executor();
1039        client
1040            .data_store
1041            .write()
1042            .upsert(
1043                &flag.key,
1044                PatchTarget::Flag(StorageItem::Item(flag.clone())),
1045            )
1046            .expect("patch should apply");
1047
1048        let result = client.variation_detail(&context, "myFlag", default);
1049        assert_eq!(result.value.unwrap(), expected);
1050        assert!(matches!(
1051            result.reason,
1052            Reason::Fallthrough {
1053                in_experiment: false
1054            }
1055        ));
1056    }
1057
1058    #[test]
1059    fn all_flags_detail_is_invalid_when_offline() {
1060        let (client, _event_rx) = make_mocked_offline_client();
1061        client.start_with_default_executor();
1062
1063        let context = ContextBuilder::new("bob")
1064            .build()
1065            .expect("Failed to create context");
1066
1067        let all_flags = client.all_flags_detail(&context, FlagDetailConfig::new());
1068        assert_json_eq!(all_flags, json!({"$valid": false, "$flagsState" : {}}));
1069    }
1070
1071    #[test]
1072    fn all_flags_detail_is_invalid_when_not_initialized() {
1073        let (client, _event_rx) = make_mocked_client();
1074
1075        let context = ContextBuilder::new("bob")
1076            .build()
1077            .expect("Failed to create context");
1078
1079        let all_flags = client.all_flags_detail(&context, FlagDetailConfig::new());
1080        assert_json_eq!(all_flags, json!({"$valid": false, "$flagsState" : {}}));
1081    }
1082
1083    #[tokio::test]
1084    async fn all_flags_detail_returns_flag_states() {
1085        let td = TestData::new();
1086        td.use_preconfigured_flag(basic_flag("myFlag1"));
1087        td.use_preconfigured_flag(basic_flag("myFlag2"));
1088        let (client, _event_rx) = make_client_with_test_data(&td);
1089        client.start_with_default_executor();
1090
1091        let context = ContextBuilder::new("bob")
1092            .build()
1093            .expect("Failed to create context");
1094
1095        let all_flags = client.all_flags_detail(&context, FlagDetailConfig::new());
1096
1097        client.close();
1098
1099        assert_json_eq!(
1100            all_flags,
1101            json!({
1102                "myFlag1": true,
1103                "myFlag2": true,
1104                "$flagsState": {
1105                    "myFlag1": {
1106                        "version": 1,
1107                        "variation": 1
1108                    },
1109                     "myFlag2": {
1110                        "version": 1,
1111                        "variation": 1
1112                    },
1113                },
1114                "$valid": true
1115            })
1116        );
1117    }
1118
1119    #[tokio::test]
1120    async fn all_flags_detail_returns_prerequisite_relations() {
1121        let td = TestData::new();
1122        td.use_preconfigured_flag(basic_flag("prereq1"));
1123        td.use_preconfigured_flag(basic_flag("prereq2"));
1124        td.use_preconfigured_flag(basic_flag_with_prereqs_and_visibility(
1125            "toplevel",
1126            &["prereq1", "prereq2"],
1127            false,
1128            false,
1129        ));
1130        let (client, _event_rx) = make_client_with_test_data(&td);
1131        client.start_with_default_executor();
1132
1133        let context = ContextBuilder::new("bob")
1134            .build()
1135            .expect("Failed to create context");
1136
1137        let all_flags = client.all_flags_detail(&context, FlagDetailConfig::new());
1138
1139        client.close();
1140
1141        assert_json_eq!(
1142            all_flags,
1143            json!({
1144                "prereq1": true,
1145                "prereq2": true,
1146                "toplevel": true,
1147                "$flagsState": {
1148                    "toplevel": {
1149                        "version": 1,
1150                        "variation": 1,
1151                        "prerequisites": ["prereq1", "prereq2"]
1152                    },
1153                    "prereq1": {
1154                        "version": 1,
1155                        "variation": 1
1156                    },
1157                     "prereq2": {
1158                        "version": 1,
1159                        "variation": 1
1160                    },
1161                },
1162                "$valid": true
1163            })
1164        );
1165    }
1166
1167    #[tokio::test]
1168    async fn all_flags_detail_returns_prerequisite_relations_when_not_visible_to_clients() {
1169        let td = TestData::new();
1170        td.use_preconfigured_flag(basic_flag_with_visibility("prereq1", false, false));
1171        td.use_preconfigured_flag(basic_flag_with_visibility("prereq2", false, false));
1172        td.use_preconfigured_flag(basic_flag_with_prereqs_and_visibility(
1173            "toplevel",
1174            &["prereq1", "prereq2"],
1175            true,
1176            false,
1177        ));
1178        let (client, _event_rx) = make_client_with_test_data(&td);
1179        client.start_with_default_executor();
1180
1181        let context = ContextBuilder::new("bob")
1182            .build()
1183            .expect("Failed to create context");
1184
1185        let mut config = FlagDetailConfig::new();
1186        config.flag_filter(FlagFilter::CLIENT);
1187
1188        let all_flags = client.all_flags_detail(&context, config);
1189
1190        client.close();
1191
1192        assert_json_eq!(
1193            all_flags,
1194            json!({
1195                "toplevel": true,
1196                "$flagsState": {
1197                    "toplevel": {
1198                        "version": 1,
1199                        "variation": 1,
1200                        "prerequisites": ["prereq1", "prereq2"]
1201                    },
1202                },
1203                "$valid": true
1204            })
1205        );
1206    }
1207
1208    #[tokio::test]
1209    async fn variation_tracks_events_correctly() {
1210        let td = TestData::new();
1211        td.use_preconfigured_flag(basic_flag("myFlag"));
1212        let (client, event_rx) = make_client_with_test_data(&td);
1213        client.start_with_default_executor();
1214
1215        let context = ContextBuilder::new("bob")
1216            .build()
1217            .expect("Failed to create context");
1218
1219        let flag_value = client.variation(&context, "myFlag", FlagValue::Bool(false));
1220
1221        assert!(flag_value.as_bool().unwrap());
1222        client.flush();
1223        client.close();
1224
1225        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
1226        assert_eq!(events.len(), 2);
1227        assert_eq!(events[0].kind(), "index");
1228        assert_eq!(events[1].kind(), "summary");
1229
1230        if let OutputEvent::Summary(event_summary) = events[1].clone() {
1231            let variation_key = VariationKey {
1232                version: Some(1),
1233                variation: Some(1),
1234            };
1235            let feature = event_summary.features.get("myFlag");
1236            assert!(feature.is_some());
1237
1238            let feature = feature.unwrap();
1239            assert!(feature.counters.contains_key(&variation_key));
1240        } else {
1241            panic!("Event should be a summary type");
1242        }
1243    }
1244
1245    #[test]
1246    fn variation_handles_offline_mode() {
1247        let (client, event_rx) = make_mocked_offline_client();
1248        client.start_with_default_executor();
1249
1250        let context = ContextBuilder::new("bob")
1251            .build()
1252            .expect("Failed to create context");
1253        let flag_value = client.variation(&context, "myFlag", FlagValue::Bool(false));
1254
1255        assert!(!flag_value.as_bool().unwrap());
1256        client.flush();
1257        client.close();
1258
1259        assert_eq!(event_rx.iter().count(), 0);
1260    }
1261
1262    #[test]
1263    fn variation_handles_unknown_flags() {
1264        let (client, event_rx) = make_mocked_client();
1265        client.start_with_default_executor();
1266        let context = ContextBuilder::new("bob")
1267            .build()
1268            .expect("Failed to create context");
1269
1270        let flag_value = client.variation(&context, "non-existent-flag", FlagValue::Bool(false));
1271
1272        assert!(!flag_value.as_bool().unwrap());
1273        client.flush();
1274        client.close();
1275
1276        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
1277        assert_eq!(events.len(), 2);
1278        assert_eq!(events[0].kind(), "index");
1279        assert_eq!(events[1].kind(), "summary");
1280
1281        if let OutputEvent::Summary(event_summary) = events[1].clone() {
1282            let variation_key = VariationKey {
1283                version: None,
1284                variation: None,
1285            };
1286
1287            let feature = event_summary.features.get("non-existent-flag");
1288            assert!(feature.is_some());
1289
1290            let feature = feature.unwrap();
1291            assert!(feature.counters.contains_key(&variation_key));
1292        } else {
1293            panic!("Event should be a summary type");
1294        }
1295    }
1296
1297    #[tokio::test]
1298    async fn variation_detail_handles_debug_events_correctly() {
1299        let td = TestData::new();
1300        let mut flag = basic_flag("myFlag");
1301        flag.debug_events_until_date = Some(64_060_606_800_000); // Jan. 1st, 4000
1302        td.use_preconfigured_flag(flag);
1303        let (client, event_rx) = make_client_with_test_data(&td);
1304        client.start_with_default_executor();
1305
1306        let context = ContextBuilder::new("bob")
1307            .build()
1308            .expect("Failed to create context");
1309
1310        let detail = client.variation_detail(&context, "myFlag", FlagValue::Bool(false));
1311
1312        assert!(detail.value.unwrap().as_bool().unwrap());
1313        assert!(matches!(
1314            detail.reason,
1315            Reason::Fallthrough {
1316                in_experiment: false
1317            }
1318        ));
1319        client.flush();
1320        client.close();
1321
1322        let events = event_rx.try_iter().collect::<Vec<OutputEvent>>();
1323        assert_eq!(events.len(), 3);
1324        assert_eq!(events[0].kind(), "index");
1325        assert_eq!(events[1].kind(), "debug");
1326        assert_eq!(events[2].kind(), "summary");
1327
1328        if let OutputEvent::Summary(event_summary) = events[2].clone() {
1329            let variation_key = VariationKey {
1330                version: Some(1),
1331                variation: Some(1),
1332            };
1333
1334            let feature = event_summary.features.get("myFlag");
1335            assert!(feature.is_some());
1336
1337            let feature = feature.unwrap();
1338            assert!(feature.counters.contains_key(&variation_key));
1339        } else {
1340            panic!("Event should be a summary type");
1341        }
1342    }
1343
1344    #[tokio::test]
1345    async fn variation_detail_tracks_events_correctly() {
1346        let td = TestData::new();
1347        td.use_preconfigured_flag(basic_flag("myFlag"));
1348        let (client, event_rx) = make_client_with_test_data(&td);
1349        client.start_with_default_executor();
1350
1351        let context = ContextBuilder::new("bob")
1352            .build()
1353            .expect("Failed to create context");
1354
1355        let detail = client.variation_detail(&context, "myFlag", FlagValue::Bool(false));
1356
1357        assert!(detail.value.unwrap().as_bool().unwrap());
1358        assert!(matches!(
1359            detail.reason,
1360            Reason::Fallthrough {
1361                in_experiment: false
1362            }
1363        ));
1364        client.flush();
1365        client.close();
1366
1367        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
1368        assert_eq!(events.len(), 2);
1369        assert_eq!(events[0].kind(), "index");
1370        assert_eq!(events[1].kind(), "summary");
1371
1372        if let OutputEvent::Summary(event_summary) = events[1].clone() {
1373            let variation_key = VariationKey {
1374                version: Some(1),
1375                variation: Some(1),
1376            };
1377
1378            let feature = event_summary.features.get("myFlag");
1379            assert!(feature.is_some());
1380
1381            let feature = feature.unwrap();
1382            assert!(feature.counters.contains_key(&variation_key));
1383        } else {
1384            panic!("Event should be a summary type");
1385        }
1386    }
1387
1388    #[test]
1389    fn variation_detail_handles_offline_mode() {
1390        let (client, event_rx) = make_mocked_offline_client();
1391        client.start_with_default_executor();
1392
1393        let context = ContextBuilder::new("bob")
1394            .build()
1395            .expect("Failed to create context");
1396
1397        let detail = client.variation_detail(&context, "myFlag", FlagValue::Bool(false));
1398
1399        assert!(!detail.value.unwrap().as_bool().unwrap());
1400        assert!(matches!(
1401            detail.reason,
1402            Reason::Error {
1403                error: eval::Error::ClientNotReady
1404            }
1405        ));
1406        client.flush();
1407        client.close();
1408
1409        assert_eq!(event_rx.iter().count(), 0);
1410    }
1411
1412    struct InMemoryPersistentDataStoreFactory {
1413        data: AllData<Flag, Segment>,
1414        initialized: bool,
1415    }
1416
1417    impl PersistentDataStoreFactory for InMemoryPersistentDataStoreFactory {
1418        fn create_persistent_data_store(
1419            &self,
1420        ) -> Result<Box<dyn PersistentDataStore + 'static>, std::io::Error> {
1421            let serialized_data =
1422                AllData::<SerializedItem, SerializedItem>::try_from(self.data.clone())?;
1423            Ok(Box::new(InMemoryPersistentDataStore {
1424                data: serialized_data,
1425                initialized: self.initialized,
1426            }))
1427        }
1428    }
1429
1430    #[test]
1431    fn variation_detail_handles_daemon_mode() {
1432        testing_logger::setup();
1433        let factory = InMemoryPersistentDataStoreFactory {
1434            data: AllData {
1435                flags: hashmap!["flag".into() => basic_flag("flag")],
1436                segments: HashMap::new(),
1437            },
1438            initialized: true,
1439        };
1440        let builder = PersistentDataStoreBuilder::new(Arc::new(factory));
1441
1442        let config = ConfigBuilder::new("sdk-key")
1443            .daemon_mode(true)
1444            .data_store(&builder)
1445            .event_processor(&NullEventProcessorBuilder::new())
1446            .build()
1447            .expect("config should build");
1448
1449        let client = Client::build(config).expect("Should be built.");
1450
1451        client.start_with_default_executor();
1452
1453        let context = ContextBuilder::new("bob")
1454            .build()
1455            .expect("Failed to create context");
1456
1457        let detail = client.variation_detail(&context, "flag", FlagValue::Bool(false));
1458
1459        assert!(detail.value.unwrap().as_bool().unwrap());
1460        assert!(matches!(
1461            detail.reason,
1462            Reason::Fallthrough {
1463                in_experiment: false
1464            }
1465        ));
1466        client.flush();
1467        client.close();
1468
1469        testing_logger::validate(|captured_logs| {
1470            assert_eq!(captured_logs.len(), 1);
1471            assert_eq!(
1472                captured_logs[0].body,
1473                "Started LaunchDarkly Client in daemon mode"
1474            );
1475        });
1476    }
1477
1478    #[test]
1479    fn daemon_mode_is_quiet_if_store_is_not_initialized() {
1480        testing_logger::setup();
1481
1482        let factory = InMemoryPersistentDataStoreFactory {
1483            data: AllData {
1484                flags: HashMap::new(),
1485                segments: HashMap::new(),
1486            },
1487            initialized: false,
1488        };
1489        let builder = PersistentDataStoreBuilder::new(Arc::new(factory));
1490
1491        let config = ConfigBuilder::new("sdk-key")
1492            .daemon_mode(true)
1493            .data_store(&builder)
1494            .event_processor(&NullEventProcessorBuilder::new())
1495            .build()
1496            .expect("config should build");
1497
1498        let client = Client::build(config).expect("Should be built.");
1499
1500        client.start_with_default_executor();
1501
1502        let context = ContextBuilder::new("bob")
1503            .build()
1504            .expect("Failed to create context");
1505
1506        client.variation_detail(&context, "flag", FlagValue::Bool(false));
1507
1508        testing_logger::validate(|captured_logs| {
1509            assert_eq!(captured_logs.len(), 1);
1510            assert_eq!(
1511                captured_logs[0].body,
1512                "Started LaunchDarkly Client in daemon mode"
1513            );
1514        });
1515    }
1516
1517    #[tokio::test]
1518    async fn variation_handles_off_flag_without_variation() {
1519        let td = TestData::new();
1520        td.use_preconfigured_flag(basic_off_flag("myFlag"));
1521        let (client, event_rx) = make_client_with_test_data(&td);
1522        client.start_with_default_executor();
1523
1524        let context = ContextBuilder::new("bob")
1525            .build()
1526            .expect("Failed to create context");
1527
1528        let result = client.variation(&context, "myFlag", FlagValue::Bool(false));
1529
1530        assert!(!result.as_bool().unwrap());
1531        client.flush();
1532        client.close();
1533
1534        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
1535        assert_eq!(events.len(), 2);
1536        assert_eq!(events[0].kind(), "index");
1537        assert_eq!(events[1].kind(), "summary");
1538
1539        if let OutputEvent::Summary(event_summary) = events[1].clone() {
1540            let variation_key = VariationKey {
1541                version: Some(1),
1542                variation: None,
1543            };
1544            let feature = event_summary.features.get("myFlag");
1545            assert!(feature.is_some());
1546
1547            let feature = feature.unwrap();
1548            assert!(feature.counters.contains_key(&variation_key));
1549        } else {
1550            panic!("Event should be a summary type");
1551        }
1552    }
1553
1554    #[tokio::test]
1555    async fn variation_detail_tracks_prereq_events_correctly() {
1556        let td = TestData::new();
1557        let mut prereq_flag = basic_flag("prereqFlag");
1558        prereq_flag.track_events = true;
1559        td.use_preconfigured_flag(prereq_flag);
1560
1561        let mut main_flag = basic_flag_with_prereq("myFlag", "prereqFlag");
1562        main_flag.track_events = true;
1563        td.use_preconfigured_flag(main_flag);
1564
1565        let (client, event_rx) = make_client_with_test_data(&td);
1566        client.start_with_default_executor();
1567
1568        let context = ContextBuilder::new("bob")
1569            .build()
1570            .expect("Failed to create context");
1571
1572        let detail = client.variation_detail(&context, "myFlag", FlagValue::Bool(false));
1573
1574        assert!(detail.value.unwrap().as_bool().unwrap());
1575        assert!(matches!(
1576            detail.reason,
1577            Reason::Fallthrough {
1578                in_experiment: false
1579            }
1580        ));
1581        client.flush();
1582        client.close();
1583
1584        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
1585        assert_eq!(events.len(), 4);
1586        assert_eq!(events[0].kind(), "index");
1587        assert_eq!(events[1].kind(), "feature");
1588        assert_eq!(events[2].kind(), "feature");
1589        assert_eq!(events[3].kind(), "summary");
1590
1591        if let OutputEvent::Summary(event_summary) = events[3].clone() {
1592            let variation_key = VariationKey {
1593                version: Some(1),
1594                variation: Some(1),
1595            };
1596            let feature = event_summary.features.get("myFlag");
1597            assert!(feature.is_some());
1598
1599            let feature = feature.unwrap();
1600            assert!(feature.counters.contains_key(&variation_key));
1601
1602            let variation_key = VariationKey {
1603                version: Some(1),
1604                variation: Some(1),
1605            };
1606            let feature = event_summary.features.get("prereqFlag");
1607            assert!(feature.is_some());
1608
1609            let feature = feature.unwrap();
1610            assert!(feature.counters.contains_key(&variation_key));
1611        }
1612    }
1613
1614    #[tokio::test]
1615    async fn variation_handles_failed_prereqs_correctly() {
1616        let td = TestData::new();
1617        let mut prereq_flag = basic_off_flag("prereqFlag");
1618        prereq_flag.track_events = true;
1619        td.use_preconfigured_flag(prereq_flag);
1620
1621        let mut main_flag = basic_flag_with_prereq("myFlag", "prereqFlag");
1622        main_flag.track_events = true;
1623        td.use_preconfigured_flag(main_flag);
1624
1625        let (client, event_rx) = make_client_with_test_data(&td);
1626        client.start_with_default_executor();
1627
1628        let context = ContextBuilder::new("bob")
1629            .build()
1630            .expect("Failed to create context");
1631
1632        let detail = client.variation(&context, "myFlag", FlagValue::Bool(false));
1633
1634        assert!(!detail.as_bool().unwrap());
1635        client.flush();
1636        client.close();
1637
1638        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
1639        assert_eq!(events.len(), 4);
1640        assert_eq!(events[0].kind(), "index");
1641        assert_eq!(events[1].kind(), "feature");
1642        assert_eq!(events[2].kind(), "feature");
1643        assert_eq!(events[3].kind(), "summary");
1644
1645        if let OutputEvent::Summary(event_summary) = events[3].clone() {
1646            let variation_key = VariationKey {
1647                version: Some(1),
1648                variation: Some(0),
1649            };
1650            let feature = event_summary.features.get("myFlag");
1651            assert!(feature.is_some());
1652
1653            let feature = feature.unwrap();
1654            assert!(feature.counters.contains_key(&variation_key));
1655
1656            let variation_key = VariationKey {
1657                version: Some(1),
1658                variation: None,
1659            };
1660            let feature = event_summary.features.get("prereqFlag");
1661            assert!(feature.is_some());
1662
1663            let feature = feature.unwrap();
1664            assert!(feature.counters.contains_key(&variation_key));
1665        }
1666    }
1667
1668    #[test]
1669    fn variation_detail_handles_flag_not_found() {
1670        let (client, event_rx) = make_mocked_client();
1671        client.start_with_default_executor();
1672
1673        let context = ContextBuilder::new("bob")
1674            .build()
1675            .expect("Failed to create context");
1676        let detail = client.variation_detail(&context, "non-existent-flag", FlagValue::Bool(false));
1677
1678        assert!(!detail.value.unwrap().as_bool().unwrap());
1679        assert!(matches!(
1680            detail.reason,
1681            Reason::Error {
1682                error: eval::Error::FlagNotFound
1683            }
1684        ));
1685        client.flush();
1686        client.close();
1687
1688        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
1689        assert_eq!(events.len(), 2);
1690        assert_eq!(events[0].kind(), "index");
1691        assert_eq!(events[1].kind(), "summary");
1692
1693        if let OutputEvent::Summary(event_summary) = events[1].clone() {
1694            let variation_key = VariationKey {
1695                version: None,
1696                variation: None,
1697            };
1698            let feature = event_summary.features.get("non-existent-flag");
1699            assert!(feature.is_some());
1700
1701            let feature = feature.unwrap();
1702            assert!(feature.counters.contains_key(&variation_key));
1703        } else {
1704            panic!("Event should be a summary type");
1705        }
1706    }
1707
1708    #[tokio::test]
1709    async fn variation_detail_handles_client_not_ready() {
1710        let (client, event_rx) = make_mocked_client_with_delay(u64::MAX, false, false);
1711        client.start_with_default_executor();
1712        let context = ContextBuilder::new("bob")
1713            .build()
1714            .expect("Failed to create context");
1715
1716        let detail = client.variation_detail(&context, "non-existent-flag", FlagValue::Bool(false));
1717
1718        assert!(!detail.value.unwrap().as_bool().unwrap());
1719        assert!(matches!(
1720            detail.reason,
1721            Reason::Error {
1722                error: eval::Error::ClientNotReady
1723            }
1724        ));
1725        client.flush();
1726        client.close();
1727
1728        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
1729        assert_eq!(events.len(), 2);
1730        assert_eq!(events[0].kind(), "index");
1731        assert_eq!(events[1].kind(), "summary");
1732
1733        if let OutputEvent::Summary(event_summary) = events[1].clone() {
1734            let variation_key = VariationKey {
1735                version: None,
1736                variation: None,
1737            };
1738            let feature = event_summary.features.get("non-existent-flag");
1739            assert!(feature.is_some());
1740
1741            let feature = feature.unwrap();
1742            assert!(feature.counters.contains_key(&variation_key));
1743        } else {
1744            panic!("Event should be a summary type");
1745        }
1746    }
1747
1748    #[test]
1749    fn identify_sends_identify_event() {
1750        let (client, event_rx) = make_mocked_client();
1751        client.start_with_default_executor();
1752
1753        let context = ContextBuilder::new("bob")
1754            .build()
1755            .expect("Failed to create context");
1756
1757        client.identify(context);
1758        client.flush();
1759        client.close();
1760
1761        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
1762        assert_eq!(events.len(), 1);
1763        assert_eq!(events[0].kind(), "identify");
1764    }
1765
1766    #[test]
1767    fn identify_sends_sends_nothing_in_offline_mode() {
1768        let (client, event_rx) = make_mocked_offline_client();
1769        client.start_with_default_executor();
1770
1771        let context = ContextBuilder::new("bob")
1772            .build()
1773            .expect("Failed to create context");
1774
1775        client.identify(context);
1776        client.flush();
1777        client.close();
1778
1779        assert_eq!(event_rx.iter().count(), 0);
1780    }
1781
1782    #[test]
1783    #[cfg(any(feature = "crypto-aws-lc-rs", feature = "crypto-openssl"))]
1784    fn secure_mode_hash() {
1785        let config = ConfigBuilder::new("secret")
1786            .offline(true)
1787            .build()
1788            .expect("config should build");
1789        let client = Client::build(config).expect("Should be built.");
1790        let context = ContextBuilder::new("Message")
1791            .build()
1792            .expect("Failed to create context");
1793
1794        assert_eq!(
1795            client
1796                .secure_mode_hash(&context)
1797                .expect("Hash should be computed"),
1798            "aa747c502a898200f9e4fa21bac68136f886a0e27aec70ba06daf2e2a5cb5597"
1799        );
1800    }
1801
1802    #[test]
1803    #[cfg(any(feature = "crypto-aws-lc-rs", feature = "crypto-openssl"))]
1804    fn secure_mode_hash_with_multi_kind() {
1805        let config = ConfigBuilder::new("secret")
1806            .offline(true)
1807            .build()
1808            .expect("config should build");
1809        let client = Client::build(config).expect("Should be built.");
1810
1811        let org = ContextBuilder::new("org-key|1")
1812            .kind("org")
1813            .build()
1814            .expect("Failed to create context");
1815        let user = ContextBuilder::new("user-key:2")
1816            .build()
1817            .expect("Failed to create context");
1818
1819        let context = MultiContextBuilder::new()
1820            .add_context(org)
1821            .add_context(user)
1822            .build()
1823            .expect("failed to build multi-context");
1824
1825        assert_eq!(
1826            client
1827                .secure_mode_hash(&context)
1828                .expect("Hash should be computed"),
1829            "5687e6383b920582ed50c2a96c98a115f1b6aad85a60579d761d9b8797415163"
1830        );
1831    }
1832
1833    #[derive(Serialize)]
1834    struct MyCustomData {
1835        pub answer: u32,
1836    }
1837
1838    #[test]
1839    fn track_sends_track_and_index_events() -> serde_json::Result<()> {
1840        let (client, event_rx) = make_mocked_client();
1841        client.start_with_default_executor();
1842
1843        let context = ContextBuilder::new("bob")
1844            .build()
1845            .expect("Failed to create context");
1846
1847        client.track_event(context.clone(), "event-with-null");
1848        client.track_data(context.clone(), "event-with-string", "string-data")?;
1849        client.track_data(context.clone(), "event-with-json", json!({"answer": 42}))?;
1850        client.track_data(
1851            context.clone(),
1852            "event-with-struct",
1853            MyCustomData { answer: 42 },
1854        )?;
1855        client.track_metric(context, "event-with-metric", 42.0, serde_json::Value::Null);
1856
1857        client.flush();
1858        client.close();
1859
1860        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
1861        assert_eq!(events.len(), 6);
1862
1863        let mut events_by_type: HashMap<&str, usize> = HashMap::new();
1864        for event in events {
1865            if let Some(count) = events_by_type.get_mut(event.kind()) {
1866                *count += 1;
1867            } else {
1868                events_by_type.insert(event.kind(), 1);
1869            }
1870        }
1871        assert!(matches!(events_by_type.get("index"), Some(1)));
1872        assert!(matches!(events_by_type.get("custom"), Some(5)));
1873
1874        Ok(())
1875    }
1876
1877    #[test]
1878    fn track_sends_nothing_in_offline_mode() -> serde_json::Result<()> {
1879        let (client, event_rx) = make_mocked_offline_client();
1880        client.start_with_default_executor();
1881
1882        let context = ContextBuilder::new("bob")
1883            .build()
1884            .expect("Failed to create context");
1885
1886        client.track_event(context.clone(), "event-with-null");
1887        client.track_data(context.clone(), "event-with-string", "string-data")?;
1888        client.track_data(context.clone(), "event-with-json", json!({"answer": 42}))?;
1889        client.track_data(
1890            context.clone(),
1891            "event-with-struct",
1892            MyCustomData { answer: 42 },
1893        )?;
1894        client.track_metric(context, "event-with-metric", 42.0, serde_json::Value::Null);
1895
1896        client.flush();
1897        client.close();
1898
1899        assert_eq!(event_rx.iter().count(), 0);
1900
1901        Ok(())
1902    }
1903
1904    #[test]
1905    fn migration_handles_flag_not_found() {
1906        let (client, _event_rx) = make_mocked_client();
1907        client.start_with_default_executor();
1908
1909        let context = ContextBuilder::new("bob")
1910            .build()
1911            .expect("Failed to create context");
1912
1913        let (stage, _tracker) =
1914            client.migration_variation(&context, "non-existent-flag-key", Stage::Off);
1915
1916        assert_eq!(stage, Stage::Off);
1917    }
1918
1919    #[tokio::test]
1920    async fn migration_uses_non_migration_flag() {
1921        let td = TestData::new();
1922        td.use_preconfigured_flag(basic_flag("boolean-flag"));
1923        let (client, _event_rx) = make_client_with_test_data(&td);
1924        client.start_with_default_executor();
1925
1926        let context = ContextBuilder::new("bob")
1927            .build()
1928            .expect("Failed to create context");
1929
1930        let (stage, _tracker) = client.migration_variation(&context, "boolean-flag", Stage::Off);
1931
1932        assert_eq!(stage, Stage::Off);
1933    }
1934
1935    #[test_case(Stage::Off)]
1936    #[test_case(Stage::DualWrite)]
1937    #[test_case(Stage::Shadow)]
1938    #[test_case(Stage::Live)]
1939    #[test_case(Stage::Rampdown)]
1940    #[test_case(Stage::Complete)]
1941    #[tokio::test]
1942    async fn migration_can_determine_correct_stage_from_flag(stage: Stage) {
1943        let td = TestData::new();
1944        td.use_preconfigured_flag(basic_migration_flag("stage-flag", stage));
1945        let (client, _event_rx) = make_client_with_test_data(&td);
1946        client.start_with_default_executor();
1947
1948        let context = ContextBuilder::new("bob")
1949            .build()
1950            .expect("Failed to create context");
1951
1952        let (evaluated_stage, _tracker) =
1953            client.migration_variation(&context, "stage-flag", Stage::Off);
1954
1955        assert_eq!(evaluated_stage, stage);
1956    }
1957
1958    #[tokio::test]
1959    async fn migration_tracks_invoked_correctly() {
1960        migration_tracks_invoked_correctly_driver(Stage::Off, Operation::Read, vec![Origin::Old])
1961            .await;
1962        migration_tracks_invoked_correctly_driver(
1963            Stage::DualWrite,
1964            Operation::Read,
1965            vec![Origin::Old],
1966        )
1967        .await;
1968        migration_tracks_invoked_correctly_driver(
1969            Stage::Shadow,
1970            Operation::Read,
1971            vec![Origin::Old, Origin::New],
1972        )
1973        .await;
1974        migration_tracks_invoked_correctly_driver(
1975            Stage::Live,
1976            Operation::Read,
1977            vec![Origin::Old, Origin::New],
1978        )
1979        .await;
1980        migration_tracks_invoked_correctly_driver(
1981            Stage::Rampdown,
1982            Operation::Read,
1983            vec![Origin::New],
1984        )
1985        .await;
1986        migration_tracks_invoked_correctly_driver(
1987            Stage::Complete,
1988            Operation::Read,
1989            vec![Origin::New],
1990        )
1991        .await;
1992        migration_tracks_invoked_correctly_driver(Stage::Off, Operation::Write, vec![Origin::Old])
1993            .await;
1994        migration_tracks_invoked_correctly_driver(
1995            Stage::DualWrite,
1996            Operation::Write,
1997            vec![Origin::Old, Origin::New],
1998        )
1999        .await;
2000        migration_tracks_invoked_correctly_driver(
2001            Stage::Shadow,
2002            Operation::Write,
2003            vec![Origin::Old, Origin::New],
2004        )
2005        .await;
2006        migration_tracks_invoked_correctly_driver(
2007            Stage::Live,
2008            Operation::Write,
2009            vec![Origin::Old, Origin::New],
2010        )
2011        .await;
2012        migration_tracks_invoked_correctly_driver(
2013            Stage::Rampdown,
2014            Operation::Write,
2015            vec![Origin::Old, Origin::New],
2016        )
2017        .await;
2018        migration_tracks_invoked_correctly_driver(
2019            Stage::Complete,
2020            Operation::Write,
2021            vec![Origin::New],
2022        )
2023        .await;
2024    }
2025
2026    async fn migration_tracks_invoked_correctly_driver(
2027        stage: Stage,
2028        operation: Operation,
2029        origins: Vec<Origin>,
2030    ) {
2031        let td = TestData::new();
2032        td.use_preconfigured_flag(basic_migration_flag("stage-flag", stage));
2033        let (client, event_rx) = make_client_with_test_data(&td);
2034        let client = Arc::new(client);
2035        client.start_with_default_executor();
2036
2037        let mut migrator = MigratorBuilder::new(client.clone())
2038            .read(
2039                |_| async move { Ok(serde_json::Value::Null) }.boxed(),
2040                |_| async move { Ok(serde_json::Value::Null) }.boxed(),
2041                Some(|_, _| true),
2042            )
2043            .write(
2044                |_| async move { Ok(serde_json::Value::Null) }.boxed(),
2045                |_| async move { Ok(serde_json::Value::Null) }.boxed(),
2046            )
2047            .build()
2048            .expect("migrator should build");
2049
2050        let context = ContextBuilder::new("bob")
2051            .build()
2052            .expect("Failed to create context");
2053
2054        if let Operation::Read = operation {
2055            migrator
2056                .read(
2057                    &context,
2058                    "stage-flag".into(),
2059                    Stage::Off,
2060                    serde_json::Value::Null,
2061                )
2062                .await;
2063        } else {
2064            migrator
2065                .write(
2066                    &context,
2067                    "stage-flag".into(),
2068                    Stage::Off,
2069                    serde_json::Value::Null,
2070                )
2071                .await;
2072        }
2073
2074        client.flush();
2075        client.close();
2076
2077        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
2078        assert_eq!(events.len(), 3);
2079        match &events[1] {
2080            OutputEvent::MigrationOp(event) => {
2081                assert!(event.invoked.len() == origins.len());
2082                assert!(event.invoked.iter().all(|i| origins.contains(i)));
2083            }
2084            _ => panic!("Expected migration event"),
2085        }
2086    }
2087
2088    #[tokio::test]
2089    async fn migration_tracks_latency() {
2090        migration_tracks_latency_driver(Stage::Off, Operation::Read, vec![Origin::Old]).await;
2091        migration_tracks_latency_driver(Stage::DualWrite, Operation::Read, vec![Origin::Old]).await;
2092        migration_tracks_latency_driver(
2093            Stage::Shadow,
2094            Operation::Read,
2095            vec![Origin::Old, Origin::New],
2096        )
2097        .await;
2098        migration_tracks_latency_driver(
2099            Stage::Live,
2100            Operation::Read,
2101            vec![Origin::Old, Origin::New],
2102        )
2103        .await;
2104        migration_tracks_latency_driver(Stage::Rampdown, Operation::Read, vec![Origin::New]).await;
2105        migration_tracks_latency_driver(Stage::Complete, Operation::Read, vec![Origin::New]).await;
2106        migration_tracks_latency_driver(Stage::Off, Operation::Write, vec![Origin::Old]).await;
2107        migration_tracks_latency_driver(
2108            Stage::DualWrite,
2109            Operation::Write,
2110            vec![Origin::Old, Origin::New],
2111        )
2112        .await;
2113        migration_tracks_latency_driver(
2114            Stage::Shadow,
2115            Operation::Write,
2116            vec![Origin::Old, Origin::New],
2117        )
2118        .await;
2119        migration_tracks_latency_driver(
2120            Stage::Live,
2121            Operation::Write,
2122            vec![Origin::Old, Origin::New],
2123        )
2124        .await;
2125        migration_tracks_latency_driver(
2126            Stage::Rampdown,
2127            Operation::Write,
2128            vec![Origin::Old, Origin::New],
2129        )
2130        .await;
2131        migration_tracks_latency_driver(Stage::Complete, Operation::Write, vec![Origin::New]).await;
2132    }
2133
2134    async fn migration_tracks_latency_driver(
2135        stage: Stage,
2136        operation: Operation,
2137        origins: Vec<Origin>,
2138    ) {
2139        let td = TestData::new();
2140        td.use_preconfigured_flag(basic_migration_flag("stage-flag", stage));
2141        let (client, event_rx) = make_client_with_test_data(&td);
2142        let client = Arc::new(client);
2143        client.start_with_default_executor();
2144
2145        let mut migrator = MigratorBuilder::new(client.clone())
2146            .track_latency(true)
2147            .read(
2148                |_| {
2149                    async move {
2150                        async_std::task::sleep(Duration::from_millis(100)).await;
2151                        Ok(serde_json::Value::Null)
2152                    }
2153                    .boxed()
2154                },
2155                |_| {
2156                    async move {
2157                        async_std::task::sleep(Duration::from_millis(100)).await;
2158                        Ok(serde_json::Value::Null)
2159                    }
2160                    .boxed()
2161                },
2162                Some(|_, _| true),
2163            )
2164            .write(
2165                |_| {
2166                    async move {
2167                        async_std::task::sleep(Duration::from_millis(100)).await;
2168                        Ok(serde_json::Value::Null)
2169                    }
2170                    .boxed()
2171                },
2172                |_| {
2173                    async move {
2174                        async_std::task::sleep(Duration::from_millis(100)).await;
2175                        Ok(serde_json::Value::Null)
2176                    }
2177                    .boxed()
2178                },
2179            )
2180            .build()
2181            .expect("migrator should build");
2182
2183        let context = ContextBuilder::new("bob")
2184            .build()
2185            .expect("Failed to create context");
2186
2187        if let Operation::Read = operation {
2188            migrator
2189                .read(
2190                    &context,
2191                    "stage-flag".into(),
2192                    Stage::Off,
2193                    serde_json::Value::Null,
2194                )
2195                .await;
2196        } else {
2197            migrator
2198                .write(
2199                    &context,
2200                    "stage-flag".into(),
2201                    Stage::Off,
2202                    serde_json::Value::Null,
2203                )
2204                .await;
2205        }
2206
2207        client.flush();
2208        client.close();
2209
2210        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
2211        assert_eq!(events.len(), 3);
2212        match &events[1] {
2213            OutputEvent::MigrationOp(event) => {
2214                assert!(event.latency.len() == origins.len());
2215                assert!(event
2216                    .latency
2217                    .values()
2218                    .all(|l| l > &Duration::from_millis(100)));
2219            }
2220            _ => panic!("Expected migration event"),
2221        }
2222    }
2223
2224    #[tokio::test]
2225    async fn migration_tracks_read_errors() {
2226        migration_tracks_read_errors_driver(Stage::Off, vec![Origin::Old]).await;
2227        migration_tracks_read_errors_driver(Stage::DualWrite, vec![Origin::Old]).await;
2228        migration_tracks_read_errors_driver(Stage::Shadow, vec![Origin::Old, Origin::New]).await;
2229        migration_tracks_read_errors_driver(Stage::Live, vec![Origin::Old, Origin::New]).await;
2230        migration_tracks_read_errors_driver(Stage::Rampdown, vec![Origin::New]).await;
2231        migration_tracks_read_errors_driver(Stage::Complete, vec![Origin::New]).await;
2232    }
2233
2234    async fn migration_tracks_read_errors_driver(stage: Stage, origins: Vec<Origin>) {
2235        let td = TestData::new();
2236        td.use_preconfigured_flag(basic_migration_flag("stage-flag", stage));
2237        let (client, event_rx) = make_client_with_test_data(&td);
2238        let client = Arc::new(client);
2239        client.start_with_default_executor();
2240
2241        let mut migrator = MigratorBuilder::new(client.clone())
2242            .track_latency(true)
2243            .read(
2244                |_| async move { Err("fail".into()) }.boxed(),
2245                |_| async move { Err("fail".into()) }.boxed(),
2246                Some(|_: &String, _: &String| true),
2247            )
2248            .write(
2249                |_| async move { Err("fail".into()) }.boxed(),
2250                |_| async move { Err("fail".into()) }.boxed(),
2251            )
2252            .build()
2253            .expect("migrator should build");
2254
2255        let context = ContextBuilder::new("bob")
2256            .build()
2257            .expect("Failed to create context");
2258
2259        migrator
2260            .read(
2261                &context,
2262                "stage-flag".into(),
2263                Stage::Off,
2264                serde_json::Value::Null,
2265            )
2266            .await;
2267        client.flush();
2268        client.close();
2269
2270        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
2271        assert_eq!(events.len(), 3);
2272        match &events[1] {
2273            OutputEvent::MigrationOp(event) => {
2274                assert!(event.errors.len() == origins.len());
2275                assert!(event.errors.iter().all(|i| origins.contains(i)));
2276            }
2277            _ => panic!("Expected migration event"),
2278        }
2279    }
2280
2281    #[tokio::test]
2282    async fn migration_tracks_authoritative_write_errors() {
2283        migration_tracks_authoritative_write_errors_driver(Stage::Off, vec![Origin::Old]).await;
2284        migration_tracks_authoritative_write_errors_driver(Stage::DualWrite, vec![Origin::Old])
2285            .await;
2286        migration_tracks_authoritative_write_errors_driver(Stage::Shadow, vec![Origin::Old]).await;
2287        migration_tracks_authoritative_write_errors_driver(Stage::Live, vec![Origin::New]).await;
2288        migration_tracks_authoritative_write_errors_driver(Stage::Rampdown, vec![Origin::New])
2289            .await;
2290        migration_tracks_authoritative_write_errors_driver(Stage::Complete, vec![Origin::New])
2291            .await;
2292    }
2293
2294    async fn migration_tracks_authoritative_write_errors_driver(
2295        stage: Stage,
2296        origins: Vec<Origin>,
2297    ) {
2298        let td = TestData::new();
2299        td.use_preconfigured_flag(basic_migration_flag("stage-flag", stage));
2300        let (client, event_rx) = make_client_with_test_data(&td);
2301        let client = Arc::new(client);
2302        client.start_with_default_executor();
2303
2304        let mut migrator = MigratorBuilder::new(client.clone())
2305            .track_latency(true)
2306            .read(
2307                |_| async move { Ok(serde_json::Value::Null) }.boxed(),
2308                |_| async move { Ok(serde_json::Value::Null) }.boxed(),
2309                None,
2310            )
2311            .write(
2312                |_| async move { Err("fail".into()) }.boxed(),
2313                |_| async move { Err("fail".into()) }.boxed(),
2314            )
2315            .build()
2316            .expect("migrator should build");
2317
2318        let context = ContextBuilder::new("bob")
2319            .build()
2320            .expect("Failed to create context");
2321
2322        migrator
2323            .write(
2324                &context,
2325                "stage-flag".into(),
2326                Stage::Off,
2327                serde_json::Value::Null,
2328            )
2329            .await;
2330
2331        client.flush();
2332        client.close();
2333
2334        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
2335        assert_eq!(events.len(), 3);
2336        match &events[1] {
2337            OutputEvent::MigrationOp(event) => {
2338                assert!(event.errors.len() == origins.len());
2339                assert!(event.errors.iter().all(|i| origins.contains(i)));
2340            }
2341            _ => panic!("Expected migration event"),
2342        }
2343    }
2344
2345    #[tokio::test]
2346    async fn migration_tracks_nonauthoritative_write_errors() {
2347        migration_tracks_nonauthoritative_write_errors_driver(
2348            Stage::DualWrite,
2349            false,
2350            true,
2351            vec![Origin::New],
2352        )
2353        .await;
2354        migration_tracks_nonauthoritative_write_errors_driver(
2355            Stage::Shadow,
2356            false,
2357            true,
2358            vec![Origin::New],
2359        )
2360        .await;
2361        migration_tracks_nonauthoritative_write_errors_driver(
2362            Stage::Live,
2363            true,
2364            false,
2365            vec![Origin::Old],
2366        )
2367        .await;
2368        migration_tracks_nonauthoritative_write_errors_driver(
2369            Stage::Rampdown,
2370            true,
2371            false,
2372            vec![Origin::Old],
2373        )
2374        .await;
2375    }
2376
2377    async fn migration_tracks_nonauthoritative_write_errors_driver(
2378        stage: Stage,
2379        fail_old: bool,
2380        fail_new: bool,
2381        origins: Vec<Origin>,
2382    ) {
2383        let td = TestData::new();
2384        td.use_preconfigured_flag(basic_migration_flag("stage-flag", stage));
2385        let (client, event_rx) = make_client_with_test_data(&td);
2386        let client = Arc::new(client);
2387        client.start_with_default_executor();
2388
2389        let mut migrator = MigratorBuilder::new(client.clone())
2390            .track_latency(true)
2391            .read(
2392                |_| async move { Ok(serde_json::Value::Null) }.boxed(),
2393                |_| async move { Ok(serde_json::Value::Null) }.boxed(),
2394                None,
2395            )
2396            .write(
2397                move |_| {
2398                    async move {
2399                        if fail_old {
2400                            Err("fail".into())
2401                        } else {
2402                            Ok(serde_json::Value::Null)
2403                        }
2404                    }
2405                    .boxed()
2406                },
2407                move |_| {
2408                    async move {
2409                        if fail_new {
2410                            Err("fail".into())
2411                        } else {
2412                            Ok(serde_json::Value::Null)
2413                        }
2414                    }
2415                    .boxed()
2416                },
2417            )
2418            .build()
2419            .expect("migrator should build");
2420
2421        let context = ContextBuilder::new("bob")
2422            .build()
2423            .expect("Failed to create context");
2424
2425        migrator
2426            .write(
2427                &context,
2428                "stage-flag".into(),
2429                Stage::Off,
2430                serde_json::Value::Null,
2431            )
2432            .await;
2433
2434        client.flush();
2435        client.close();
2436
2437        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
2438        assert_eq!(events.len(), 3);
2439        match &events[1] {
2440            OutputEvent::MigrationOp(event) => {
2441                assert!(event.errors.len() == origins.len());
2442                assert!(event.errors.iter().all(|i| origins.contains(i)));
2443            }
2444            _ => panic!("Expected migration event"),
2445        }
2446    }
2447
2448    #[tokio::test]
2449    async fn migration_tracks_consistency() {
2450        migration_tracks_consistency_driver(Stage::Shadow, "same", "same", true).await;
2451        migration_tracks_consistency_driver(Stage::Shadow, "same", "different", false).await;
2452        migration_tracks_consistency_driver(Stage::Live, "same", "same", true).await;
2453        migration_tracks_consistency_driver(Stage::Live, "same", "different", false).await;
2454    }
2455
2456    async fn migration_tracks_consistency_driver(
2457        stage: Stage,
2458        old_return: &'static str,
2459        new_return: &'static str,
2460        expected_consistency: bool,
2461    ) {
2462        let td = TestData::new();
2463        td.use_preconfigured_flag(basic_migration_flag("stage-flag", stage));
2464        let (client, event_rx) = make_client_with_test_data(&td);
2465        let client = Arc::new(client);
2466        client.start_with_default_executor();
2467
2468        let mut migrator = MigratorBuilder::new(client.clone())
2469            .track_latency(true)
2470            .read(
2471                |_| {
2472                    async move {
2473                        async_std::task::sleep(Duration::from_millis(100)).await;
2474                        Ok(serde_json::Value::String(old_return.to_string()))
2475                    }
2476                    .boxed()
2477                },
2478                |_| {
2479                    async move {
2480                        async_std::task::sleep(Duration::from_millis(100)).await;
2481                        Ok(serde_json::Value::String(new_return.to_string()))
2482                    }
2483                    .boxed()
2484                },
2485                Some(|lhs, rhs| lhs == rhs),
2486            )
2487            .write(
2488                |_| {
2489                    async move {
2490                        async_std::task::sleep(Duration::from_millis(100)).await;
2491                        Ok(serde_json::Value::Null)
2492                    }
2493                    .boxed()
2494                },
2495                |_| {
2496                    async move {
2497                        async_std::task::sleep(Duration::from_millis(100)).await;
2498                        Ok(serde_json::Value::Null)
2499                    }
2500                    .boxed()
2501                },
2502            )
2503            .build()
2504            .expect("migrator should build");
2505
2506        let context = ContextBuilder::new("bob")
2507            .build()
2508            .expect("Failed to create context");
2509
2510        migrator
2511            .read(
2512                &context,
2513                "stage-flag".into(),
2514                Stage::Off,
2515                serde_json::Value::Null,
2516            )
2517            .await;
2518
2519        client.flush();
2520        client.close();
2521
2522        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
2523        assert_eq!(events.len(), 3);
2524        match &events[1] {
2525            OutputEvent::MigrationOp(event) => {
2526                assert!(event.consistency_check == Some(expected_consistency))
2527            }
2528            _ => panic!("Expected migration event"),
2529        }
2530    }
2531
2532    #[tokio::test]
2533    async fn client_flush_blocking_completes_successfully() {
2534        let (client, event_rx) = make_mocked_client();
2535        client.start_with_default_executor();
2536        client.wait_for_initialization(Duration::from_secs(1)).await;
2537
2538        let context = ContextBuilder::new("user-key")
2539            .build()
2540            .expect("Failed to create context");
2541
2542        client.identify(context);
2543
2544        let result = client.flush_blocking(Duration::from_secs(5)).await;
2545        assert!(result, "flush_blocking should complete successfully");
2546
2547        client.close();
2548
2549        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
2550        assert!(!events.is_empty(), "Should have received identify event");
2551    }
2552
2553    #[tokio::test]
2554    async fn client_flush_blocking_with_zero_timeout() {
2555        let (client, event_rx) = make_mocked_client();
2556        client.start_with_default_executor();
2557        client.wait_for_initialization(Duration::from_secs(1)).await;
2558
2559        let context = ContextBuilder::new("user-key")
2560            .build()
2561            .expect("Failed to create context");
2562
2563        client.identify(context);
2564
2565        let result = client.flush_blocking(Duration::ZERO).await;
2566        assert!(
2567            result,
2568            "flush_blocking with zero timeout should complete successfully"
2569        );
2570
2571        client.close();
2572
2573        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
2574        assert!(!events.is_empty(), "Should have received identify event");
2575    }
2576
2577    #[tokio::test]
2578    async fn client_flush_blocking_with_no_events() {
2579        let (client, _event_rx) = make_mocked_client();
2580        client.start_with_default_executor();
2581        client.wait_for_initialization(Duration::from_secs(1)).await;
2582
2583        let result = client.flush_blocking(Duration::from_secs(1)).await;
2584        assert!(
2585            result,
2586            "flush_blocking with no events should complete immediately"
2587        );
2588
2589        client.close();
2590    }
2591
2592    #[tokio::test]
2593    async fn client_flush_blocking_multiple_concurrent_calls() {
2594        let (client, event_rx) = make_mocked_client();
2595        client.start_with_default_executor();
2596        client.wait_for_initialization(Duration::from_secs(1)).await;
2597
2598        let context = ContextBuilder::new("user-key")
2599            .build()
2600            .expect("Failed to create context");
2601
2602        client.identify(context);
2603
2604        // Make multiple concurrent flush_blocking calls
2605        let (result1, result2, result3) = tokio::join!(
2606            client.flush_blocking(Duration::from_secs(5)),
2607            client.flush_blocking(Duration::from_secs(5)),
2608            client.flush_blocking(Duration::from_secs(5)),
2609        );
2610
2611        assert!(result1, "First flush_blocking should succeed");
2612        assert!(result2, "Second flush_blocking should succeed");
2613        assert!(result3, "Third flush_blocking should succeed");
2614
2615        client.close();
2616
2617        let events = event_rx.iter().collect::<Vec<OutputEvent>>();
2618        assert!(!events.is_empty(), "Should have received identify event");
2619    }
2620
2621    fn make_mocked_client_with_delay(
2622        delay: u64,
2623        offline: bool,
2624        daemon_mode: bool,
2625    ) -> (Client, Receiver<OutputEvent>) {
2626        let updates = Arc::new(MockDataSource::new_with_init_delay(delay));
2627        let (event_sender, event_rx) = create_event_sender();
2628
2629        let config = ConfigBuilder::new("sdk-key")
2630            .offline(offline)
2631            .daemon_mode(daemon_mode)
2632            .data_source(MockDataSourceBuilder::new().data_source(updates))
2633            .event_processor(
2634                EventProcessorBuilder::<launchdarkly_sdk_transport::HyperTransport>::new()
2635                    .event_sender(Arc::new(event_sender)),
2636            )
2637            .build()
2638            .expect("config should build");
2639
2640        let client = Client::build(config).expect("Should be built.");
2641
2642        (client, event_rx)
2643    }
2644
2645    fn make_mocked_offline_client() -> (Client, Receiver<OutputEvent>) {
2646        make_mocked_client_with_delay(0, true, false)
2647    }
2648
2649    fn make_mocked_client() -> (Client, Receiver<OutputEvent>) {
2650        make_mocked_client_with_delay(0, false, false)
2651    }
2652
2653    fn make_client_with_test_data(td: &TestData) -> (Client, Receiver<OutputEvent>) {
2654        let (event_sender, event_rx) = create_event_sender();
2655        let config = ConfigBuilder::new("sdk-key")
2656            .data_source(td)
2657            .event_processor(
2658                EventProcessorBuilder::<launchdarkly_sdk_transport::HyperTransport>::new()
2659                    .event_sender(Arc::new(event_sender)),
2660            )
2661            .build()
2662            .expect("config should build");
2663        let client = Client::build(config).expect("Should be built.");
2664        (client, event_rx)
2665    }
2666
2667    fn make_client_with_test_data_fdv2(td: &TestData) -> Client {
2668        let mut data_system = DataSystemBuilder::custom();
2669        data_system.synchronizer(td.clone());
2670        let config = ConfigBuilder::new("sdk-key")
2671            .data_system(&data_system)
2672            .event_processor(&NullEventProcessorBuilder::new())
2673            .build()
2674            .expect("config should build");
2675        Client::build(config).expect("Should be built.")
2676    }
2677
2678    /// Polls `condition` until it holds, failing after a bounded wait. FDv2
2679    /// delivers test-data updates asynchronously, so callers wait for them.
2680    async fn wait_until(mut condition: impl FnMut() -> bool) {
2681        for _ in 0..100 {
2682            if condition() {
2683                return;
2684            }
2685            tokio::time::sleep(Duration::from_millis(10)).await;
2686        }
2687        panic!("condition was not met within the timeout");
2688    }
2689
2690    #[tokio::test]
2691    async fn fdv2_test_data_serves_and_updates_flags() {
2692        let td = TestData::new();
2693        td.update(FlagBuilder::new("my-flag").variation_for_all(true));
2694
2695        let client = make_client_with_test_data_fdv2(&td);
2696        client.start_with_default_executor();
2697        client.wait_for_initialization(Duration::from_secs(5)).await;
2698
2699        let context = ContextBuilder::new("user")
2700            .build()
2701            .expect("context should build");
2702
2703        // The initial full payload is served through the FDv2 data system.
2704        assert!(client.bool_variation(&context, "my-flag", false));
2705
2706        // A later update propagates asynchronously to the running client.
2707        td.update(FlagBuilder::new("my-flag").variation_for_all(false));
2708        wait_until(|| !client.bool_variation(&context, "my-flag", true)).await;
2709
2710        client.close();
2711    }
2712
2713    #[test]
2714    fn client_builds_successfully() {
2715        let config = ConfigBuilder::new("sdk-key")
2716            .offline(true)
2717            .build()
2718            .expect("config should build");
2719
2720        let client = Client::build(config).expect("client should build successfully");
2721
2722        assert!(
2723            !client.started.load(Ordering::SeqCst),
2724            "client should not be started yet"
2725        );
2726        assert!(client.offline, "client should be in offline mode");
2727        assert_eq!(client.sdk_key, "sdk-key", "sdk_key should match");
2728    }
2729}
launchdarkly_server_sdk/client.rs

launchdarkly_server_sdk/
client.rs
