use async_trait::async_trait;
use bytes::Bytes;
use structfs_ll_store::{AsyncLLReader, AsyncLLWriter, LLError, LLPath};
use crate::{
async_traits::{AsyncReader, AsyncWriter},
bridge::{path_from_bytes, path_from_ll},
Codec, Error, Format, Path, Record,
};
pub struct AsyncLLToCore<T, C> {
inner: T,
codec: C,
read_format: Format,
write_format: Format,
}
impl<T, C> AsyncLLToCore<T, C> {
pub fn new(inner: T, codec: C, format: Format) -> Self {
Self {
inner,
codec,
read_format: format.clone(),
write_format: format,
}
}
pub fn with_formats(inner: T, codec: C, read_format: Format, write_format: Format) -> Self {
Self {
inner,
codec,
read_format,
write_format,
}
}
pub fn inner(&self) -> &T {
&self.inner
}
pub fn inner_mut(&mut self) -> &mut T {
&mut self.inner
}
pub fn into_inner(self) -> T {
self.inner
}
}
#[async_trait]
impl<T: AsyncLLReader, C: Send + Sync> AsyncReader for AsyncLLToCore<T, C> {
async fn read_async(&mut self, from: &Path) -> Result<Option<Record>, Error> {
let components: Vec<Vec<u8>> = from.as_ll().iter().map(|b| b.to_vec()).collect();
let refs: Vec<&[u8]> = components.iter().map(|v| v.as_slice()).collect();
let bytes = match self.inner.ll_read_async(&refs).await {
Ok(Some(b)) => b,
Ok(None) => return Ok(None),
Err(e) => return Err(Error::Ll(e)),
};
Ok(Some(Record::raw(bytes, self.read_format.clone())))
}
}
#[async_trait]
impl<T: AsyncLLWriter, C: Codec + Send + Sync> AsyncWriter for AsyncLLToCore<T, C> {
async fn write_async(&mut self, to: &Path, data: Record) -> Result<Path, Error> {
let bytes = data.into_bytes(&self.codec, &self.write_format)?;
let components: Vec<Vec<u8>> = to.as_ll().iter().map(|b| b.to_vec()).collect();
let refs: Vec<&[u8]> = components.iter().map(|v| v.as_slice()).collect();
let result_path = self
.inner
.ll_write_async(&refs, bytes)
.await
.map_err(Error::Ll)?;
path_from_ll(&result_path)
}
}
pub struct AsyncCoreToLL<T, C> {
inner: T,
codec: C,
format: Format,
}
impl<T, C> AsyncCoreToLL<T, C> {
pub fn new(inner: T, codec: C, format: Format) -> Self {
Self {
inner,
codec,
format,
}
}
pub fn inner(&self) -> &T {
&self.inner
}
pub fn inner_mut(&mut self) -> &mut T {
&mut self.inner
}
pub fn into_inner(self) -> T {
self.inner
}
}
#[async_trait]
impl<T: AsyncReader, C: Codec + Send + Sync> AsyncLLReader for AsyncCoreToLL<T, C> {
async fn ll_read_async(&mut self, path: &[&[u8]]) -> Result<Option<Bytes>, LLError> {
let path = path_from_bytes(path).map_err(|e| LLError::Protocol {
code: 1,
detail: Bytes::copy_from_slice(e.to_string().as_bytes()),
})?;
let record = match self.inner.read_async(&path).await {
Ok(Some(r)) => r,
Ok(None) => return Ok(None),
Err(e) => {
return Err(LLError::Protocol {
code: 2,
detail: Bytes::copy_from_slice(e.to_string().as_bytes()),
})
}
};
let bytes =
record
.into_bytes(&self.codec, &self.format)
.map_err(|e| LLError::Protocol {
code: 3,
detail: Bytes::copy_from_slice(e.to_string().as_bytes()),
})?;
Ok(Some(bytes))
}
}
#[async_trait]
impl<T: AsyncWriter, C: Send + Sync> AsyncLLWriter for AsyncCoreToLL<T, C> {
async fn ll_write_async(&mut self, path: &[&[u8]], data: Bytes) -> Result<LLPath, LLError> {
let path = path_from_bytes(path).map_err(|e| LLError::Protocol {
code: 1,
detail: Bytes::copy_from_slice(e.to_string().as_bytes()),
})?;
let record = Record::raw(data, self.format.clone());
let result_path =
self.inner
.write_async(&path, record)
.await
.map_err(|e| LLError::Protocol {
code: 2,
detail: Bytes::copy_from_slice(e.to_string().as_bytes()),
})?;
Ok(result_path.into_ll())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{path, NoCodec};
use std::collections::HashMap;
struct TestAsyncLLStore {
data: HashMap<Vec<Vec<u8>>, Bytes>,
}
impl TestAsyncLLStore {
fn new() -> Self {
Self {
data: HashMap::new(),
}
}
}
#[async_trait]
impl AsyncLLReader for TestAsyncLLStore {
async fn ll_read_async(&mut self, path: &[&[u8]]) -> Result<Option<Bytes>, LLError> {
let key: Vec<Vec<u8>> = path.iter().map(|c| c.to_vec()).collect();
Ok(self.data.get(&key).cloned())
}
}
#[async_trait]
impl AsyncLLWriter for TestAsyncLLStore {
async fn ll_write_async(&mut self, path: &[&[u8]], data: Bytes) -> Result<LLPath, LLError> {
let key: Vec<Vec<u8>> = path.iter().map(|c| c.to_vec()).collect();
self.data.insert(key, data);
Ok(path.iter().map(|c| Bytes::copy_from_slice(c)).collect())
}
}
struct TestAsyncCoreStore {
data: HashMap<Path, Record>,
}
impl TestAsyncCoreStore {
fn new() -> Self {
Self {
data: HashMap::new(),
}
}
}
#[async_trait]
impl AsyncReader for TestAsyncCoreStore {
async fn read_async(&mut self, from: &Path) -> Result<Option<Record>, Error> {
Ok(self.data.get(from).cloned())
}
}
#[async_trait]
impl AsyncWriter for TestAsyncCoreStore {
async fn write_async(&mut self, to: &Path, data: Record) -> Result<Path, Error> {
self.data.insert(to.clone(), data);
Ok(to.clone())
}
}
#[tokio::test]
async fn async_ll_to_core_read() {
let mut ll = TestAsyncLLStore::new();
ll.data.insert(
vec![b"users".to_vec(), b"123".to_vec()],
Bytes::from_static(b"hello"),
);
let mut bridge = AsyncLLToCore::new(ll, NoCodec, Format::OCTET_STREAM);
let result = bridge.read_async(&path!("users/123")).await.unwrap();
assert!(result.is_some());
assert_eq!(
result.unwrap().as_bytes(),
Some(&Bytes::from_static(b"hello"))
);
}
#[tokio::test]
async fn async_ll_to_core_write() {
let ll = TestAsyncLLStore::new();
let mut bridge = AsyncLLToCore::new(ll, NoCodec, Format::OCTET_STREAM);
let record = Record::raw(Bytes::from_static(b"data"), Format::OCTET_STREAM);
bridge
.write_async(&path!("test/path"), record)
.await
.unwrap();
let key = vec![b"test".to_vec(), b"path".to_vec()];
assert!(bridge.inner().data.contains_key(&key));
}
#[tokio::test]
async fn async_core_to_ll_read() {
let mut core = TestAsyncCoreStore::new();
core.data.insert(
path!("users/123"),
Record::raw(Bytes::from_static(b"hello"), Format::OCTET_STREAM),
);
let mut bridge = AsyncCoreToLL::new(core, NoCodec, Format::OCTET_STREAM);
let result = bridge.ll_read_async(&[b"users", b"123"]).await.unwrap();
assert_eq!(result, Some(Bytes::from_static(b"hello")));
}
#[tokio::test]
async fn async_core_to_ll_write() {
let core = TestAsyncCoreStore::new();
let mut bridge = AsyncCoreToLL::new(core, NoCodec, Format::OCTET_STREAM);
bridge
.ll_write_async(&[b"test", b"path"], Bytes::from_static(b"data"))
.await
.unwrap();
assert!(bridge.inner().data.contains_key(&path!("test/path")));
}
#[tokio::test]
async fn async_invalid_utf8_path_rejected() {
let core = TestAsyncCoreStore::new();
let mut bridge = AsyncCoreToLL::new(core, NoCodec, Format::OCTET_STREAM);
let result = bridge.ll_read_async(&[&[0xFF, 0xFE]]).await;
assert!(matches!(result, Err(LLError::Protocol { .. })));
}
}