improve ports

add Send+Sync as supertraits for View and Observer and notify() immutable

src/channel.rs

@@ -23,7 +23,7 @@ use {
                   Traits
 <<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
                     \*/
-pub trait ChannelData : Default + IntoIterator {
+pub trait ChannelData : Default + IntoIterator + Send + Sync {
     fn channel_insert(&mut self, x: Self::Item);
 }
 
@@ -33,7 +33,8 @@ pub trait ChannelData : Default + IntoIterator {
                Queue Channel
 <<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
                     \*/
-impl<T> ChannelData for Vec<T> {
+impl<T> ChannelData for Vec<T>
+where T: Send + Sync {
     fn channel_insert(&mut self, x: T) {
         self.push(x);
     }
@@ -44,7 +45,8 @@ impl<T> ChannelData for Vec<T> {
                  Set Channel
 <<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
                     \*/
-impl<T: Eq + Hash> ChannelData for HashSet<T> {
+impl<T> ChannelData for HashSet<T>
+where T: Eq + Hash + Send + Sync {
     fn channel_insert(&mut self, x: T) {
         self.insert(x);
     }
@@ -55,7 +57,8 @@ impl<T: Eq + Hash> ChannelData for HashSet<T> {
              Singleton Channel
 <<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
                     \*/
-impl<T> ChannelData for Option<T> {
+impl<T> ChannelData for Option<T>
+where T: Send + Sync {
     fn channel_insert(&mut self, x: T) {
         *self = Some(x);
     }
@@ -80,10 +83,11 @@ pub struct ChannelReceiver<Data: ChannelData>(Arc<Mutex<ChannelState<Data>>>);
 
 //<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
 
-impl<Data: ChannelData> Observer for ChannelSender<Data> {
+impl<Data: ChannelData> Observer for ChannelSender<Data>
+where Data::IntoIter: Send + Sync {
     type Msg = Data::Item;
 
-    fn notify(&mut self, msg: Data::Item) {
+    fn notify(&self, msg: Data::Item) {
         let mut state =  self.0.lock().unwrap();
 
         if state.send_buf.is_none() {
@@ -163,15 +167,15 @@ pub fn channel<Data: ChannelData>() -> (ChannelSender<Data>, ChannelReceiver<Dat
     (ChannelSender(state.clone()), ChannelReceiver(state))
 }
 
-pub fn set_channel<T: Eq + Hash>() -> (ChannelSender<HashSet<T>>, ChannelReceiver<HashSet<T>>) {
+pub fn set_channel<T: Eq + Hash + Send + Sync>() -> (ChannelSender<HashSet<T>>, ChannelReceiver<HashSet<T>>) {
     channel::<HashSet<T>>()
 }
 
-pub fn queue_channel<T>() -> (ChannelSender<Vec<T>>, ChannelReceiver<Vec<T>>) {
+pub fn queue_channel<T: Send + Sync>() -> (ChannelSender<Vec<T>>, ChannelReceiver<Vec<T>>) {
     channel::<Vec<T>>()
 }
 
-pub fn singleton_channel<T>() -> (ChannelSender<Option<T>>, ChannelReceiver<Option<T>>) {
+pub fn singleton_channel<T: Send + Sync>() -> (ChannelSender<Option<T>>, ChannelReceiver<Option<T>>) {
     channel::<Option<T>>()
 }
 

src/main.rs

@@ -5,36 +5,82 @@ pub mod view;
 pub mod port;
 pub mod channel;
 
-use async_std::task;
+use {
-use async_std::prelude::*;
+    async_std::{
-use async_std::stream;
+        prelude::*, task, stream
+    },
+    std::{
+        sync::{Arc, RwLock}
+    },
+    cgmath::{Vector2},
+    crate::{
+        view::{View, Observer},
+        port::{InnerViewPort, OuterViewPort}
+    }
+};
 
-use std::sync::{Arc, RwLock};
+struct SingletonBuffer<T: Clone + Eq + Send + Sync + 'static> {
-use cgmath::Vector2;
+    data: Arc<RwLock<Option<T>>>,
+    port: InnerViewPort<(), T>
+}
 
-use view::{View, Observer};
+impl<T: Clone + Eq + Send + Sync + 'static> SingletonBuffer<T> {
-use port::{ViewPortIn, ViewPortOut};
+    fn new(
+        port: InnerViewPort<(), T>
+    ) -> Self {
+        let data = Arc::new(RwLock::new(None));
+
+        port.set_view_fn({
+            let data = data.clone();
+            move |new_val| data.read().unwrap().clone()
+        });
+
+        SingletonBuffer {
+            data,
+            port
+        }
+    }
+
+    fn update(&mut self, new_value: T) {
+        let mut data = self.data.write().unwrap();
+        if *data != Some(new_value.clone()) {
+            *data = Some(new_value);
+            drop(data);
+            self.port.notify(());
+        }
+    }
+}
 
 #[async_std::main]
 async fn main() {
-    let (view_in, mut view_out) = port::view_port::<usize, char>();
+    let view_port = port::ViewPort::<(), char>::new();
 
-    let mut observer_stream = view_in.stream().map({
+    let mut buf = SingletonBuffer::new(view_port.inner());
-        let view = view_in.clone();
+
-        move |idx| (idx, view.view(idx).unwrap())
+    let view = view_port.outer().get_view();
+    let mut stream = view_port.outer().stream().map({
+        move |_| view.read().unwrap().as_ref().unwrap().view(()).unwrap()
     });
 
-    let fut = task::spawn(async move {
+    let fut = task::spawn({
-        while let Some((idx, val)) = observer_stream.next().await {
+        async move {
-            println!("view[{}] = {}", idx, val);
+            while let Some(val) = stream.next().await {
+                println!("{}", val);
+            }
+            println!("end print task");
         }
     });
 
-    view_out.set_view_fn(|idx| Some(if idx % 2 == 0 { 'λ' } else { 'y' }) );
+    buf.update('a');
+    buf.update('b');
+    task::sleep(std::time::Duration::from_secs(1)).await;
+    buf.update('c');
+    buf.update('d');
+    task::sleep(std::time::Duration::from_secs(1)).await;
+    buf.update('e');
 
-    view_out.notify(1);
+    drop(buf);
-    view_out.notify(2);
+    drop(view_port);
-    view_out.notify(5);
 
     fut.await;
 }

src/port.rs

@@ -1,12 +1,13 @@
-
+use {
-use std::{
+    std::{
-    sync::{Arc, Weak, RwLock},
+        sync::{Arc, Weak, RwLock},
-    collections::HashSet,
+        collections::HashSet,
-    hash::Hash
+        hash::Hash,
-};
+    },
-use crate::{
+    crate::{
-    view::{View, Observer, FnView, FnObserver},
+        view::{View, Observer, FnView, FnObserver},
-    channel::{ChannelReceiver}
+        channel::{ChannelReceiver}
+    }
 };
 
                     /*\
@@ -14,62 +15,77 @@ use crate::{
                  View Port
 <<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
                     \*/
-
+#[derive(Clone)]
-pub struct ViewPort<K, V> {
+pub struct ViewPort<K: Send + Sync + 'static, V: Send + Sync + 'static> {
-    view: Option<Arc<dyn View<Key = K, Value = V> + Send + Sync>>,
+    view: Arc<RwLock<Option<Arc<dyn View<Key = K, Value = V>>>>>,
-    observers: Vec<Arc<RwLock<dyn Observer<Msg = K> + Send + Sync>>>
+    observers: Arc<RwLock<Vec<Arc<dyn Observer<Msg = K>>>>>
 }
 
-pub fn view_port<K, V>() -> (ViewPortIn<K, V>, ViewPortOut<K, V>) {
+impl<K, V> ViewPort<K, V>
-    let state = Arc::new(RwLock::new(ViewPort{ view: None, observers: Vec::new() }));
+where K: Send + Sync + 'static,
-    (ViewPortIn(state.clone()), ViewPortOut(state))
+      V: Send + Sync + 'static {
+    pub fn new() -> Self {
+        ViewPort {
+            view: Arc::new(RwLock::new(None)),
+            observers: Arc::new(RwLock::new(Vec::new()))
+        }
+    }
+
+    pub fn set_view(&self, view: Arc<dyn View<Key = K, Value = V>>) {
+        *self.view.write().unwrap() = Some(view);
+    }
+
+    pub fn add_observer(&self, observer: Arc<dyn Observer<Msg = K>>) {
+        self.observers.write().unwrap().push(observer);
+    }
+
+    pub fn inner(&self) -> InnerViewPort<K, V> {
+        InnerViewPort(ViewPort{ view: self.view.clone(), observers: self.observers.clone() })
+    }
+
+    pub fn outer(&self) -> OuterViewPort<K, V> {
+        OuterViewPort(ViewPort{ view: self.view.clone(), observers: self.observers.clone() })
+    }
 }
 
 //<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
 
 #[derive(Clone)]
-pub struct ViewPortIn<K, V>(Arc<RwLock<ViewPort<K, V>>>);
+pub struct InnerViewPort<K: Send + Sync + 'static, V: Send + Sync + 'static>(ViewPort<K, V>);
-impl<K: Send + Sync + 'static, V> ViewPortIn<K, V> {
+
-    pub fn add_observer(&self, observer: Arc<RwLock<dyn Observer<Msg = K> + Send + Sync>>) {
+#[derive(Clone)]
-        self.0
+pub struct OuterViewPort<K: Send + Sync + 'static, V: Send + Sync + 'static>(ViewPort<K, V>);
-            .write().unwrap()
+
-            .observers
+//<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
-            .push(observer);
+
+impl<K: Send + Sync + 'static, V: Send + Sync + 'static> OuterViewPort<K, V> {
+    pub fn get_view(&self) -> Arc<RwLock<Option<Arc<dyn View<Key = K, Value = V>>>>> {
+        self.0.view.clone()
     }
 
-    pub fn add_observer_fn(&self, obs_fn: impl FnMut(K) + Send + Sync + 'static) {
+    pub fn add_observer(self, observer: Arc<dyn Observer<Msg = K>>) -> Arc<RwLock<Option<Arc<dyn View<Key = K, Value = V>>>>> {
-        self.add_observer(Arc::new(RwLock::new(FnObserver::new(obs_fn))));
+        self.0.add_observer(observer);
+        self.0.view
+    }
+
+    pub fn add_observer_fn(self, obs_fn: impl Fn(K) + Send + Sync + 'static) -> Arc<RwLock<Option<Arc<dyn View<Key = K, Value = V>>>>> {
+        self.add_observer(Arc::new(FnObserver::new(obs_fn)))
     }
 }
 
-impl<K: Eq + Hash + Send + Sync + 'static, V> ViewPortIn<K, V> {
+impl<K: Eq + Hash + Send + Sync + 'static, V: Send + Sync + 'static> OuterViewPort<K, V> {
     pub fn stream(&self) -> ChannelReceiver<HashSet<K>> {
         let (s, r) = crate::channel::set_channel();
-        self.add_observer(Arc::new(RwLock::new(s)));
+        self.0.add_observer(Arc::new(s));
         r
     }
 }
 
-impl<K, V> View for ViewPortIn<K, V> {
-    type Key = K;
-    type Value = V;
-
-    fn view(&self, key: K) -> Option<V> {
-        if let Some(view) = self.0.read().unwrap().view.as_ref() {
-            view.view(key)
-        } else {
-            println!("Warning: trying to access InPort with uninitialized View!");
-            None
-        }
-    }
-}
-
 //<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
 
-pub struct ViewPortOut<K, V>(Arc<RwLock<ViewPort<K, V>>>);
+impl<K: Send + Sync + 'static, V: Send + Sync + 'static> InnerViewPort<K, V> {
-impl<K: Send + Sync + 'static, V: Send + Sync + 'static> ViewPortOut<K, V> {
     pub fn set_view(&self, view: Arc<dyn View<Key = K, Value = V> + Send + Sync>) {
-        self.0.write().unwrap().view = Some(view);
+        *self.0.view.write().unwrap() = Some(view);
     }
 
     pub fn set_view_fn(&self, view_fn: impl Fn(K) -> Option<V> + Send + Sync + 'static) {
@@ -77,13 +93,14 @@ impl<K: Send + Sync + 'static, V: Send + Sync + 'static> ViewPortOut<K, V> {
     }
 }
 
-impl<K, V> Observer for ViewPortOut<K, V>
+impl<K, V> Observer for InnerViewPort<K, V>
-where K: Clone {
+where K: Clone + Send + Sync + 'static,
+      V: Send + Sync + 'static {
     type Msg = K;
 
-    fn notify(&mut self, msg: K) {
+    fn notify(&self, msg: K) {
-        for observer in self.0.read().unwrap().observers.iter() {
+        for observer in self.0.observers.read().unwrap().iter() {
-            observer.write().unwrap().notify(msg.clone());
+            observer.notify(msg.clone());
         }
     }    
 }

src/view.rs

@@ -1,17 +1,17 @@
 
 //<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
 
-pub trait View {
+pub trait View : Send + Sync {
     type Key;
     type Value;
 
     fn view(&self, key: Self::Key) -> Option<Self::Value>;
 }
 
-pub trait Observer {
+pub trait Observer : Send + Sync {
     type Msg;
 
-    fn notify(&mut self, key: Self::Msg);
+    fn notify(&self, key: Self::Msg);
 }
 
 //<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
@@ -29,14 +29,19 @@ pub trait GridObserver = Observer<Msg = Vector2<i16>>;
 
 //<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
 
-pub struct FnView<K, V, F: Fn(K) -> Option<V>> {
+pub struct FnView<K, V, F>
+where K: Send + Sync,
+      V: Send + Sync,
+      F: Fn(K) -> Option<V> + Send + Sync {
     f: F,
     _phantom0: std::marker::PhantomData<K>,
     _phantom1: std::marker::PhantomData<V>
 }
 
 impl<K, V, F> FnView<K, V, F>
-where F: Fn(K) -> Option<V> {
+where K: Send + Sync,
+      V: Send + Sync,
+      F: Fn(K) -> Option<V> + Send + Sync {
     pub fn new(f: F) -> Self {
         FnView {
             f,
@@ -47,7 +52,9 @@ where F: Fn(K) -> Option<V> {
 }
 
 impl<K, V, F> View for FnView<K, V, F>
-where F: Fn(K) -> Option<V> {
+where K: Send + Sync,
+      V: Send + Sync,
+      F: Fn(K) -> Option<V> + Send + Sync {
     type Key = K;
     type Value = V;
 
@@ -58,13 +65,16 @@ where F: Fn(K) -> Option<V> {
 
 //<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
 
-pub struct FnObserver<T, F: FnMut(T)> {
+pub struct FnObserver<T, F>
+where T: Send + Sync,
+      F: Fn(T) + Send + Sync {
     f: F,
     _phantom: std::marker::PhantomData<T>
 }
 
 impl<T, F> FnObserver<T, F>
-where F: FnMut(T) {
+where T: Send + Sync,
+      F: Fn(T) + Send + Sync {
     pub fn new(f: F) -> Self {
         FnObserver {
             f,
@@ -74,11 +84,75 @@ where F: FnMut(T) {
 }
 
 impl<T, F> Observer for FnObserver<T, F>
-where F: FnMut(T) {
+where T: Send + Sync,
+      F: Fn(T) + Send + Sync {
     type Msg = T;
 
-    fn notify(&mut self, key: T) {
+    fn notify(&self, msg: T) {
-        (self.f)(key);
+        (self.f)(msg);
+    }
+}
+
+//<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
+
+impl<T: View> View for Option<T> {
+    type Key = T::Key;
+    type Value = T::Value;
+
+    fn view(&self, key: T::Key) -> Option<T::Value> {
+        if let Some(view) = self.as_ref() {
+            view.view(key)
+        } else {
+            None
+        }
+    }
+}
+
+impl<T: Observer> Observer for Option<T> {
+    type Msg = T::Msg;
+
+    fn notify(&self, msg: T::Msg) {
+        if let Some(obs) = self.as_ref() {
+            obs.notify(msg);
+        }
+    }
+}
+
+//<<<<>>>><<>><><<>><<<*>>><<>><><<>><<<<>>>>
+
+impl<T: View> View for std::sync::RwLock<T> {
+    type Key = T::Key;
+    type Value = T::Value;
+
+    fn view(&self, key: T::Key) -> Option<T::Value> {
+        self.read().unwrap().view(key)
+    }
+}
+
+impl<T: Observer> Observer for std::sync::RwLock<T> {
+    type Msg = T::Msg;
+
+    fn notify(&self, msg: T::Msg) {
+        self.read().unwrap().notify(msg)
+    }
+}
+
+use std::ops::Deref;
+
+impl<T: View> View for std::sync::Arc<T> {
+    type Key = T::Key;
+    type Value = T::Value;
+
+    fn view(&self, key: T::Key) -> Option<T::Value> {
+        self.deref().view(key)
+    }
+}
+
+impl<T: Observer> Observer for std::sync::Arc<T> {
+    type Msg = T::Msg;
+
+    fn notify(&self, msg: T::Msg) {
+        self.deref().notify(msg)
     }
 }