Merge pull request #82 from hit-box/tarantool

Tarantool Backend

Cargo.toml

@@ -8,5 +8,6 @@ members = [
     "hitbox-redis",
     "hitbox-stretto",
     "hitbox-tower",
+    "hitbox-tarantool",
     "examples",
 ]

hitbox-backend/src/serializer.rs

@@ -27,7 +27,7 @@ pub trait Serializer {
 }
 
 #[derive(Deserialize, Serialize)]
-struct SerializableCachedValue<U> {
+pub struct SerializableCachedValue<U> {
     data: U,
     expired: DateTime<Utc>,
 }
@@ -38,6 +38,15 @@ impl<U> SerializableCachedValue<U> {
     }
 }
 
+impl<T> From<CachedValue<T>> for SerializableCachedValue<T> {
+    fn from(value: CachedValue<T>) -> SerializableCachedValue<T> {
+        SerializableCachedValue {
+            data: value.data,
+            expired: value.expired,
+        }
+    }
+}
+
 #[derive(Default)]
 pub struct JsonSerializer<Raw = Vec<u8>> {
     _raw: PhantomData<Raw>,

hitbox-tarantool/CHANGELOG.md

@@ -0,0 +1,12 @@
+# Changelog
+
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## Unreleased
+
+### Added
+
+- Initial release

hitbox-tarantool/Cargo.toml

@@ -0,0 +1,34 @@
+[package]
+name = "hitbox-tarantool"
+version = "0.1.0"
+authors = [
+    "Evgeniy <[email protected]>",
+    "Belousov Max <[email protected]>",
+    "Andrey Ermilov <[email protected]>",
+]
+license = "MIT"
+edition = "2021"
+description = "Hitbox tarantool backend."
+readme = "README.md"
+repository = "https://github.com/hit-box/hitbox/"
+categories = ["caching", "asynchronous"]
+keywords = ["cache", "async", "cache-backend", "hitbox", "tarantool"]
+
+[dependencies]
+hitbox-backend = { path = "../hitbox-backend", version = "0.1.0" }
+hitbox-core = { path = "../hitbox-core", version = "0.1.0" }
+async-trait = "0.1"
+serde = "1"
+rusty_tarantool = "0.3.0"
+typed-builder = "0.15"
+
+[dev-dependencies]
+tokio = { version = "1", features = [
+    "time",
+    "macros",
+    "test-util",
+    "rt-multi-thread",
+] }
+testcontainers = "0.14"
+chrono = "0.4"
+once_cell = "1"

hitbox-tarantool/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2019 Makc
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

hitbox-tarantool/src/backend.rs

@@ -0,0 +1,147 @@
+use async_trait::async_trait;
+use hitbox_backend::{
+    serializer::SerializableCachedValue, BackendError, BackendResult, CacheBackend, DeleteStatus,
+};
+use hitbox_core::{CacheKey, CacheableResponse, CachedValue};
+use rusty_tarantool::tarantool::{Client, ClientConfig, ExecWithParamaters};
+use serde::{Deserialize, Serialize};
+use std::io::Error;
+use typed_builder::TypedBuilder;
+
+const TARANTOOL_INIT_LUA: &str = include_str!("init.lua");
+
+/// Tarantool cache backend based on rusty_tarantool crate.
+///
+/// # Examples
+/// ```
+/// use hitbox_tarantool::Tarantool;
+///
+/// #[tokio::main]
+/// async fn main() {
+///     let mut backend = Tarantool::builder().build();
+///     // backend.init().await.unwrap();
+/// }
+/// ```
+#[derive(Clone, TypedBuilder)]
+#[builder(build_method(vis="", name=__build))]
+pub struct Tarantool {
+    #[builder(default = "hitbox".to_string())]
+    user: String,
+    #[builder(default = "hitbox".to_string())]
+    password: String,
+    #[builder(default = "127.0.0.1".to_string())]
+    host: String,
+    #[builder(default = "3301".to_string())]
+    port: String,
+}
+
+pub struct TarantoolBackend {
+    client: Client,
+}
+
+#[allow(non_camel_case_types)]
+impl<
+        __user: ::typed_builder::Optional<String>,
+        __password: ::typed_builder::Optional<String>,
+        __host: ::typed_builder::Optional<String>,
+        __port: ::typed_builder::Optional<String>,
+    > TarantoolBuilder<(__user, __password, __host, __port)>
+{
+    pub fn build(self) -> TarantoolBackend {
+        let t = self.__build();
+        let client =
+            ClientConfig::new(format!("{}:{}", t.host, t.port), t.user, t.password).build();
+        TarantoolBackend { client }
+    }
+}
+
+impl TarantoolBackend {
+    /// Init backend and configure tarantool instance
+    /// This function is idempotent
+    pub async fn init(&mut self) -> BackendResult<()> {
+        self.client
+            .eval(TARANTOOL_INIT_LUA, &("hitbox_cache",))
+            .await
+            .map_err(|err| BackendError::InternalError(Box::new(err)))?;
+
+        Ok(())
+    }
+
+    fn map_err(err: Error) -> BackendError {
+        BackendError::InternalError(Box::new(err))
+    }
+}
+
+#[doc(hidden)]
+#[derive(Serialize, Deserialize)]
+pub struct CacheEntry<T> {
+    pub key: String,
+    pub ttl: Option<u32>,
+    pub value: SerializableCachedValue<T>,
+}
+
+#[async_trait]
+impl CacheBackend for TarantoolBackend {
+    async fn get<T>(&self, key: &CacheKey) -> BackendResult<Option<CachedValue<T::Cached>>>
+    where
+        T: CacheableResponse,
+        <T as CacheableResponse>::Cached: serde::de::DeserializeOwned,
+    {
+        self.client
+            .prepare_fn_call("hitbox.get")
+            .bind_ref(&(key.serialize()))
+            .map_err(TarantoolBackend::map_err)?
+            .execute()
+            .await
+            .map_err(TarantoolBackend::map_err)?
+            .decode_single::<Option<CacheEntry<T::Cached>>>()
+            .map_err(TarantoolBackend::map_err)
+            .map(|v| v.map(|v| v.value.into_cached_value()))
+    }
+
+    async fn delete(&self, key: &CacheKey) -> BackendResult<DeleteStatus> {
+        let result: bool = self
+            .client
+            .prepare_fn_call("hitbox.delete")
+            .bind_ref(&(key.serialize()))
+            .map_err(TarantoolBackend::map_err)?
+            .execute()
+            .await
+            .map_err(TarantoolBackend::map_err)?
+            .decode_single()
+            .map_err(TarantoolBackend::map_err)?;
+        match result {
+            true => Ok(DeleteStatus::Deleted(1)),
+            false => Ok(DeleteStatus::Missing),
+        }
+    }
+
+    async fn set<T>(
+        &self,
+        key: &CacheKey,
+        value: &CachedValue<T::Cached>,
+        ttl: Option<u32>,
+    ) -> BackendResult<()>
+    where
+        T: CacheableResponse + Send,
+        T::Cached: serde::Serialize + Send + Sync,
+    {
+        let entry: CacheEntry<T::Cached> = CacheEntry {
+            key: key.serialize(),
+            ttl,
+            value: value.clone().into(),
+        };
+        self.client
+            .prepare_fn_call("hitbox.set")
+            .bind_ref(&entry)
+            .map_err(TarantoolBackend::map_err)?
+            .execute()
+            .await
+            .map(|_| ())
+            .map_err(TarantoolBackend::map_err)
+    }
+
+    async fn start(&self) -> BackendResult<()> {
+        Ok(())
+    }
+}

hitbox-tarantool/src/init.lua

@@ -0,0 +1,73 @@
+local fiber = require("fiber")
+local log = require("log")
+
+local SCAN_INTERVAL = 0.1
+local MAX_TUPLES_FOR_DELETE = 1000
+
+box.cfg({})
+
+local space_name = ...
+
+box.schema.space.create(space_name, { if_not_exists = true })
+box.space[space_name]:create_index("primary", {
+	type = "HASH",
+	parts = { { 1, "string" } },
+	if_not_exists = true,
+})
+box.space[space_name]:create_index("by_ttl", {
+	parts = { { 2, "integer" } },
+	if_not_exists = true,
+})
+
+if not _G.__hitbox_cache_fiber then
+	_G.__hitbox_cache_fiber = fiber.create(function()
+		fiber.name("hitbox_cache_fiber")
+		while true do
+			box.ctl.wait_rw()
+
+			local ok, err = pcall(function()
+				local for_del = box.space[space_name].index.by_ttl
+					:pairs({ math.floor(fiber.time()) }, { iterator = "LE" })
+					:take(MAX_TUPLES_FOR_DELETE)
+					:totable()
+
+				box.atomic(function()
+					for _, t in pairs(for_del) do
+						box.space[space_name]:delete(t[1])
+					end
+				end)
+
+				return true
+			end)
+
+			if not ok then
+				log.error(err)
+			end
+
+			fiber.testcancel()
+			fiber.sleep(SCAN_INTERVAL)
+		end
+	end)
+end
+
+-- lua api for hitbox
+_G.hitbox = {
+	---Get cache entry by key
+	---@param key string
+	---@return table?
+	get = function(key)
+		return box.space[space_name]:get(key)
+	end,
+	---Insert cache entry
+	---@param entry table {key: string, ttl: number, value: any}
+	---@return table saved entry
+	set = function(entry)
+		return box.space[space_name]:replace(entry)
+	end,
+	---Delete cache entry
+	---@param key string
+	---@return boolean
+	delete = function(key)
+		return box.space[space_name]:delete(key) and true or false
+	end,
+}

hitbox-tarantool/src/lib.rs

@@ -0,0 +1,6 @@
+//! hitbox [Backend] implementation for Tarantool.
+//! [Backend]: hitbox_backend::Backend
+pub mod backend;
+
+#[doc(inline)]
+pub use crate::backend::Tarantool;