wip / connect to tokyo (main.rs)

src/main.rs

@@ -7,7 +7,8 @@
 mod cache;
 mod sieve;
 use bytes::Bytes;
-use std::net::SocketAddr;
+use cache::CacheWithTTLEntry;
+use std::{cell::RefCell, net::SocketAddr, sync::Arc};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::{TcpSocket, TcpStream},
@@ -18,7 +19,9 @@ use std::io;
 
 use crate::{cache::CacheWithTTL, sieve::ESieve};
 
-async fn service(mut socket: TcpStream, addr: SocketAddr) {
+type SharedCache = Arc<RefCell<CacheWithTTL<Bytes, ESieve<CacheWithTTLEntry<Bytes>>>>>;
+
+async fn service(mut socket: TcpStream, addr: SocketAddr, cache: SharedCache) {
     println!("new client: {}", addr);
     let mut buf = [0; 1024];
 
@@ -36,9 +39,19 @@ async fn service(mut socket: TcpStream, addr: SocketAddr) {
             };
 
             // Write the data back
-            if let Err(e) = socket.write_all(&buf[0..n]).await {
-                eprintln!("failed to write to socket; err = {:?}", e);
-                return;
+            // FIXME: this is not cancel safe, should move the select! on the reading part only
+            eprintln!(
+                "Receieved this buffer from {} => {:?} - but returning a value from cache",
+                addr,
+                &buf[0..n]
+            );
+            if let Some(bytes) = cache.borrow_mut().get("user-id-1") {
+                if let Err(e) = socket.write_all(&bytes).await {
+                    eprintln!("failed to write to socket; err = {:?}", e);
+                    return;
+                }
+            } else {
+                eprintln!("no cache value found!")
             }
         };
 
@@ -60,6 +73,7 @@ async fn main() -> io::Result<()> {
     // TODO: 1 shared cache, should be multiple
     // TODO: capacity should be total_capacity / cache_count
     let mut cache = CacheWithTTL::<Bytes, ESieve<_>>::new(1000);
+    let cache = Arc::new(RefCell::new(cache));
     // cache.set(
     //     "key-1",
     //     Bytes::from_static(b"value-1"),
@@ -87,7 +101,9 @@ async fn main() -> io::Result<()> {
     loop {
         let listen = async {
             let (socket, addr) = listener.accept().await.unwrap();
-            tasks.push(tokio::spawn(async move { service(socket, addr).await }));
+            tasks.push(tokio::spawn(async move {
+                service(socket, addr, cache.clone()).await
+            }));
         };
 
         select! {

src/sieve.rs

@@ -4,8 +4,7 @@ use crate::cache::Cache;
 use std::{
     cell::RefCell,
     collections::{HashMap, VecDeque},
-    rc::Rc,
-    sync::{Mutex, RwLock},
+    sync::{Arc, Mutex, RwLock},
 };
 
 struct SieveNode {
@@ -15,8 +14,8 @@ struct SieveNode {
 
 struct Sieve {
     capacity: usize,
-    cache: HashMap<&'static str, Rc<RefCell<SieveNode>>>,
-    log: VecDeque<Rc<RefCell<SieveNode>>>,
+    cache: HashMap<&'static str, Arc<RefCell<SieveNode>>>,
+    log: VecDeque<Arc<RefCell<SieveNode>>>,
     hand: Option<usize>,
 }
 
@@ -30,7 +29,7 @@ impl Sieve {
         }
     }
 
-    fn get_mut_node(&mut self, key: &'static str) -> Option<Rc<RefCell<SieveNode>>> {
+    fn get_mut_node(&mut self, key: &'static str) -> Option<Arc<RefCell<SieveNode>>> {
         self.cache.get_mut(key).cloned()
     }
 
@@ -74,15 +73,15 @@ impl Sieve {
 
         // if we do not hit full capacity we have this shortcut
         if self.log.len() < self.capacity {
-            let node = Rc::new(RefCell::new(node));
+            let node = Arc::new(RefCell::new(node));
             self.log.push_front(node.clone());
             self.cache.insert(key, node);
             return;
         }
 
         // otherwise, we use sieve algorithm and then push
         self.evict();
-        let node = Rc::new(RefCell::new(node));
+        let node = Arc::new(RefCell::new(node));
         self.log.push_front(node.clone());
         self.cache.insert(key, node);
     }
@@ -130,7 +129,10 @@ pub struct ESieve<T: Clone> {
     sieve: Mutex<Sieve>,
 }
 
-// TODO: miss ttl support, maybe have a wrapper to handle that since it'll be the same impl for every cache
+// FIXME: do we need this?
+// unsafe impl<T: Clone> Send for ESieve<T> {}
+// unsafe impl<T: Clone> Sync for ESieve<T> {}
+
 impl<T: Clone> Cache<T> for ESieve<T> {
     fn new(capacity: usize) -> Self {
         Self {