updater_thread now works and we listen on unix_socket

src/core/instance/mod.rs

@@ -1,59 +1,102 @@
 use core::fmt;
 use std::collections::HashMap;
 use std::fs::create_dir;
+use std::{thread, vec};
+use crossbeam::channel;
 use std::sync::{Arc, Mutex};
-use log::{trace, debug, info, warn, error};
+use tracing::{trace, debug, info, warn, error};
 use uuid::Uuid;
 use std::fs::File;
 use colored::Colorize;
 use std::path::Path;
 use interner::shared::{VecStringPool, StringPool};
 
+
 use crate::error::{MizeError, MizeResult, IntoMizeResult, MizeResultTrait};
-use crate::instance;
 use crate::instance::store::Store;
-use crate::instance::connection::Connection;
+use crate::instance::updater::updater_thread;
+use crate::instance::peer::Peer;
 use crate::id::{IntoMizeId, MizeId, Namespace};
 use crate::instance::subscription::Subscription;
 use crate::item::{Item, ItemData};
 use crate::memstore::MemStore;
+use crate::instance::updater::Operation;
 
-pub mod connection;
+use self::peer::PeerListener;
+
+#[cfg(feature = "async")]
+use tokio::runtime::Runtime;
+use core::future::Future;
+
+pub mod peer;
 pub mod store;
 pub mod subscription;
+pub mod updater;
 
 static MSG_CHANNEL_SIZE: usize = 200;
 
 /// The Instance type is the heart of the mize system
