diff options
| author | Alex <alex@adnab.me> | 2023-01-26 16:19:04 +0000 |
|---|---|---|
| committer | Alex <alex@adnab.me> | 2023-01-26 16:19:04 +0000 |
| commit | 246f7468cd18c8ef4f3c0c4c209853cd2500cc76 (patch) | |
| tree | 6e5f9ddd2159ba5396cc441a82b8240c4306323f /src/api/k2v/batch.rs | |
| parent | 611792ddcf86f0a728e22abaa6e172d3679d5ca6 (diff) | |
| parent | 1dff62564fdda392a97986dca55232f30a1f4234 (diff) | |
| download | garage-246f7468cd18c8ef4f3c0c4c209853cd2500cc76.tar.gz garage-246f7468cd18c8ef4f3c0c4c209853cd2500cc76.zip | |
Merge pull request 'K2V PollRange, version 2' (#471) from k2v-watch-range-2 into main
Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/471
Diffstat (limited to 'src/api/k2v/batch.rs')
| -rw-r--r-- | src/api/k2v/batch.rs | 85 |
1 files changed, 71 insertions, 14 deletions
diff --git a/src/api/k2v/batch.rs b/src/api/k2v/batch.rs index 82b4f7e3..26d678da 100644 --- a/src/api/k2v/batch.rs +++ b/src/api/k2v/batch.rs @@ -5,7 +5,6 @@ use hyper::{Body, Request, Response, StatusCode}; use serde::{Deserialize, Serialize}; use garage_util::data::*; -use garage_util::error::Error as GarageError; use garage_table::{EnumerationOrder, TableSchema}; @@ -26,11 +25,7 @@ pub async fn handle_insert_batch( let mut items2 = vec![]; for it in items { - let ct = it - .ct - .map(|s| CausalContext::parse(&s)) - .transpose() - .ok_or_bad_request("Invalid causality token")?; + let ct = it.ct.map(|s| CausalContext::parse_helper(&s)).transpose()?; let v = match it.v { Some(vs) => DvvsValue::Value( BASE64_STANDARD @@ -68,10 +63,7 @@ pub async fn handle_read_batch( resps.push(resp?); } - let resp_json = serde_json::to_string_pretty(&resps).map_err(GarageError::from)?; - Ok(Response::builder() - .status(StatusCode::OK) - .body(Body::from(resp_json))?) + Ok(json_ok_response(&resps)?) } async fn handle_read_batch_query( @@ -163,10 +155,7 @@ pub async fn handle_delete_batch( resps.push(resp?); } - let resp_json = serde_json::to_string_pretty(&resps).map_err(GarageError::from)?; - Ok(Response::builder() - .status(StatusCode::OK) - .body(Body::from(resp_json))?) + Ok(json_ok_response(&resps)?) } async fn handle_delete_batch_query( @@ -260,6 +249,53 @@ async fn handle_delete_batch_query( }) } +pub(crate) async fn handle_poll_range( + garage: Arc<Garage>, + bucket_id: Uuid, + partition_key: &str, + req: Request<Body>, +) -> Result<Response<Body>, Error> { + use garage_model::k2v::sub::PollRange; + + let query = parse_json_body::<PollRangeQuery>(req).await?; + + let timeout_msec = query.timeout.unwrap_or(300).clamp(1, 600) * 1000; + + let resp = garage + .k2v + .rpc + .poll_range( + PollRange { + partition: K2VItemPartition { + bucket_id, + partition_key: partition_key.to_string(), + }, + start: query.start, + end: query.end, + prefix: query.prefix, + }, + query.seen_marker, + timeout_msec, + ) + .await?; + + if let Some((items, seen_marker)) = resp { + let resp = PollRangeResponse { + items: items + .into_iter() + .map(|(_k, i)| ReadBatchResponseItem::from(i)) + .collect::<Vec<_>>(), + seen_marker, + }; + + Ok(json_ok_response(&resp)?) + } else { + Ok(Response::builder() + .status(StatusCode::NOT_MODIFIED) + .body(Body::empty())?) + } +} + #[derive(Deserialize)] struct InsertBatchItem { pk: String, @@ -364,3 +400,24 @@ struct DeleteBatchResponse { #[serde(rename = "deletedItems")] deleted_items: usize, } + +#[derive(Deserialize)] +struct PollRangeQuery { + #[serde(default)] + prefix: Option<String>, + #[serde(default)] + start: Option<String>, + #[serde(default)] + end: Option<String>, + #[serde(default)] + timeout: Option<u64>, + #[serde(default, rename = "seenMarker")] + seen_marker: Option<String>, +} + +#[derive(Serialize)] +struct PollRangeResponse { + items: Vec<ReadBatchResponseItem>, + #[serde(rename = "seenMarker")] + seen_marker: String, +} |