libsql: move push_with_retry to sync.rs

Author: LucioFranco

libsql/src/local/database.rs

@@ -443,7 +443,7 @@ impl Database {
             frame_no,
             frame_no + 1
         );
-        let max_frame_no = self
+        let max_frame_no = sync_ctx
             .push_with_retry(
                 uri,
                 &sync_ctx.auth_token,
@@ -454,43 +454,6 @@ impl Database {
         Ok(max_frame_no)
     }
 
-    #[cfg(feature = "sync")]
-    async fn push_with_retry(
-        &self,
-        uri: String,
-        auth_token: &Option<String>,
-        frame: Vec<u8>,
-        max_retries: usize,
-    ) -> Result<u32> {
-        let mut nr_retries = 0;
-        loop {
-            let client = reqwest::Client::new();
-            let mut builder = client.post(uri.to_owned());
-            match auth_token {
-                Some(ref auth_token) => {
-                    builder = builder
-                        .header("Authorization", format!("Bearer {}", auth_token.to_owned()));
-                }
-                None => {}
-            }
-            let res = builder.body(frame.to_vec()).send().await.unwrap();
-            if res.status().is_success() {
-                let resp = res.json::<serde_json::Value>().await.unwrap();
-                let max_frame_no = resp.get("max_frame_no").unwrap().as_u64().unwrap();
-                return Ok(max_frame_no as u32);
-            }
-            if nr_retries > max_retries {
-                return Err(crate::errors::Error::ConnectionFailed(format!(
-                    "Failed to push frame: {}",
-                    res.status()
-                )));
-            }
-            let delay = std::time::Duration::from_millis(100 * (1 << nr_retries));
-            tokio::time::sleep(delay).await;
-            nr_retries += 1;
-        }
-    }
-
     pub(crate) fn path(&self) -> &str {
         &self.db_path
     }

libsql/src/sync.rs

@@ -1,3 +1,5 @@
+use crate::Result;
+
 const DEFAULT_MAX_RETRIES: usize = 5;
 pub struct SyncContext {
     pub sync_url: String,
@@ -15,4 +17,59 @@ impl SyncContext {
             max_retries: DEFAULT_MAX_RETRIES,
         }
     }
+
+    pub(crate) async fn push_with_retry(
+        &self,
+        uri: String,
+        auth_token: &Option<String>,
+        frame: Vec<u8>,
+        max_retries: usize,
+    ) -> Result<u32> {
+        let mut nr_retries = 0;
+        loop {
+            // TODO(lucio): add custom connector + tls support here
+            let client = hyper::client::Client::builder().build_http::<hyper::Body>();
+
+            let mut req = http::Request::post(uri.clone());
+
+            match auth_token {
+                Some(ref auth_token) => {
+                    let auth_header =
+                        http::HeaderValue::try_from(format!("Bearer {}", auth_token.to_owned()))
+                            .unwrap();
+
+                    req.headers_mut()
+                        .expect("valid http request")
+                        .insert("Authorization", auth_header);
+                }
+                None => {}
+            }
+
+            // TODO(lucio): convert this to use bytes to make this clone cheap, it should be
+            // to possible use BytesMut when reading frames from the WAL and efficiently use Bytes
+            // from that.
+            let req = req.body(frame.clone().into()).expect("valid body");
+
+            let res = client.request(req).await.unwrap();
+
+            // TODO(lucio): only retry on server side errors
+            if res.status().is_success() {
+                let res_body = hyper::body::to_bytes(res.into_body()).await.unwrap();
+                let resp = serde_json::from_slice::<serde_json::Value>(&res_body[..]).unwrap();
+
+                let max_frame_no = resp.get("max_frame_no").unwrap().as_u64().unwrap();
+                return Ok(max_frame_no as u32);
+            }
+
+            if nr_retries > max_retries {
+                return Err(crate::errors::Error::ConnectionFailed(format!(
+                    "Failed to push frame: {}",
+                    res.status()
+                )));
+            }
+            let delay = std::time::Duration::from_millis(100 * (1 << nr_retries));
+            tokio::time::sleep(delay).await;
+            nr_retries += 1;
+        }
+    }
 }