Merge #111

111: Adding the wait_for_pending_update method r=curquiza a=sanders41

Resolves  #77

Co-authored-by: Paul Sanders <psanders1@gmail.com>

Author: bors[bot]

Cargo.toml

@@ -14,9 +14,11 @@ log = "0.4"
 serde = { version = "1.0", features = ["derive"] }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
+futures = "0.3"
 isahc = { version = "1.0", features = ["http2", "text-decoding"], default_features = false }
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
+js-sys = "0.3.47"
 web-sys = { version = "0.3", features = ["RequestInit", "Headers", "Window", "Response", "console"] }
 wasm-bindgen = "0.2"
 wasm-bindgen-futures = "0.4"

src/progress.rs

@@ -3,6 +3,7 @@
 use crate::{errors::Error, indexes::Index, request::*};
 use serde::Deserialize;
 use std::collections::{BTreeMap, BTreeSet};
+use std::time::Duration;
 
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -36,6 +37,7 @@ impl<'a> Progress<'a> {
     /// let mut movies_index = client.get_or_create("movies").await.unwrap();
     /// let progress = movies_index.delete_all_documents().await.unwrap();
     /// let status = progress.get_status().await.unwrap();
+    /// # client.delete_index("movies").await.unwrap();
     /// # });
     /// ```
     pub async fn get_status(&self) -> Result<UpdateStatus, Error> {
@@ -50,6 +52,107 @@ impl<'a> Progress<'a> {
         )
         .await
     }
+
+    /// Wait until MeiliSearch processes an update, and get its status.
+    /// 
+    /// `interval` = The frequency at which the server should be polled. Default = 50ms
+    /// `timeout` = The maximum time to wait for processing to complete. Default = 5000ms
+    /// 
+    /// If the waited time exceeds `timeout` then `None` will be returned.
+    /// 
+    /// # Example
+    ///
+    /// ```
+    /// # use meilisearch_sdk::{client::*, document, indexes::*, progress::*};
+    /// # use serde::{Serialize, Deserialize};
+    /// #
+    /// # #[derive(Debug, Serialize, Deserialize, PartialEq)]
+    /// # struct Document {
+    /// #    id: usize,
+    /// #    value: String,
+    /// #    kind: String,
+    /// # }
+    /// # 
+    /// # impl document::Document for Document {
+    /// #    type UIDType = usize;
+    /// #
+    /// #    fn get_uid(&self) -> &Self::UIDType {
+    /// #        &self.id
+    /// #    }
+    /// # }
+    /// #
+    /// # futures::executor::block_on(async move {
+    /// let client = Client::new("http://localhost:7700", "masterKey");
+    /// let movies = client.create_index("movies_wait_for_pending", None).await.unwrap();
+    /// 
+    /// let progress = movies.add_documents(&[
+    ///     Document { id: 0, kind: "title".into(), value: "The Social Network".to_string() },
+    ///     Document { id: 1, kind: "title".into(), value: "Harry Potter and the Sorcerer's Stone".to_string() },
+    /// ], None).await.unwrap();
+    /// 
+    /// let status = progress.wait_for_pending_update(None, None).await.unwrap();
+    /// 
+    /// # client.delete_index("movies_wait_for_pending").await.unwrap();
+    /// assert!(matches!(status.unwrap(), UpdateStatus::Processed { .. }));
+    /// # });
+    /// ```
+    pub async fn wait_for_pending_update(
+        &self,
+        interval: Option<Duration>,
+        timeout: Option<Duration>,
+    ) -> Option<Result<UpdateStatus, Error>> {
+        let interval = interval.unwrap_or_else(|| Duration::from_millis(50));
+        let timeout = timeout.unwrap_or_else(|| Duration::from_millis(5000));
+
+        let mut elapsed_time = Duration::new(0, 0);
+        let mut status_result: Result<UpdateStatus, Error>;
+
+        while timeout > elapsed_time {
+            status_result = self.get_status().await;
+
+            match status_result {
+                Ok (status) => {
+                    match status {
+                        UpdateStatus::Failed { .. } | UpdateStatus::Processed { .. } => {
+                            return Some(self.get_status().await);
+                        },
+                        UpdateStatus::Enqueued { .. } => {
+                            elapsed_time += interval;
+                            async_sleep(interval).await;
+                        },
+                    }
+                },
+                Err (error) => return Some(Err(error)),
+            };
+        }
+
+        None
+    }
+}
+
+#[cfg(not(target_arch = "wasm32"))]
+pub(crate) async fn async_sleep(interval: Duration) {
+    let (sender, receiver) = futures::channel::oneshot::channel::<()>();
+    std::thread::spawn(move || {
+        std::thread::sleep(interval);
+        let _ = sender.send(());
+    });
+    let _ = receiver.await;
+}
+
+#[cfg(target_arch = "wasm32")]
+pub(crate) async fn async_sleep(interval: Duration) {
+    use wasm_bindgen_futures::JsFuture;
+
+    JsFuture::from(js_sys::Promise::new(&mut |yes, _| {
+        web_sys::window()
+            .unwrap()
+            .set_timeout_with_callback_and_timeout_and_arguments_0(
+                &yes,
+                interval.as_millis().try_into().unwrap(),
+            )
+            .unwrap();
+    })).await.unwrap();
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -133,3 +236,84 @@ pub enum UpdateStatus {
         content: ProcessedUpdateResult,
     },
 }