+#[derive(Clone)]
 pub struct Instance {
     // question vec of stores???
     // would mean replication logic and such things is handeled by the instance
     // i think it would be better thogh to then implement a ReplicatedStore
     pub store: Box<dyn Store>,
-    peers: Arc<Mutex<Vec<Box<dyn Connection>>>>,
+    peers: Arc<Mutex<Vec<Peer>>>,
     subs: Arc<Mutex<HashMap<MizeId, Subscription>>>,
-    pub id_pool: VecStringPool,
-    pub namespace_pool: StringPool,
+    pub id_pool: Arc<Mutex<VecStringPool>>,
+    pub namespace_pool: Arc<Mutex<StringPool>>,
     pub namespace: Namespace,
+    op_tx: channel::Sender<Operation>,
+    context: Vec<MizeId>,
+    threads: Vec<String>,
+    #[cfg(feature = "async")]
+    runtime: Arc<Mutex<Runtime>>,
+}
+
+pub struct InstanceRef {
+    inner: Arc<Mutex<Instance>>,
+}
+
+pub struct InstanceAsync {
+    inner: Arc<Mutex<Instance>>,
 }
 
 
 impl Instance {
-    pub fn empty() -> Instance {
+    pub fn empty() -> MizeResult<Instance> {
         let store = MemStore::new();
-        let id_pool = VecStringPool::default();
-        let namespace_pool = StringPool::default();
+        let id_pool = Arc::new(Mutex::new(VecStringPool::default()));
+        let namespace_pool_raw = StringPool::default();
         let peers = Arc::new(Mutex::new(Vec::new()));
         let subs = Arc::new(Mutex::new(HashMap::new()));
-        let namespace = Namespace ( namespace_pool.get("mize.default.namespace") );
-        let mut instance = Instance { store: Box::new(store), peers, subs, id_pool, namespace_pool, namespace };
-        return instance;
+        let namespace = Namespace ( namespace_pool_raw.get("mize.default.namespace") );
+        let (op_tx, op_rx) = channel::unbounded();
+
+        let mut instance = Instance { 
+            store: Box::new(store), 
+            peers, subs, id_pool,
+            namespace, op_tx,
+            namespace_pool: Arc::new(Mutex::new(namespace_pool_raw)),
+            context: vec![],
+            threads: vec![],
+
+            #[cfg(feature = "async")]
+            runtime: Arc::new(Mutex::new(Runtime::new().mize_result_msg("Could not create async runtime")?)),
+        };
+
+        let instance_clone = instance.clone();
+        let closure = move || updater_thread(op_rx, instance_clone);
+        instance.spawn("updater_thread", closure)?;
+
+
+        return Ok(instance);
     }
 
     pub fn new() -> MizeResult<Instance> {
         trace!("[ {} ] Instance::new()", "CALL".yellow());
 
-        let mut instance = Instance::empty();
+        let mut instance = Instance::empty()?;
 
         instance.init();
 
@@ -77,7 +120,7 @@ impl Instance {
     pub fn with_config(config: ItemData) -> MizeResult<Instance> {
         trace!("[ {} ] Instance::with_config()", "CALL".yellow());
         trace!("config: {}", config);
-        let mut instance = Instance::empty();
+        let mut instance = Instance::empty()?;
         instance.set("0", config.clone());
         instance.init()?;
 
@@ -94,56 +137,86 @@ impl Instance {
         println!("MIGRATING");
         let old_store = &self.store;
 
-        let inst_data = self.store.get_value_data_full(self.id_from_string("0".to_owned()))?;
-        new_store.set(self.id_from_string("0".to_owned()), inst_data.to_owned())?;
+        let id = self.id_from_string("0".to_owned())?;
+        let inst_data = self.store.get_value_data_full(id.clone())?;
+        new_store.set(id, inst_data.to_owned())?;
 
         for id in old_store.id_iter()? {
-            let data = self.store.get_value_data_full(self.id_from_string(id?))?;
+            let data = self.store.get_value_data_full(self.id_from_string(id?)?)?;
 
             let id_of_new_store = new_store.new_id()?;
-            new_store.set(self.id_from_string(id_of_new_store), data.to_owned())?;
+            new_store.set(self.id_from_string(id_of_new_store)?, data.to_owned())?;
         };
 
         self.store = new_store;
 
         Ok(())
     }
-}
 
-impl Instance {
     pub fn new_item(&self) -> MizeResult<Item> {
-        let id = self.id_from_string(self.store.new_id()?);
+        let id = self.id_from_string(self.store.new_id()?)?;
         return Ok(Item::new(id, &self));
     }
 
     pub fn get<I: IntoMizeId>(&self, id: I) -> MizeResult<Item> {
-        let id = id.to_mize_id(self);
+        let id = id.to_mize_id(self)?;
         return Ok(Item::new(id, &self));
     }
 
     pub fn set<I: IntoMizeId, V: Into<ItemData>>(&mut self, id: I, value: V) -> MizeResult<()> {
-        let id = id.to_mize_id(self);
-        let item_data = value.into();
-        let mut item = self.get(id.clone())?;
-        item.merge(item_data)?;
+        let id = id.to_mize_id(self)?;
+        self.op_tx.send(Operation::Set(id, value.into()));
         Ok(())
     }
 
-    pub fn new_id<T: IntoMizeId>(&self, value: T) -> MizeId {
+    pub fn new_id<T: IntoMizeId>(&self, value: T) -> MizeResult<MizeId> {
         value.to_mize_id(self)
     }
 
-    pub fn id_from_string(&self, string: String) -> MizeId {
+    pub fn id_from_string(&self, string: String) -> MizeResult<MizeId> {
         let vec_string: Vec<String> = string.split("/").map(|v| v.to_owned()).collect();
-        MizeId { path: self.id_pool.get(vec_string), namespace: self.namespace.clone() }
+        let id_pool_inner = self.id_pool.lock()?;
+
+        let id = MizeId { path: id_pool_inner.get(vec_string), namespace: self.namespace.clone() };
+
+        Ok(id)
     }
 
-    pub fn id_from_vec_string(&self, vec_string: Vec<String>) -> MizeId {
-        MizeId { path: self.id_pool.get(vec_string), namespace: self.namespace.clone() }
+    pub fn id_from_vec_string(&self, vec_string: Vec<String>) -> MizeResult<MizeId> {
+        let id_pool_inner = self.id_pool.lock()?;
+        let id = MizeId { path: id_pool_inner.get(vec_string), namespace: self.namespace.clone() };
+        Ok(id)
     }
 
-    pub fn namespace_from_string(&self, ns_str: String) -> Namespace {
-        Namespace ( self.namespace_pool.get(ns_str) )
+    pub fn namespace_from_string(&self, ns_str: String) -> MizeResult<Namespace> {
+        let namespace_pool_inner = self.namespace_pool.lock()?;
+        let namespace = Namespace ( namespace_pool_inner.get(ns_str) );
+        Ok(namespace)
+    }
+
+    pub fn add_listener<T: PeerListener +'static>(&mut self, listener: T) -> MizeResult<()> {
+        let mut instance_clone = self.clone();
+        self.spawn("some_listener", move || listener.listen(instance_clone));
+        Ok(())
+    }
+
+    pub fn add_peer(&mut self, peer: Peer) -> MizeResult<()> {
+        let mut peer_inner = self.peers.lock()?;
+        peer_inner.push(peer);
+        Ok(())
+    }
+
+    pub fn spawn(&mut self, name: &str, func: impl FnOnce() -> MizeResult<()> + Send + 'static) -> MizeResult<()> {
+        self.threads.push(name.to_owned());
+        thread::spawn(move || func());
+        Ok(())
+    }
+
+    #[cfg(feature = "async")]
+    pub fn spawn_async<F: Future<Output = impl Send + Sync + 'static> + Send + Sync + 'static>(&mut self, name: &str, func: F) {
+        self.threads.push(name.to_owned());
+        let runtime_inner = self.runtime.lock().unwrap();
+        runtime_inner.spawn(func);
     }
 }