Fut/async file

Cargo.toml

@@ -10,7 +10,7 @@ serde = { version = "1.0.131", features = ["derive"] }
 serde_json = { version = "1.0", default-features = true, features = ["alloc"] }
 anyhow = "1.0.31"
 thiserror = "1.0"
-tokio = { version = "1.16.1", features = ["full"] }
+tokio = { version = "1.16.1", features = ["full", "tracing"] }
 byteorder = "1.4.3"
 rand = "0.8.5"
 maligned = "0.2.1"
@@ -47,10 +47,13 @@ eieio = "1.0.0"
 either = "1.8.1"
 enum-unitary = "0.5.0"
 atom_box = "0.1.2"
+console-subscriber = "0.1.8"
 [dev-dependencies]
 tracing-subscriber = "0.3.16"
 tracing-log = "0.1.3"
 chrono = "0.4.22"
 env_logger = "0.9.1"
 console_log = { version = "0.2.0", features = ["color"] }
 
+[build]
+rustflags = ["--cfg", "tokio_unstable"]

src/iterator.rs

@@ -1,28 +1,30 @@
 use crate::iterator::PreFetchStatus::Prefetched;
 use crate::kv::_BADGER_PREFIX;
-use crate::types::{ArcMx, ArcRW, Closer, TArcMx, TArcRW};
-use crate::MergeIterOverIterator;
+use crate::types::{ArcMx, ArcRW, Channel, Closer, TArcMx, TArcRW};
 use crate::{
     kv::KV,
     types::XArc,
     value_log::{MetaBit, ValuePointer},
     Decode, Result, Xiterator,
 };
+use crate::{MergeIterOverIterator, ValueStruct};
 use log::Metadata;
 use parking_lot::RwLock;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
 use std::{io::Cursor, sync::atomic::AtomicU64};
+use tokio::io::AsyncWriteExt;
+use tokio::time::Sleep;
 
 #[derive(Debug, PartialEq)]
 pub(crate) enum PreFetchStatus {
     Empty,
     Prefetched,
 }
 
-type KVItem = TArcRW<KVItemInner>;
+pub(crate) type KVItem = TArcRW<KVItemInner>;
 
 // Returned during iteration. Both the key() and value() output is only valid until
 // iterator.next() is called.
@@ -41,11 +43,39 @@ pub(crate) struct KVItemInner {
 }
 
 impl KVItemInner {
+    pub(crate) fn new(key: Vec<u8>, value: ValueStruct, kv: XArc<KV>) -> KVItemInner {
+        Self {
+            status: Arc::new(tokio::sync::RwLock::new(PreFetchStatus::Empty)),
+            kv,
+            key,
+            value: Arc::new(Default::default()),
+            vptr: value.value,
+            meta: value.meta,
+            user_meta: value.user_meta,
+            cas_counter: Arc::new(AtomicU64::new(value.cas_counter)),
+            wg: Closer::new("kv".to_owned()),
+            err: Ok(()),
+        }
+    }
+
     // Returns the key. Remember to copy if you need to access it outside the iteration loop.
     pub(crate) fn key(&self) -> &[u8] {
         &self.key
     }
 
+    pub async fn get_value(&self) -> Result<Vec<u8>> {
+        let ch = Channel::new(1);
+        self.value(|value| {
+            let tx = ch.tx();
+            Box::pin(async move {
+                tx.send(value).await.unwrap();
+                Ok(())
+            })
+        })
+        .await?;
+        Ok(ch.recv().await.unwrap())
+    }
+
     // Value retrieves the value of the item from the value log. It calls the
     // consumer function with a slice argument representing the value. In case
     // of error, the consumer function is not called.