metacache done

ecstore/src/disk/mod.rs

@@ -594,7 +594,7 @@ pub struct MetadataResolutionParams {
     pub candidates: Vec<Vec<FileMetaShallowVersion>>,
 }
 
-#[derive(Clone, Debug, Default, Serialize, Deserialize)]
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
 pub struct MetaCacheEntry {
     // name is the full name of the object including prefixes
     pub name: String,

ecstore/src/io.rs

@@ -1,3 +1,5 @@
+use std::io::Read;
+use std::io::Write;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio::fs::File;
@@ -62,7 +64,174 @@ impl AsyncWrite for Writer {
     }
 }
 
-// #[tokio::test]
-// async fn test_reader{
+pub struct AsyncToSync<R> {
+    inner: R,
+}
 
-// }
+impl<R: AsyncRead + Unpin> AsyncToSync<R> {
+    pub fn new_reader(inner: R) -> Self {
+        Self { inner }
+    }
+    fn read_async(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<std::io::Result<usize>> {
+        let mut read_buf = ReadBuf::new(buf);
+        // Poll the underlying AsyncRead to fill the ReadBuf
+        match Pin::new(&mut self.inner).poll_read(cx, &mut read_buf) {
+            Poll::Ready(Ok(())) => Poll::Ready(Ok(read_buf.filled().len())),
+            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
+impl<R: AsyncWrite + Unpin> AsyncToSync<R> {
+    pub fn new_writer(inner: R) -> Self {
+        Self { inner }
+    }
+    // This function will perform a write using AsyncWrite
+    fn write_async(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
+        let result = Pin::new(&mut self.inner).poll_write(cx, buf);
+        match result {
+            Poll::Ready(Ok(n)) => Poll::Ready(Ok(n)),
+            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+
+    // This function will perform a flush using AsyncWrite
+    fn flush_async(&mut self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.inner).poll_flush(cx)
+    }
+}
+
+impl<R: AsyncRead + Unpin> Read for AsyncToSync<R> {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
+        loop {
+            match self.read_async(&mut cx, buf) {
+                Poll::Ready(Ok(n)) => return Ok(n),
+                Poll::Ready(Err(e)) => return Err(e),
+                Poll::Pending => {
+                    // If Pending, we need to wait for the readiness.
+                    // Here, we can use an arbitrary mechanism to yield control,
+                    // this might be blocking until some readiness occurs can be complex.
+                    // A full blocking implementation would require an async runtime to block on.
+                    std::thread::sleep(std::time::Duration::from_millis(1)); // Replace with proper waiting if needed
+                }
+            }
+        }
+    }
+}
+
+impl<W: AsyncWrite + Unpin> Write for AsyncToSync<W> {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
+        loop {
+            match self.write_async(&mut cx, buf) {
+                Poll::Ready(Ok(n)) => return Ok(n),
+                Poll::Ready(Err(e)) => return Err(e),
+                Poll::Pending => {
+                    // Here we are blocking and waiting for the async operation to complete.
+                    std::thread::sleep(std::time::Duration::from_millis(1)); // Not efficient, see notes.
+                }
+            }
+        }
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
+        loop {
+            match self.flush_async(&mut cx) {
+                Poll::Ready(Ok(())) => return Ok(()),
+                Poll::Ready(Err(e)) => return Err(e),
+                Poll::Pending => {
+                    // Again, blocking to wait for flush.
+                    std::thread::sleep(std::time::Duration::from_millis(1)); // Not efficient, see notes.
+                }
+            }
+        }
+    }
+}
+
+pub struct VecAsyncWriter {
+    buffer: Vec<u8>,
+}
+
+impl VecAsyncWriter {
+    /// Create a new VecAsyncWriter with an empty Vec<u8>.  
+    pub fn new(buffer: Vec<u8>) -> Self {
+        VecAsyncWriter { buffer }
+    }
+
+    /// Retrieve the underlying buffer.  
+    pub fn get_buffer(&self) -> &[u8] {
+        &self.buffer
+    }
+}
+
+// Implementing AsyncWrite trait for VecAsyncWriter
+impl AsyncWrite for VecAsyncWriter {
+    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
+        let len = buf.len();
+
+        // Assume synchronous writing for simplicity
+        self.get_mut().buffer.extend_from_slice(buf);
+
+        // Returning the length of written data
+        Poll::Ready(Ok(len))
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        // In this case, flushing is a no-op for a Vec<u8>
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        // Similar to flush, shutdown has no effect here
+        Poll::Ready(Ok(()))
+    }
+}
+
+pub struct VecAsyncReader {
+    buffer: Vec<u8>,
+    position: usize,
+}
+
+impl VecAsyncReader {
+    /// Create a new VecAsyncReader with the given Vec<u8>.  
+    pub fn new(buffer: Vec<u8>) -> Self {
+        VecAsyncReader { buffer, position: 0 }
+    }
+
+    /// Reset the reader position.  
+    pub fn reset(&mut self) {
+        self.position = 0;
+    }
+}
+
+// Implementing AsyncRead trait for VecAsyncReader
+impl AsyncRead for VecAsyncReader {
+    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf) -> Poll<io::Result<()>> {
+        let this = self.get_mut();
+
+        // Check how many bytes are available to read
+        let len = this.buffer.len();
+        let bytes_available = len - this.position;
+
+        if bytes_available == 0 {
+            // If there's no more data to read, return ready with an Eof
+            return Poll::Ready(Ok(()));
+        }
+
+        // Calculate how much we can read into the provided buffer
+        let to_read = std::cmp::min(bytes_available, buf.remaining());
+
+        // Write the data to the buf
+        buf.put_slice(&this.buffer[this.position..this.position + to_read]);
+
+        // Update the position
+        this.position += to_read;
+
+        // Indicate how many bytes were read
+        Poll::Ready(Ok(()))
+    }
+}

ecstore/src/metacache/writer.rs

@@ -1,30 +1,23 @@
-use std::io::ErrorKind;
-use std::io::Read;
-use std::io::Write;
-use std::str::from_utf8;
-
 use crate::disk::MetaCacheEntry;
 use crate::error::Error;
 use crate::error::Result;
+use std::io::Read;
+use std::io::Write;
+use std::str::from_utf8;
 
 const METACACHE_STREAM_VERSION: u8 = 2;
 
 pub struct MetacacheWriter<W> {
     wr: W,
-    buf: Vec<u8>,
     created: bool,
 }
 
-impl<W: Write + Unpin> MetacacheWriter<W> {
-    pub fn new(wr: W, block_size: usize) -> Self {
-        Self {
-            wr,
-            buf: Vec::with_capacity(block_size),
-            created: false,
-        }
+impl<W: Write> MetacacheWriter<W> {
+    pub fn new(wr: W) -> Self {
+        Self { wr, created: false }
     }
 
-    async fn write(&mut self, objs: &[MetaCacheEntry]) -> Result<()> {
+    pub fn write(&mut self, objs: &[MetaCacheEntry]) -> Result<()> {
         if objs.is_empty() {
             return Ok(());
         }
@@ -49,9 +42,8 @@ impl<W: Write + Unpin> MetacacheWriter<W> {
         Ok(())
     }
 
-    async fn close(&mut self) -> Result<()> {
+    pub fn close(&mut self) -> Result<()> {
         rmp::encode::write_bool(&mut self.wr, false)?;
-
         self.wr.flush()?;
         Ok(())
     }
@@ -74,8 +66,15 @@ impl<R: Read> MetacacheReader<R> {
         }
     }
 
-    pub fn check_init(&mut self) {
+    fn check_init(&mut self) {
         if !self.init {
+            // let mut buf = match self.read_buf(1).await {
+            //     Ok(res) => res,
+            //     Err(err) => {
+            //         self.err = Some(Error::msg(err.to_string()));
+            //         return;
+            //     }
+            // };
             let ver = match rmp::decode::read_u8(&mut self.rd) {
                 Ok(res) => res,
                 Err(err) => {
@@ -94,7 +93,7 @@ impl<R: Read> MetacacheReader<R> {
         }
     }
 
-    pub fn peek(&mut self) -> Result<MetaCacheEntry> {
+    pub fn peek(&mut self) -> Result<Option<MetaCacheEntry>> {
         self.check_init();
 
         if let Some(err) = &self.err {
@@ -104,8 +103,7 @@ impl<R: Read> MetacacheReader<R> {
         match rmp::decode::read_bool(&mut self.rd) {
             Ok(res) => {
                 if !res {
-                    self.err = Some(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof)));
-                    return Err(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof)));
+                    return Ok(None);
                 }
             }
             Err(err) => {
@@ -158,49 +156,64 @@ impl<R: Read> MetacacheReader<R> {
 
         let metadata = self.buf.clone();
 
-        Ok(MetaCacheEntry {
+        Ok(Some(MetaCacheEntry {
             name,
             metadata,
             cached: None,
             reusable: false,
-        })
+        }))
+    }
+
+    pub fn read_all(&mut self) -> Result<Vec<MetaCacheEntry>> {
+        let mut ret = Vec::new();
+
+        loop {
+            if let Some(entry) = self.peek()? {
+                ret.push(entry);
+                continue;
+            }
+
+            break;
+        }
+
+        Ok(ret)
     }
 }
 
 #[tokio::test]
 async fn test_writer() {
-    use std::fs::File;
-    use std::fs::OpenOptions;
+    use crate::io::AsyncToSync;
+    use crate::io::VecAsyncReader;
+    use crate::io::VecAsyncWriter;
 
-    let file_path = "./test_writer.txt";
-    let f = OpenOptions::new()
-        .create(true)
-        .read(true)
-        .write(true)
-        .truncate(true)
-        .open(file_path)
-        .unwrap();
+    let mut f = VecAsyncWriter::new(Vec::new());
 
-    // let wr = Writer::File(f);
-
-    let mut w = MetacacheWriter::new(f, 1024);
+    let mut w = MetacacheWriter::new(AsyncToSync::new_writer(&mut f));
 
     let mut objs = Vec::new();
     for i in 0..10 {
-        objs.push(MetaCacheEntry {
+        let info = MetaCacheEntry {
             name: format!("item{}", i),
             metadata: vec![0u8, 10],
             cached: None,
             reusable: false,
-        });
+        };
+        println!("old {:?}", &info);
+        objs.push(info);
     }
 
-    w.write(&objs).await.unwrap();
-    w.close().await.unwrap();
+    w.write(&objs).unwrap();
 
-    let nf = File::open(file_path).unwrap();
+    w.close().unwrap();
 
-    let meta = nf.metadata().unwrap();
+    let nf = VecAsyncReader::new(f.get_buffer().to_vec());
 
-    println!("{}", meta.len());
+    let mut r = MetacacheReader::new(AsyncToSync::new_reader(nf));
+    let nobjs = r.read_all().unwrap();
+
+    for info in nobjs.iter() {
+        println!("new {:?}", &info);
+    }
+
+    assert_eq!(objs, nobjs)
 }