implement waiting for config and online (first draft)

rainerschoe · rainerschoe · commit 0557b0ce0414 · 2025-09-05T14:17:40.000+02:00
Signed-off-by: Travis &lt;rschoe@de.ibm.com&gt;
diff --git a/src/client/app_configuration_client.rs b/src/client/app_configuration_client.rs
@@ -67,6 +67,14 @@ pub trait ConfigurationProvider {
     /// For remote configurations, it returns whether it's connected to the
     /// remote or not
     fn is_online(&self) -> Result<bool>;
+
+    /// For remote configurations: Blocks until it's connected to the remote.
+    fn wait_until_online(&self);
+
+    /// Blocks until a configuration is available.
+    /// Note: This is different than wait_until_online, as configuration could be available
+    /// through alternate sources (Cache / Fallback)
+    fn wait_until_configuration_is_available(&self);
 }
 
 /// AppConfiguration client for browsing, and evaluating features and properties.
diff --git a/src/client/app_configuration_http.rs b/src/client/app_configuration_http.rs
@@ -92,6 +92,15 @@ impl<T: LiveConfiguration> ConfigurationProvider for AppConfigurationClientHttp<
     fn is_online(&self) -> Result<bool> {
         self.live_configuration.is_online()
     }
+
+    fn wait_until_online(&self) {
+        self.live_configuration.wait_until_online();
+    }
+
+    fn wait_until_configuration_is_available(&self) {
+        self.live_configuration
+            .wait_until_configuration_is_available();
+    }
 }
 
 #[cfg(test)]
diff --git a/src/client/app_configuration_ibm_cloud.rs b/src/client/app_configuration_ibm_cloud.rs
@@ -86,6 +86,14 @@ impl ConfigurationProvider for AppConfigurationClientIBMCloud {
     fn is_online(&self) -> Result<bool> {
         self.client.is_online()
     }
+
+    fn wait_until_online(&self) {
+        self.client.wait_until_online();
+    }
+
+    fn wait_until_configuration_is_available(&self) {
+        self.client.wait_until_configuration_is_available();
+    }
 }
 
 #[cfg(test)]
diff --git a/src/client/app_configuration_offline.rs b/src/client/app_configuration_offline.rs
@@ -15,6 +15,7 @@
 use crate::errors::Result;
 use crate::models::{Configuration, FeatureSnapshot, PropertySnapshot};
 use crate::ConfigurationProvider;
+use log::error;
 
 /// AppConfiguration client using a local file with a configuration snapshot
 #[derive(Debug)]
@@ -55,4 +56,14 @@ impl ConfigurationProvider for AppConfigurationOffline {
     fn is_online(&self) -> Result<bool> {
         Ok(false)
     }
+
+    fn wait_until_online(&self) {
+        error!("Waiting for AppConfigurationOffline to get online. This will never happen.");
+        std::thread::park(); // block forever
+    }
+
+    fn wait_until_configuration_is_available(&self) {
+        // No wait required:
+        // AppConfigurationOffline always has a configuration available.
+    }
 }
diff --git a/src/models/configuration.rs b/src/models/configuration.rs
@@ -22,6 +22,7 @@ use crate::ConfigurationDataError;
 use super::feature_snapshot::FeatureSnapshot;
 use super::property_snapshot::PropertySnapshot;
 use crate::ConfigurationProvider;
+use log::error;
 
 /// Represents all the configuration data needed for the client to perform
 /// feature/propery evaluation.
@@ -192,6 +193,15 @@ impl ConfigurationProvider for Configuration {
     fn is_online(&self) -> Result<bool> {
         Ok(false)
     }
+
+    fn wait_until_online(&self) {
+        error!("Waiting for Configuration to get online. This will never happen.");
+        std::thread::park(); // block forever
+    }
+
+    fn wait_until_configuration_is_available(&self) {
+        // No wait required
+    }
 }
 
 #[cfg(test)]
