kv & sql: timeout options

Author: barraguda

src/kv.rs

@@ -57,6 +57,7 @@ pub enum KvError {
 pub struct Kv {
     pub package_id: PackageId,
     pub db: String,
+    pub timeout: u64,
 }
 
 impl Kv {
@@ -69,7 +70,7 @@ impl Kv {
                 db: self.db.clone(),
                 action: KvAction::Get { key },
             })?)
-            .send_and_await_response(5)?;
+            .send_and_await_response(self.timeout)?;
 
         match res {
             Ok(Message::Response { body, .. }) => {
@@ -101,7 +102,7 @@ impl Kv {
                 action: KvAction::Set { key, tx_id },
             })?)
             .blob_bytes(value)
-            .send_and_await_response(5)?;
+            .send_and_await_response(self.timeout)?;
 
         match res {
             Ok(Message::Response { body, .. }) => {
@@ -126,7 +127,7 @@ impl Kv {
                 db: self.db.clone(),
                 action: KvAction::Delete { key, tx_id },
             })?)
-            .send_and_await_response(5)?;
+            .send_and_await_response(self.timeout)?;
 
         match res {
             Ok(Message::Response { body, .. }) => {
@@ -151,7 +152,7 @@ impl Kv {
                 db: self.db.clone(),
                 action: KvAction::BeginTx,
             })?)
-            .send_and_await_response(5)?;
+            .send_and_await_response(self.timeout)?;
 
         match res {
             Ok(Message::Response { body, .. }) => {
@@ -176,7 +177,7 @@ impl Kv {
                 db: self.db.clone(),
                 action: KvAction::Commit { tx_id },
             })?)
-            .send_and_await_response(5)?;
+            .send_and_await_response(self.timeout)?;
 
         match res {
             Ok(Message::Response { body, .. }) => {
@@ -194,15 +195,17 @@ impl Kv {
 }
 
 /// Opens or creates a kv db.
-pub fn open(package_id: PackageId, db: &str) -> anyhow::Result<Kv> {
+pub fn open(package_id: PackageId, db: &str, timeout: Option<u64>) -> anyhow::Result<Kv> {
+    let timeout = timeout.unwrap_or(5);
+
     let res = Request::new()
         .target(("our", "kv", "distro", "sys"))
         .body(serde_json::to_vec(&KvRequest {
             package_id: package_id.clone(),
             db: db.to_string(),
             action: KvAction::Open,
         })?)
-        .send_and_await_response(5)?;
+        .send_and_await_response(timeout)?;
 
     match res {
         Ok(Message::Response { body, .. }) => {
@@ -212,6 +215,7 @@ pub fn open(package_id: PackageId, db: &str) -> anyhow::Result<Kv> {
                 KvResponse::Ok => Ok(Kv {
                     package_id,
                     db: db.to_string(),
+                    timeout,
                 }),
                 KvResponse::Err { error } => Err(error.into()),
                 _ => Err(anyhow::anyhow!("kv: unexpected response {:?}", response)),
@@ -222,15 +226,17 @@ pub fn open(package_id: PackageId, db: &str) -> anyhow::Result<Kv> {
 }
 
 /// Removes and deletes a kv db.
-pub fn remove_db(package_id: PackageId, db: &str) -> anyhow::Result<()> {
+pub fn remove_db(package_id: PackageId, db: &str, timeout: Option<u64>) -> anyhow::Result<()> {
+    let timeout = timeout.unwrap_or(5);
+
     let res = Request::new()
         .target(("our", "kv", "distro", "sys"))
         .body(serde_json::to_vec(&KvRequest {
             package_id: package_id.clone(),
             db: db.to_string(),
             action: KvAction::RemoveDb,
         })?)
-        .send_and_await_response(5)?;
+        .send_and_await_response(timeout)?;
 
     match res {
         Ok(Message::Response { body, .. }) => {

src/sqlite.rs

@@ -79,6 +79,7 @@ pub enum SqliteError {
 pub struct Sqlite {
     pub package_id: PackageId,
     pub db: String,
+    pub timeout: u64,
 }
 
 impl Sqlite {
@@ -96,7 +97,7 @@ impl Sqlite {
                 action: SqliteAction::Read { query },
             })?)
             .blob_bytes(serde_json::to_vec(&params)?)
-            .send_and_await_response(5)?;
+            .send_and_await_response(self.timeout)?;
 
         match res {
             Ok(Message::Response { body, .. }) => {
@@ -141,7 +142,7 @@ impl Sqlite {
                 action: SqliteAction::Write { statement, tx_id },
             })?)
             .blob_bytes(serde_json::to_vec(&params)?)
-            .send_and_await_response(5)?;
+            .send_and_await_response(self.timeout)?;
 
         match res {
             Ok(Message::Response { body, .. }) => {
@@ -169,7 +170,7 @@ impl Sqlite {
                 db: self.db.clone(),
                 action: SqliteAction::BeginTx,
             })?)
-            .send_and_await_response(5)?;
+            .send_and_await_response(self.timeout)?;
 
         match res {
             Ok(Message::Response { body, .. }) => {
@@ -197,7 +198,7 @@ impl Sqlite {
                 db: self.db.clone(),
                 action: SqliteAction::Commit { tx_id },
             })?)
-            .send_and_await_response(5)?;
+            .send_and_await_response(self.timeout)?;
 
         match res {
             Ok(Message::Response { body, .. }) => {
@@ -218,15 +219,17 @@ impl Sqlite {
 }
 
 /// Open or create sqlite database.
-pub fn open(package_id: PackageId, db: &str) -> anyhow::Result<Sqlite> {
+pub fn open(package_id: PackageId, db: &str, timeout: Option<u64>) -> anyhow::Result<Sqlite> {
+    let timeout = timeout.unwrap_or(5);
+
     let res = Request::new()
         .target(("our", "sqlite", "distro", "sys"))
         .body(serde_json::to_vec(&SqliteRequest {
             package_id: package_id.clone(),
             db: db.to_string(),
             action: SqliteAction::Open,
         })?)
-        .send_and_await_response(5)?;
+        .send_and_await_response(timeout)?;
 
     match res {
         Ok(Message::Response { body, .. }) => {
@@ -236,6 +239,7 @@ pub fn open(package_id: PackageId, db: &str) -> anyhow::Result<Sqlite> {
                 SqliteResponse::Ok => Ok(Sqlite {
                     package_id,
                     db: db.to_string(),
+                    timeout,
                 }),
                 SqliteResponse::Err { error } => Err(error.into()),
                 _ => Err(anyhow::anyhow!(
@@ -249,15 +253,17 @@ pub fn open(package_id: PackageId, db: &str) -> anyhow::Result<Sqlite> {
 }
 
 /// Remove and delete sqlite database.
-pub fn remove_db(package_id: PackageId, db: &str) -> anyhow::Result<()> {
+pub fn remove_db(package_id: PackageId, db: &str, timeout: Option<u64>) -> anyhow::Result<()> {
+    let timeout = timeout.unwrap_or(5);
+
     let res = Request::new()
         .target(("our", "sqlite", "distro", "sys"))
         .body(serde_json::to_vec(&SqliteRequest {
             package_id: package_id.clone(),
             db: db.to_string(),
             action: SqliteAction::RemoveDb,
         })?)
-        .send_and_await_response(5)?;
+        .send_and_await_response(timeout)?;
 
     match res {
         Ok(Message::Response { body, .. }) => {