Merge pull request #1825 from tursodatabase/lucio/sync-switch-hyper

libsql: Switch to hyper for local offline

Author: LucioFranco

Cargo.lock

@@ -42,7 +42,6 @@ fallible-iterator = { version = "0.3", optional = true }
 
 libsql_replication = { version = "0.6", path = "../libsql-replication", optional = true }
 async-stream = { version = "0.3.5", optional = true }
-reqwest = { version = "0.12.9", default-features = false, features = [ "rustls-tls", "json" ], optional = true }
 
 [dev-dependencies]
 criterion = { version = "0.5", features = ["html_reports", "async", "async_futures", "async_tokio"] }
@@ -105,7 +104,6 @@ sync = [
   "dep:bytes",
   "dep:tokio",
   "dep:futures",
-  "dep:reqwest",
   "dep:serde_json",
 ]
 hrana = [

libsql/Cargo.toml

@@ -476,21 +476,40 @@ impl Database {
     ) -> Result<u32> {
         let mut nr_retries = 0;
         loop {
-            let client = reqwest::Client::new();
-            let mut builder = client.post(uri.to_owned());
+            // 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) => {
-                    builder = builder
-                        .header("Authorization", format!("Bearer {}", auth_token.to_owned()));
+                    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 => {}
             }
-            let res = builder.body(frame.to_vec()).send().await.unwrap();
+
+            // 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 resp = res.json::<serde_json::Value>().await.unwrap();
+                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: {}",