diff --git a/src/network/live_configuration/live_configuration.rs b/src/network/live_configuration/live_configuration.rs
@@ -12,13 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Condvar, Mutex};
 
 use super::current_mode::CurrentModeOfflineReason;
 use super::update_thread_worker::UpdateThreadWorker;
 use super::{CurrentMode, Error, OfflineMode, Result};
 use crate::models::Configuration;
 use crate::network::http_client::ServerClient;
+use crate::network::live_configuration::current_mode;
 use crate::utils::{ThreadHandle, ThreadStatus};
 use crate::{ConfigurationId, ConfigurationProvider};
 
@@ -37,10 +38,10 @@ pub trait LiveConfiguration: ConfigurationProvider {
 pub(crate) struct LiveConfigurationImpl {
     /// Configuration object that will be returned to consumers. This is also the object
     /// that the thread in the backend will be updating.
-    configuration: Arc<Mutex<Option<Configuration>>>,
+    configuration: Arc<(Mutex<Option<Configuration>>, Condvar)>,
 
     /// Current operation mode.
-    current_mode: Arc<Mutex<CurrentMode>>,
+    current_mode: Arc<(Mutex<CurrentMode>, Condvar)>,
 
     /// Handler to the internal thread that takes care of updating the [`LiveConfigurationImpl::configuration`].
     update_thread: ThreadHandle<Result<()>>,
@@ -57,10 +58,11 @@ impl LiveConfigurationImpl {
         server_client: T,
         configuration_id: ConfigurationId,
     ) -> Self {
-        let configuration = Arc::new(Mutex::new(None));
-        let current_mode = Arc::new(Mutex::new(CurrentMode::Offline(
-            CurrentModeOfflineReason::Initializing,
-        )));
+        let configuration = Arc::new((Mutex::new(None), Condvar::new()));
+        let current_mode = Arc::new((
+            Mutex::new(CurrentMode::Offline(CurrentModeOfflineReason::Initializing)),
+            Condvar::new(),
+        ));
 
         let worker = UpdateThreadWorker::new(
             server_client,
@@ -84,20 +86,25 @@ impl LiveConfigurationImpl {
     /// configured for this object.
     fn get_configuration(&self) -> Result<Configuration> {
         // TODO: Can we return a reference instead?
-        match &*self.current_mode.lock()? {
+        let (current_mode_mutex, _) = &*self.current_mode;
+        match &*current_mode_mutex.lock()? {
             CurrentMode::Online => {
-                match &*self.configuration.lock()? {
+                let (configuration_mutex, _) = &*self.configuration;
+                match &*configuration_mutex.lock()? {
                     // We store the configuration retrieved from the server into the Arc<Mutex> before switching the flag to Online
                     None => unreachable!(),
                     Some(configuration) => Ok(configuration.clone()),
                 }
             }
             CurrentMode::Offline(current_mode_offline_reason) => match &self.offline_mode {
                 OfflineMode::Fail => Err(Error::Offline(current_mode_offline_reason.clone())),
-                OfflineMode::Cache => match &*self.configuration.lock()? {
-                    None => Err(Error::ConfigurationNotYetAvailable),
-                    Some(configuration) => Ok(configuration.clone()),
-                },
+                OfflineMode::Cache => {
+                    let (configuration_mutex, _) = &*self.configuration;
+                    match &*configuration_mutex.lock()? {
+                        None => Err(Error::ConfigurationNotYetAvailable),
+                        Some(configuration) => Ok(configuration.clone()),
+                    }
+                }
                 OfflineMode::FallbackData(app_configuration_offline) => {
                     Ok(app_configuration_offline.config_snapshot.clone())
                 }
@@ -107,13 +114,16 @@ impl LiveConfigurationImpl {
                     "Thread finished with status: {:?}",
                     result
                 ))),
-                OfflineMode::Cache => match &*self.configuration.lock()? {
-                    None => Err(Error::UnrecoverableError(format!(
-                        "Initial configuration failed to retrieve: {:?}",
-                        result
-                    ))),
-                    Some(configuration) => Ok(configuration.clone()),
-                },
+                OfflineMode::Cache => {
+                    let (configuration_mutex, _) = &*self.configuration;
+                    match &*configuration_mutex.lock()? {
+                        None => Err(Error::UnrecoverableError(format!(
+                            "Initial configuration failed to retrieve: {:?}",
+                            result
+                        ))),
+                        Some(configuration) => Ok(configuration.clone()),
+                    }
+                }
                 OfflineMode::FallbackData(app_configuration_offline) => {
                     Ok(app_configuration_offline.config_snapshot.clone())
                 }
@@ -142,6 +152,24 @@ impl ConfigurationProvider for LiveConfigurationImpl {
     fn is_online(&self) -> crate::Result<bool> {
         Ok(self.get_current_mode()? == CurrentMode::Online)
     }
+
+    fn wait_until_configuration_is_available(&self) {
+        let (configuration_mutex, condition_variable) = &*self.configuration;
+        let configuration_guard = configuration_mutex.lock().unwrap();
+        condition_variable
+            .wait_while(configuration_guard, |configuration| configuration.is_none())
+            .unwrap();
+    }
+
+    fn wait_until_online(&self) {
+        let (current_mode_mutex, condition_variable) = &*self.current_mode;
+        let current_mode_guard = current_mode_mutex.lock().unwrap();
+        condition_variable
+            .wait_while(current_mode_guard, |current_mode| {
+                *current_mode == CurrentMode::Online
+            })
+            .unwrap();
+    }
 }
 
 impl LiveConfiguration for LiveConfigurationImpl {
@@ -150,7 +178,8 @@ impl LiveConfiguration for LiveConfigurationImpl {
     }
 
     fn get_current_mode(&self) -> Result<CurrentMode> {
-        Ok(self.current_mode.lock()?.clone())
+        let (current_mode_mutex, _) = &*self.current_mode;
+        Ok(current_mode_mutex.lock()?.clone())
     }
 }
 
diff --git a/src/network/live_configuration/update_thread_worker.rs b/src/network/live_configuration/update_thread_worker.rs
@@ -13,13 +13,14 @@
 // limitations under the License.
 
 use std::sync::mpsc::Receiver;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Condvar, Mutex};
 
 use super::current_mode::CurrentModeOfflineReason;
 use super::CurrentMode;
 use super::{Error, Result};
 use crate::models::Configuration;
 use crate::network::http_client::{ServerClient, WebsocketReader};
+use crate::network::live_configuration::current_mode;
 use crate::network::NetworkError;
 use crate::ConfigurationId;
 
@@ -28,16 +29,16 @@ pub(crate) const SERVER_HEARTBEAT: &str = "test message";
 pub(crate) struct UpdateThreadWorker<T: ServerClient> {
     server_client: T,
     configuration_id: ConfigurationId,
-    configuration: Arc<Mutex<Option<Configuration>>>,
-    current_mode: Arc<Mutex<CurrentMode>>,
+    configuration: Arc<(Mutex<Option<Configuration>>, Condvar)>,
+    current_mode: Arc<(Mutex<CurrentMode>, Condvar)>,
 }
 
 impl<T: ServerClient> UpdateThreadWorker<T> {
     pub(crate) fn new(
         server_client: T,
         configuration_id: ConfigurationId,
-        configuration: Arc<Mutex<Option<Configuration>>>,
-        current_mode: Arc<Mutex<CurrentMode>>,
+        configuration: Arc<(Mutex<Option<Configuration>>, Condvar)>,
+        current_mode: Arc<(Mutex<CurrentMode>, Condvar)>,
     ) -> Self {
         Self {
             server_client,
@@ -93,7 +94,10 @@ impl<T: ServerClient> UpdateThreadWorker<T> {
     /// [`UpdateThreadWorker::current_mode`] is set to [`CurrentMode::Defunct`].
     pub(crate) fn run(&self, thread_termination_receiver: Receiver<()>) -> Result<()> {
         let result = self.run_internal(thread_termination_receiver);
-        *self.current_mode.lock().unwrap() = CurrentMode::Defunct(result.clone());
+        let (current_mode_mutex, condition_variable) = &*self.current_mode;
+        let mut current_mode = current_mode_mutex.lock().unwrap();
+        *current_mode = CurrentMode::Defunct(result.clone());
+        condition_variable.notify_all();
         result
     }
 
@@ -103,14 +107,27 @@ impl<T: ServerClient> UpdateThreadWorker<T> {
     fn update_configuration_from_server_and_current_mode(&self) -> Result<()> {
         match self.server_client.get_configuration(&self.configuration_id) {
             Ok(config) => {
-                *self.configuration.lock()? = Some(config);
-                *self.current_mode.lock()? = CurrentMode::Online;
+                let (current_config_mutex, condition_variable) = &*self.configuration;
+                let mut current_config = current_config_mutex.lock()?;
+                *current_config = Some(config);
+                condition_variable.notify_all();
+
+                let (current_mode_mutex, condition_variable) = &*self.current_mode;
+                let mut current_mode = current_mode_mutex.lock()?;
+                *current_mode = CurrentMode::Online;
+                condition_variable.notify_all();
+
                 Ok(())
             }
             Err(e) => {
                 Self::recoverable_error(e)?;
-                *self.current_mode.lock()? =
+
+                let (current_mode_mutex, condition_variable) = &*self.current_mode;
+                let mut current_mode = current_mode_mutex.lock()?;
+                *current_mode =
                     CurrentMode::Offline(CurrentModeOfflineReason::FailedToGetNewConfiguration);
+                condition_variable.notify_all();
+
                 Ok(())
             }
         }
@@ -125,10 +142,11 @@ impl<T: ServerClient> UpdateThreadWorker<T> {
     /// there is any error receiving the messages. It's up to the caller to implement
     /// the recovery procedure for these scenarios.
     fn handle_websocket_message<WS: WebsocketReader>(&self, mut socket: WS) -> Result<Option<WS>> {
+        let (current_mode_mutex, condition_variable) = &*self.current_mode;
         match socket.read_msg() {
             Ok(msg) => match msg {
                 tungstenite::Message::Text(utf8_bytes) => {
-                    let current_mode_clone = self.current_mode.lock()?.clone();
+                    let current_mode_clone = current_mode_mutex.lock()?.clone();
                     match (utf8_bytes.as_str(), current_mode_clone) {
                         (SERVER_HEARTBEAT, CurrentMode::Offline(_)) => {
                             self.update_configuration_from_server_and_current_mode()?;
@@ -141,8 +159,9 @@ impl<T: ServerClient> UpdateThreadWorker<T> {
                     Ok(Some(socket))
                 }
                 tungstenite::Message::Close(_) => {
-                    *self.current_mode.lock()? =
+                    *current_mode_mutex.lock()? =
                         CurrentMode::Offline(CurrentModeOfflineReason::WebsocketClosed);
+                    condition_variable.notify_all();
                     Ok(None)
                 }
                 _ => {
@@ -151,8 +170,9 @@ impl<T: ServerClient> UpdateThreadWorker<T> {
                 }
             },
             Err(_) => {
-                *self.current_mode.lock()? =
+                *current_mode_mutex.lock()? =
                     CurrentMode::Offline(CurrentModeOfflineReason::WebsocketError);
+                condition_variable.notify_all();
                 Ok(None)
             }
         }

Original file line number	Diff line number	Diff line change
`@@ -86,6 +86,14 @@ impl ConfigurationProvider for AppConfigurationClientIBMCloud {`
`86`	`86`	`fn is_online(&self) -> Result<bool> {`
`87`	`87`	`self.client.is_online()`
`88`	`88`	`}`
	`89`	`+`
	`90`	`+ fn wait_until_online(&self) {`
	`91`	`+ self.client.wait_until_online();`
	`92`	`+ }`
	`93`	`+`
	`94`	`+ fn wait_until_configuration_is_available(&self) {`
	`95`	`+ self.client.wait_until_configuration_is_available();`
	`96`	`+ }`
`89`	`97`	`}`
`90`	`98`
`91`	`99`	`#[cfg(test)]`