Add VSS Http thin client implementation for get/put/listKeyVersions api's

Author: G8XSU

vss-accessor/Cargo.toml

@@ -4,6 +4,8 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+prost = "0.11.9"
+reqwest = { version = "0.11.13", features = ["rustls-tls"] }
 
 [dev-dependencies]
 prost-build = { version = "0.11.3" }

vss-accessor/src/lib.rs

@@ -1 +1,107 @@
+use ::prost::Message;
+use reqwest;
+use reqwest::Client;
+use std::error::Error;
 
+use crate::vss::{
+	GetObjectRequest, GetObjectResponse, KeyValue, ListKeyVersionsRequest, ListKeyVersionsResponse, PutObjectRequest,
+	PutObjectResponse,
+};
+use crate::vss_error::VssError;
+
+mod vss_error;
+
+pub mod vss {
+	include!("generated-src/org.vss.rs");
+}
+
+/// Thin-client to access hosted instance of Versioned Storage Service (VSS).
+/// Api for VssAccessor is minimalistic and directly mimics vss-server-side api's as it is.
+pub struct VssAccessor {
+	base_url: String,
+	client: Client,
+}
+
+impl VssAccessor {
+	/// Constructs new instance of VssAccessor.
+	/// `base_url` is the vss-server endpoint to be used.
+	pub fn new(base_url: &str) -> Result<Self, Box<dyn Error>> {
+		let client = Client::new();
+		Ok(Self { base_url: String::from(base_url), client })
+	}
+
+	/// get_object fetches a value against a given key.
+	/// Makes a service call to GetObject endpoint of vss-server.
+	/// For api-contract/usage, refer docs for: [`GetObjectRequest`] and [`GetObjectResponse`]
+	pub async fn get_object(&self, store: String, key: String) -> Result<GetObjectResponse, VssError> {
+		let url = format!("{}/getObject", self.base_url);
+
+		let request = GetObjectRequest { store_id: store, key };
+
+		let response_raw = self.client.post(url).body(request.encode_to_vec()).send().await?;
+		let status = response_raw.status();
+		let payload = response_raw.bytes().await?;
+
+		if status.is_success() {
+			let response = GetObjectResponse::decode(&payload[..])?;
+			Ok(response)
+		} else {
+			Err(VssError::new(status, payload))
+		}
+	}
+
+	/// put_object writes a value against the key in request.
+	/// Makes a service call to PutObject endpoint of vss-server, with single item.
+	/// For api-contract/usage, refer docs for: [`PutObjectRequest`] and [`PutObjectResponse`]
+	pub async fn put_object(
+		&self, store: String, global_version: Option<i64>, key: String, version: i64, value: &[u8],
+	) -> Result<PutObjectResponse, VssError> {
+		let kv = KeyValue { key: String::from(key), version, value: value.to_vec() };
+		return self.put_objects_tx(store, global_version, vec![kv]).await;
+	}
+
+	/// put_objects_tx writes multiple transaction_items in single transaction.
+	/// Makes a service call to PutObject endpoint of vss-server, with multiple items.
+	/// Items in the request are written in a single all-or-nothing transaction.
+	/// For api-contract/usage, refer docs for: [`PutObjectRequest`] and [`PutObjectResponse`]
+	pub async fn put_objects_tx(
+		&self, store: String, global_version: Option<i64>, transaction_items: Vec<KeyValue>,
+	) -> Result<PutObjectResponse, VssError> {
+		let url = format!("{}/putObjects", self.base_url);
+
+		let request = PutObjectRequest { store_id: store, global_version, transaction_items };
+
+		let response_raw = self.client.post(url).body(request.encode_to_vec()).send().await?;
+		let status = response_raw.status();
+		let payload = response_raw.bytes().await?;
+
+		if status.is_success() {
+			let response = PutObjectResponse::decode(&payload[..])?;
+			Ok(response)
+		} else {
+			Err(VssError::new(status, payload))
+		}
+	}
+
+	/// list_key_versions lists keys and their corresponding version.
+	/// Makes a service call to ListKeyVersions endpoint of vss-server.
+	/// For api-contract/usage, refer docs for: [`ListKeyVersionsRequest`] and [`ListKeyVersionsResponse`]
+	pub async fn list_key_versions(
+		&self, store: String, key_prefix: String, page_size: Option<i32>, page_token: Option<String>,
+	) -> Result<ListKeyVersionsResponse, VssError> {
+		let url = format!("{}/listKeyVersions", self.base_url);
+
+		let request = ListKeyVersionsRequest { store_id: store, key_prefix: Some(key_prefix), page_size, page_token };
+
+		let response_raw = self.client.post(url).body(request.encode_to_vec()).send().await?;
+		let status = response_raw.status();
+		let payload = response_raw.bytes().await?;
+
+		if status.is_success() {
+			let response = ListKeyVersionsResponse::decode(&payload[..])?;
+			Ok(response)
+		} else {
+			Err(VssError::new(status, payload))
+		}
+	}
+}

vss-accessor/src/vss_error.rs

@@ -0,0 +1,75 @@
+use crate::vss::{ErrorCode, ErrorResponse};
+use prost::bytes::Bytes;
+use prost::{DecodeError, Message};
+use reqwest::StatusCode;
+use std::error::Error;
+use std::fmt::{Display, Formatter};
+
+#[derive(Debug)]
+pub enum VssError {
+	InvalidRequestError(ErrorResponse),
+	ConflictError(ErrorResponse),
+	InternalServerError(ErrorResponse),
+	InternalError(String),
+}
+
+impl VssError {
+	pub fn new(status: StatusCode, payload: Bytes) -> VssError {
+		match ErrorResponse::decode(&payload[..]) {
+			Ok(error_response) => VssError::from(error_response),
+			Err(e) => {
+				let message =
+					format!("Unable to decode ErrorResponse from server, HttpStatusCode: {}, DecodeErr: {}", status, e);
+				VssError::InternalError(message)
+			}
+		}
+	}
+}
+
+impl Display for VssError {
+	fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+		match self {
+			VssError::InvalidRequestError(error_response) => {
+				write!(f, "Request sent to VSS Server was invalid : {}", error_response.message)
+			}
+			VssError::ConflictError(error_response) => {
+				write!(f, "Potential version conflict in write operation : {}", error_response.message)
+			}
+			VssError::InternalServerError(error_response) => {
+				write!(f, "InternalServerError : {}", error_response.message)
+			}
+			VssError::InternalError(message) => {
+				write!(f, "InternalError : {}", message)
+			}
+		}
+	}
+}
+
+impl Error for VssError {}
+
+impl From<ErrorResponse> for VssError {
+	fn from(error_response: ErrorResponse) -> Self {
+		return match error_response.error_code() {
+			ErrorCode::InvalidRequestException => VssError::InvalidRequestError(error_response),
+			ErrorCode::ConflictException => VssError::ConflictError(error_response),
+			ErrorCode::InternalServerException => VssError::InternalServerError(error_response),
+			_ => VssError::InternalError(format!(
+				"Server responded with an unknown error code: {}, \
+             message: {}",
+				error_response.error_code, error_response.message
+			)),
+		};
+	}
+}
+
+impl From<DecodeError> for VssError {
+	fn from(err: DecodeError) -> Self {
+		VssError::InternalError(err.to_string())
+	}
+}
+
+impl From<reqwest::Error> for VssError {
+	fn from(err: reqwest::Error) -> Self {
+		VssError::InternalError(err.to_string())
+	}
+}