Merge pull request #1829 from tursodatabase/lucio/local-offline-hyperv2

libsql: use hyper for offline writes
tursodatabase · Nov 20, 2024 · 49e6393 · 49e6393
2 parents 240eee8 + 66622f2
commit 49e6393
Show file tree

Hide file tree

Showing 6 changed files with 58 additions and 221 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/libsql/Cargo.toml b/libsql/Cargo.toml
@@ -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 = [

diff --git a/libsql/src/database.rs b/libsql/src/database.rs
@@ -663,7 +663,8 @@ impl Database {
 
 #[cfg(any(
     all(feature = "tls", feature = "replication"),
-    all(feature = "tls", feature = "remote")
+    all(feature = "tls", feature = "remote"),
+    all(feature = "tls", feature = "sync")
 ))]
 fn connector() -> Result<hyper_rustls::HttpsConnector<hyper::client::HttpConnector>> {
     let mut http = hyper::client::HttpConnector::new();
@@ -680,7 +681,8 @@ fn connector() -> Result<hyper_rustls::HttpsConnector<hyper::client::HttpConnect
 
 #[cfg(any(
     all(not(feature = "tls"), feature = "replication"),
-    all(not(feature = "tls"), feature = "remote")
+    all(not(feature = "tls"), feature = "remote"),
+    all(not(feature = "tls"), feature = "sync")
 ))]
 fn connector() -> Result<hyper::client::HttpConnector> {
     panic!("The `tls` feature is disabled, you must provide your own http connector");

diff --git a/libsql/src/database/builder.rs b/libsql/src/database/builder.rs
@@ -424,7 +424,17 @@ cfg_sync! {
 
             let path = path.to_str().ok_or(crate::Error::InvalidUTF8Path)?.to_owned();
 
+            let https = super::connector()?;
+            use tower::ServiceExt;
+
+            let svc = https
+                .map_err(|e| e.into())
+                .map_response(|s| Box::new(s) as Box<dyn crate::util::Socket>);
+
+            let connector = crate::util::ConnectorService::new(svc);
+
             let db = crate::local::Database::open_local_with_offline_writes(
+                connector,
                 path,
                 flags,
                 url,

diff --git a/libsql/src/local/database.rs b/libsql/src/local/database.rs
@@ -131,6 +131,7 @@ impl Database {
     #[cfg(feature = "sync")]
     #[doc(hidden)]
     pub async fn open_local_with_offline_writes(
+        connector: crate::util::ConnectorService,
         db_path: impl Into<String>,
         flags: OpenFlags,
         endpoint: String,
@@ -144,6 +145,7 @@ impl Database {
         };
         let mut db = Database::open(&db_path, flags)?;
         db.sync_ctx = Some(tokio::sync::Mutex::new(SyncContext::new(
+            connector,
             endpoint,
             Some(auth_token),
         )));
@@ -412,7 +414,7 @@ impl Database {
             // frames the server already knows about, we need to update the
             // frame number to the one returned by the server.
             let max_frame_no = sync_ctx
-                .push_one_frame(frame.to_vec(), generation, frame_no)
+                .push_one_frame(frame.freeze(), generation, frame_no)
                 .await?;
 
             if max_frame_no > frame_no {