+
+#[cfg(test)]
+mod test {
+    use crate::{client::*, document, progress::*};
+    use serde::{Serialize, Deserialize};
+    use futures_await_test::async_test;
+    use std::time;
+
+    #[derive(Debug, Serialize, Deserialize, PartialEq)]
+    struct Document {
+       id: usize,
+       value: String,
+       kind: String,
+    }
+    
+    impl document::Document for Document {
+       type UIDType = usize;
+    
+       fn get_uid(&self) -> &Self::UIDType {
+           &self.id
+       }
+    }
+
+    #[async_test]
+    async fn test_wait_for_pending_updates_with_args() {
+        let client = Client::new("http://localhost:7700", "masterKey");
+        let movies = client.create_index("movies_wait_for_pending_args", None).await.unwrap();
+        let progress = movies.add_documents(&[
+            Document {
+                id: 0,
+                kind: "title".into(),
+                value: "The Social Network".to_string(),
+            },
+            Document {
+                id: 1,
+                kind: "title".into(),
+                value: "Harry Potter and the Sorcerer's Stone".to_string(),
+            },
+        ], None).await.unwrap();
+        let status = progress.wait_for_pending_update(
+            Some(Duration::from_millis(1)), Some(Duration::from_millis(6000))
+        ).await.unwrap();
+    
+        client.delete_index("movies_wait_for_pending_args").await.unwrap();
+        assert!(matches!(status.unwrap(), UpdateStatus::Processed { .. }));
+    }
+
+    #[async_test]
+    async fn test_wait_for_pending_updates_time_out() {
+        let client = Client::new("http://localhost:7700", "masterKey");
+        let movies = client.create_index("movies_wait_for_pending_timeout", None).await.unwrap();
+        let progress = movies.add_documents(&[
+            Document {
+                id: 0,
+                kind: "title".into(),
+                value: "The Social Network".to_string(),
+            },
+            Document {
+                id: 1,
+                kind: "title".into(),
+                value: "Harry Potter and the Sorcerer's Stone".to_string(),
+            },
+        ], None).await.unwrap();
+        let status = progress.wait_for_pending_update(
+            Some(Duration::from_millis(1)), Some(Duration::from_nanos(1))
+        ).await;
+    
+        client.delete_index("movies_wait_for_pending_timeout").await.unwrap();
+        assert_eq!(status.is_none(), true);
+    }
+
+    #[async_test]
+    async fn test_async_sleep() {
+        let sleep_duration = time::Duration::from_millis(10);
+        let now = time::Instant::now();
+
+        async_sleep(sleep_duration).await;
+
+        assert!(now.elapsed() >= sleep_duration);
+    }
+}