event_graph abstraction, making the event action generic

bin/ircd2/src/crypto.rs

@@ -24,9 +24,10 @@ use crypto_box::{
 };
 use rand::rngs::OsRng;
 
-use darkfi::event_graph::PrivMsgEvent;
-
-use crate::settings::{ChannelInfo, ContactInfo};
+use crate::{
+    privmsg::PrivMsgEvent,
+    settings::{ChannelInfo, ContactInfo},
+};
 
 #[derive(serde::Serialize)]
 pub struct KeyPair {

bin/ircd2/src/irc/client.rs

@@ -26,17 +26,13 @@ use futures::{
 
 use log::{debug, error, info, warn};
 
-use darkfi::{
-    event_graph::{model::Event, EventAction, PrivMsgEvent},
-    system::Subscription,
-    Error, Result,
-};
+use darkfi::{event_graph::model::Event, system::Subscription, Error, Result};
 
 use crate::{
     crypto::{decrypt_privmsg, decrypt_target, encrypt_privmsg},
     settings,
     settings::RPL,
-    ChannelInfo,
+    ChannelInfo, PrivMsgEvent,
 };
 
 use super::{ClientSubMsg, IrcConfig, NotifierMsg};
@@ -53,7 +49,7 @@ pub struct IrcClient<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> {
     server_notifier: smol::channel::Sender<(NotifierMsg, u64)>,
     subscription: Subscription<ClientSubMsg>,
 
-    missed_events: Arc<Mutex<Vec<Event>>>,
+    missed_events: Arc<Mutex<Vec<Event<PrivMsgEvent>>>>,
 }
 
 impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
@@ -64,7 +60,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
         irc_config: IrcConfig,
         server_notifier: smol::channel::Sender<(NotifierMsg, u64)>,
         subscription: Subscription<ClientSubMsg>,
-        missed_events: Arc<Mutex<Vec<Event>>>,
+        missed_events: Arc<Mutex<Vec<Event<PrivMsgEvent>>>>,
     ) -> Self {
         Self {
             write_stream,
@@ -565,13 +561,10 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
         hash_vec.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
 
         for event in hash_vec {
-            match event.action {
-                EventAction::PrivMsg(mut m) => {
-                    if let Err(e) = self.process_msg(&mut m).await {
-                        error!("[CLIENT {}] Process msg: {}", self.address, e);
-                        break
-                    }
-                }
+            let mut action = event.action.clone();
+            if let Err(e) = self.process_msg(&mut action).await {
+                error!("[CLIENT {}] Process msg: {}", self.address, e);
+                continue
             }
         }
         Ok(())

bin/ircd2/src/irc/mod.rs

@@ -32,7 +32,6 @@ use darkfi::{
         model::{Event, EventId, ModelPtr},
         protocol_event::{Seen, SeenPtr, UnreadEventsPtr},
         view::ViewPtr,
-        EventAction, PrivMsgEvent,
     },
     net::P2pPtr,
     system::SubscriberPtr,
@@ -40,7 +39,10 @@ use darkfi::{
     Error, Result,
 };
 
-use crate::settings::{Args, ChannelInfo, ContactInfo};
+use crate::{
+    settings::{Args, ChannelInfo, ContactInfo},
+    PrivMsgEvent,
+};
 
 mod client;
 
@@ -114,21 +116,21 @@ pub enum NotifierMsg {
 pub struct IrcServer {
     settings: Args,
     p2p: P2pPtr,
-    model: ModelPtr,
-    view: ViewPtr,
-    unread_events: UnreadEventsPtr,
+    model: ModelPtr<PrivMsgEvent>,
+    view: ViewPtr<PrivMsgEvent>,
+    unread_events: UnreadEventsPtr<PrivMsgEvent>,
     clients_subscriptions: SubscriberPtr<ClientSubMsg>,
     seen: SeenPtr<EventId>,
-    missed_events: Arc<Mutex<Vec<Event>>>,
+    missed_events: Arc<Mutex<Vec<Event<PrivMsgEvent>>>>,
 }
 
 impl IrcServer {
     pub async fn new(
         settings: Args,
         p2p: P2pPtr,
-        model: ModelPtr,
-        view: ViewPtr,
-        unread_events: UnreadEventsPtr,
+        model: ModelPtr<PrivMsgEvent>,
+        view: ViewPtr<PrivMsgEvent>,
+        unread_events: UnreadEventsPtr<PrivMsgEvent>,
         clients_subscriptions: SubscriberPtr<ClientSubMsg>,
     ) -> Result<Self> {
         let seen = Seen::new();
@@ -177,9 +179,9 @@ impl IrcServer {
     }
 
     async fn listen_to_view(
-        view: ViewPtr,
+        view: ViewPtr<PrivMsgEvent>,
         seen: SeenPtr<EventId>,
-        missed_events: Arc<Mutex<Vec<Event>>>,
+        missed_events: Arc<Mutex<Vec<Event<PrivMsgEvent>>>>,
         clients_subscriptions: SubscriberPtr<ClientSubMsg>,
     ) -> Result<()> {
         loop {
@@ -190,9 +192,8 @@ impl IrcServer {
 
             missed_events.lock().await.push(event.clone());
 
-            let msg = match event.action {
-                EventAction::PrivMsg(x) => x,
-            };
+            let msg = event.action.clone();
+
             clients_subscriptions.notify(ClientSubMsg::Privmsg(msg)).await;
         }
     }
@@ -200,9 +201,9 @@ impl IrcServer {
     /// Start listening to msgs from irc clients
     pub async fn listen_to_msgs(
         p2p: P2pPtr,
-        model: ModelPtr,
+        model: ModelPtr<PrivMsgEvent>,
         seen: SeenPtr<EventId>,
-        unread_events: UnreadEventsPtr,
+        unread_events: UnreadEventsPtr<PrivMsgEvent>,
         recv: smol::channel::Receiver<(NotifierMsg, u64)>,
         clients_subscriptions: SubscriberPtr<ClientSubMsg>,
     ) -> Result<()> {
@@ -214,7 +215,7 @@ impl IrcServer {
                 NotifierMsg::Privmsg(msg) => {
                     let event = Event {
                         previous_event_hash: prev,
-                        action: EventAction::PrivMsg(msg.clone()),
+                        action: msg.clone(),
                         timestamp: get_current_time(),
                         read_confirms: 0,
                     };

bin/ircd2/src/main.rs

@@ -50,11 +50,12 @@ use crate::{
     crypto::KeyPair,
     // events_queue::EventsQueue,
     irc::IrcServer,
+    privmsg::PrivMsgEvent,
+    // view::View,
     // model::Model,
     // protocol_event::{ProtocolEvent, Seen, UnreadEvents},
     rpc::JsonRpcInterface,
     settings::{Args, ChannelInfo, CONFIG_FILE, CONFIG_FILE_CONTENTS},
-    // view::View,
 };
 
 async_daemonize!(realmain);
@@ -83,7 +84,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'_>>) -> Result<(
     ////////////////////
     // Initialize the base structures
     ////////////////////
-    let events_queue = EventsQueue::new();
+    let events_queue = EventsQueue::<PrivMsgEvent>::new();
     let model = Arc::new(Mutex::new(Model::new(events_queue.clone())));
     let view = Arc::new(Mutex::new(View::new(events_queue)));
     let model_clone = model.clone();

bin/ircd2/src/privmsg.rs

@@ -16,6 +16,28 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-// use std::io;
+use darkfi::event_graph::EventMsg;
+use darkfi_serial::{SerialDecodable, SerialEncodable};
 
-// use darkfi_serial::{Decodable, Encodable, ReadExt, SerialDecodable, SerialEncodable};
+#[derive(SerialEncodable, SerialDecodable, Clone, Debug)]
+pub struct PrivMsgEvent {
+    pub nick: String,
+    pub msg: String,
+    pub target: String,
+}
+
+impl std::string::ToString for PrivMsgEvent {
+    fn to_string(&self) -> String {
+        format!(":{}!anon@dark.fi PRIVMSG {} :{}\r\n", self.nick, self.target, self.msg)
+    }
+}
+
+impl EventMsg for PrivMsgEvent {
+    fn new() -> Self {
+        Self {
+            nick: "root".to_string(),
+            msg: "Let there be dark".to_string(),
+            target: "root".to_string(),
+        }
+    }
+}

src/event_graph/events_queue.rs

@@ -17,24 +17,33 @@
  */
 
 use async_std::sync::Arc;
+use darkfi_serial::{Decodable, Encodable};
 
 use crate::{event_graph::model::Event, Error, Result};
 
-pub type EventsQueuePtr = Arc<EventsQueue>;
+use super::EventMsg;
 
-pub struct EventsQueue(smol::channel::Sender<Event>, smol::channel::Receiver<Event>);
+pub type EventsQueuePtr<T> = Arc<EventsQueue<T>>;
 
-impl EventsQueue {
-    pub fn new() -> EventsQueuePtr {
+pub struct EventsQueue<T: Send + Sync>(
+    smol::channel::Sender<Event<T>>,
+    smol::channel::Receiver<Event<T>>,
+);
+
+impl<T> EventsQueue<T>
+where
+    T: Send + Sync + Encodable + Decodable + Clone + EventMsg,
+{
+    pub fn new() -> EventsQueuePtr<T> {
         let (sn, rv) = smol::channel::unbounded();
         Arc::new(Self(sn, rv))
     }
 
-    pub async fn fetch(&self) -> Result<Event> {
+    pub async fn fetch(&self) -> Result<Event<T>> {
         self.1.recv().await.map_err(Error::from)
     }
 
-    pub async fn dispatch(&self, event: &Event) -> Result<()> {
+    pub async fn dispatch(&self, event: &Event<T>) -> Result<()> {
         self.0.send(event.clone()).await.map_err(Error::from)
     }
 }

src/event_graph/mod.rs

@@ -16,54 +16,13 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::io;
-
-use darkfi_serial::{Decodable, Encodable, ReadExt, SerialDecodable, SerialEncodable};
-
 pub mod events_queue;
 pub mod model;
 pub mod protocol_event;
 pub mod view;
 
-#[derive(SerialEncodable, SerialDecodable, Clone)]
-pub struct PrivMsgEvent {
-    pub nick: String,
-    pub msg: String,
-    pub target: String,
-}
-
-#[derive(Clone)]
-pub enum EventAction {
-    PrivMsg(PrivMsgEvent),
-}
-
-impl std::string::ToString for PrivMsgEvent {
-    fn to_string(&self) -> String {
-        format!(":{}!anon@dark.fi PRIVMSG {} :{}\r\n", self.nick, self.target, self.msg)
-    }
-}
-
-impl Encodable for EventAction {
-    fn encode<S: io::Write>(&self, mut s: S) -> core::result::Result<usize, io::Error> {
-        match self {
-            Self::PrivMsg(event) => {
-                let mut len = 0;
-                len += 0u8.encode(&mut s)?;
-                len += event.encode(s)?;
-                Ok(len)
-            }
-        }
-    }
-}
-
-impl Decodable for EventAction {
-    fn decode<D: io::Read>(mut d: D) -> core::result::Result<Self, io::Error> {
-        let type_id = d.read_u8()?;
-        match type_id {
-            0 => Ok(Self::PrivMsg(PrivMsgEvent::decode(d)?)),
-            _ => Err(io::Error::new(io::ErrorKind::Other, "Bad type ID byte for Event")),
-        }
-    }
+pub trait EventMsg {
+    fn new() -> Self;
 }
 
 pub fn get_current_time() -> u64 {

src/event_graph/model.rs

@@ -16,31 +16,34 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{cmp::Ordering, collections::HashMap, fmt};
+use std::{cmp::Ordering, collections::HashMap, fmt::Debug};
 
 use async_std::sync::{Arc, Mutex};
-use darkfi_serial::{Encodable, SerialDecodable, SerialEncodable};
+use darkfi_serial::{Decodable, Encodable, SerialDecodable, SerialEncodable};
 use log::error;
 use ripemd::{Digest, Ripemd256};
 
 use crate::event_graph::events_queue::EventsQueuePtr;
 
-use super::{EventAction, PrivMsgEvent};
+use super::EventMsg;
 
 pub type EventId = [u8; 32];
 
 const MAX_DEPTH: u32 = 300;
 const MAX_HEIGHT: u32 = 300;
 
-#[derive(SerialEncodable, SerialDecodable, Clone)]
-pub struct Event {
+#[derive(SerialEncodable, SerialDecodable, Clone, Debug)]
+pub struct Event<T: Send + Sync> {
     pub previous_event_hash: EventId,
-    pub action: EventAction,
+    pub action: T,
     pub timestamp: u64,
     pub read_confirms: u8,
 }
 
-impl Event {
+impl<T> Event<T>
+where
+    T: Send + Sync + Encodable + Decodable + Clone + EventMsg,
+{
     pub fn hash(&self) -> EventId {
         let mut bytes = Vec::new();
         let mut event_to_be_hashed = self.clone();
@@ -56,45 +59,34 @@ impl Event {
     }
 }
 
-impl fmt::Debug for Event {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match &self.action {
-            EventAction::PrivMsg(event) => {
-                write!(f, "PRIVMSG {}: {} ({})", event.nick, event.msg, self.timestamp)
-            }
-        }
-    }
-}
-
 #[derive(Debug, Clone)]
-struct EventNode {
+struct EventNode<T: Send + Sync> {
     // Only current root has this set to None
     parent: Option<EventId>,
-    event: Event,
+    event: Event<T>,
     children: Vec<EventId>,
 }
 
-pub type ModelPtr = Arc<Mutex<Model>>;
+pub type ModelPtr<T> = Arc<Mutex<Model<T>>>;
 
-pub struct Model {
+pub struct Model<T: Send + Sync + Debug> {
     // This is periodically updated so we discard old nodes
     current_root: EventId,
-    orphans: HashMap<EventId, Event>,
-    event_map: HashMap<EventId, EventNode>,
-    events_queue: EventsQueuePtr,
+    orphans: HashMap<EventId, Event<T>>,
+    event_map: HashMap<EventId, EventNode<T>>,
+    events_queue: EventsQueuePtr<T>,
 }
 
-impl Model {
-    pub fn new(events_queue: EventsQueuePtr) -> Self {
+impl<T> Model<T>
+where
+    T: Send + Sync + Encodable + Decodable + Clone + EventMsg + Debug,
+{
+    pub fn new(events_queue: EventsQueuePtr<T>) -> Self {
         let root_node = EventNode {
             parent: None,
             event: Event {
                 previous_event_hash: [0u8; 32],
-                action: EventAction::PrivMsg(PrivMsgEvent {
-                    nick: "root".to_string(),
-                    msg: "Let there be dark".to_string(),
-                    target: "root".to_string(),
-                }),
+                action: T::new(),
                 timestamp: 1674512021323,
                 read_confirms: 0,
             },
@@ -113,12 +105,12 @@ impl Model {
         self.find_head()
     }
 
-    pub async fn add(&mut self, event: Event) {
+    pub async fn add(&mut self, event: Event<T>) {
         self.orphans.insert(event.hash(), event);
         self.reorganize().await;
     }
 
-    pub fn is_orphan(&self, event: &Event) -> bool {
+    pub fn is_orphan(&self, event: &Event<T>) -> bool {
         !self.event_map.contains_key(&event.previous_event_hash)
     }
 
@@ -136,11 +128,11 @@ impl Model {
         leaves
     }
 
-    pub fn get_event(&self, event: &EventId) -> Option<Event> {
+    pub fn get_event(&self, event: &EventId) -> Option<Event<T>> {
         self.event_map.get(event).map(|en| en.event.clone())
     }
 
-    pub fn get_offspring(&self, event: &EventId) -> Vec<Event> {
+    pub fn get_offspring(&self, event: &EventId) -> Vec<Event<T>> {
         let mut offspring = vec![];
         let mut event = *event;
         let head = self.find_head();
@@ -410,22 +402,36 @@ mod tests {
     use super::*;
     use crate::event_graph::{events_queue::EventsQueue, get_current_time};
 
+    #[derive(SerialEncodable, SerialDecodable, Clone, Debug)]
+    pub struct PrivMsgEvent {
+        pub nick: String,
+        pub msg: String,
+        pub target: String,
+    }
+
+    impl std::string::ToString for PrivMsgEvent {
+        fn to_string(&self) -> String {
+            format!(":{}!anon@dark.fi PRIVMSG {} :{}\r\n", self.nick, self.target, self.msg)
+        }
+    }
+
+    impl EventMsg for PrivMsgEvent {
+        fn new() -> Self {
+            Self {
+                nick: "root".to_string(),
+                msg: "Let there be dark".to_string(),
+                target: "root".to_string(),
+            }
+        }
+    }
+
     fn create_message(
         previous_event_hash: EventId,
         nick: &str,
         msg: &str,
         timestamp: u64,
-    ) -> Event {
-        Event {
-            previous_event_hash,
-            action: EventAction::PrivMsg(PrivMsgEvent {
-                nick: nick.to_string(),
-                msg: msg.to_string(),
-                target: "".to_string(),
-            }),
-            timestamp,
-            read_confirms: 4,
-        }
+    ) -> Event<PrivMsgEvent> {
+        Event { previous_event_hash, action: PrivMsgEvent::new(), timestamp, read_confirms: 4 }
     }
 
     /* THIS IS FAILING
@@ -610,7 +616,7 @@ mod tests {
 
     #[test]
     fn test_event_hash() {
-        let events_queue = EventsQueue::new();
+        let events_queue = EventsQueue::<PrivMsgEvent>::new();
         let model = Model::new(events_queue);
         let root_id = model.current_root;
 

src/event_graph/protocol_event.rs

@@ -16,15 +16,18 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::collections::{HashMap, VecDeque};
+use std::{
+    collections::{HashMap, VecDeque},
+    fmt::Debug,
+};
 
 use async_std::sync::{Arc, Mutex};
 use async_trait::async_trait;
-use darkfi_serial::{SerialDecodable, SerialEncodable};
+use darkfi_serial::{Decodable, Encodable, SerialDecodable, SerialEncodable};
 use log::debug;
 use rand::{rngs::OsRng, RngCore};
 
-use super::get_current_time;
+use super::{get_current_time, EventMsg};
 use crate::{
     event_graph::model::{Event, EventId, ModelPtr},
     net,
@@ -103,15 +106,18 @@ impl<T: Eq + PartialEq + Clone> Seen<T> {
     }
 }
 
-pub type UnreadEventsPtr = Arc<Mutex<UnreadEvents>>;
+pub type UnreadEventsPtr<T> = Arc<Mutex<UnreadEvents<T>>>;
 
 #[derive(Debug)]
-pub struct UnreadEvents {
-    pub events: HashMap<EventId, Event>,
+pub struct UnreadEvents<T: Send + Sync> {
+    pub events: HashMap<EventId, Event<T>>,
 }
 
-impl UnreadEvents {
-    pub fn new() -> UnreadEventsPtr {
+impl<T> UnreadEvents<T>
+where
+    T: Send + Sync + Encodable + Decodable + Clone + EventMsg,
+{
+    pub fn new() -> UnreadEventsPtr<T> {
         Arc::new(Mutex::new(Self { events: HashMap::new() }))
     }
 
@@ -119,13 +125,13 @@ impl UnreadEvents {
         self.events.contains_key(key)
     }
 
-    fn _get(&self, key: &EventId) -> Option<Event> {
+    fn _get(&self, key: &EventId) -> Option<Event<T>> {
         self.events.get(key).cloned()
     }
 
     // Increase the read_confirms for an event, if it has exceeded the MAX_CONFIRM
     // then remove it from the hash_map and return Some(event), otherwise return None
-    fn inc_read_confirms(&mut self, key: &EventId) -> Option<Event> {
+    fn inc_read_confirms(&mut self, key: &EventId) -> Option<Event<T>> {
         let mut result = None;
 
         if let Some(event) = self.events.get_mut(key) {
@@ -142,7 +148,7 @@ impl UnreadEvents {
         result
     }
 
-    pub fn insert(&mut self, event: &Event) {
+    pub fn insert(&mut self, event: &Event<T>) {
         // prune expired events
         let mut prune_ids = vec![];
         for (id, e) in self.events.iter() {
@@ -158,37 +164,43 @@ impl UnreadEvents {
     }
 }
 
-pub struct ProtocolEvent {
+pub struct ProtocolEvent<T>
+where
+    T: Send + Sync + Encodable + Decodable + Debug + 'static,
+{
     jobsman: net::ProtocolJobsManagerPtr,
-    event_sub: net::MessageSubscription<Event>,
+    event_sub: net::MessageSubscription<Event<T>>,
     inv_sub: net::MessageSubscription<Inv>,
     getdata_sub: net::MessageSubscription<GetData>,
     syncevent_sub: net::MessageSubscription<SyncEvent>,
     p2p: net::P2pPtr,
     channel: net::ChannelPtr,
-    model: ModelPtr,
+    model: ModelPtr<T>,
     seen_event: SeenPtr<EventId>,
     seen_inv: SeenPtr<InvId>,
-    unread_events: UnreadEventsPtr,
+    unread_events: UnreadEventsPtr<T>,
 }
 
-impl ProtocolEvent {
+impl<T> ProtocolEvent<T>
+where
+    T: Send + Sync + Encodable + Decodable + Clone + EventMsg + Debug + 'static,
+{
     pub async fn init(
         channel: net::ChannelPtr,
         p2p: net::P2pPtr,
-        model: ModelPtr,
+        model: ModelPtr<T>,
         seen_event: SeenPtr<EventId>,
         seen_inv: SeenPtr<InvId>,
-        unread_events: UnreadEventsPtr,
+        unread_events: UnreadEventsPtr<T>,
     ) -> net::ProtocolBasePtr {
         let message_subsytem = channel.get_message_subsystem();
-        message_subsytem.add_dispatch::<Event>().await;
+        message_subsytem.add_dispatch::<Event<T>>().await;
         message_subsytem.add_dispatch::<Inv>().await;
         message_subsytem.add_dispatch::<GetData>().await;
         message_subsytem.add_dispatch::<SyncEvent>().await;
 
         let event_sub =
-            channel.clone().subscribe_msg::<Event>().await.expect("Missing Event dispatcher!");
+            channel.clone().subscribe_msg::<Event<T>>().await.expect("Missing Event dispatcher!");
 
         let inv_sub = channel.subscribe_msg::<Inv>().await.expect("Missing Inv dispatcher!");
 
@@ -330,14 +342,14 @@ impl ProtocolEvent {
         }
     }
 
-    async fn new_event(&self, event: &Event) -> Result<()> {
+    async fn new_event(&self, event: &Event<T>) -> Result<()> {
         let mut model = self.model.lock().await;
         model.add(event.clone()).await;
 
         Ok(())
     }
 
-    async fn send_inv(&self, event: &Event) -> Result<()> {
+    async fn send_inv(&self, event: &Event<T>) -> Result<()> {
         let id = OsRng.next_u64();
         self.p2p.broadcast(Inv { invs: vec![InvItem { id, hash: event.hash() }] }).await?;
 
@@ -351,7 +363,10 @@ impl ProtocolEvent {
 }
 
 #[async_trait]
-impl net::ProtocolBase for ProtocolEvent {
+impl<T> net::ProtocolBase for ProtocolEvent<T>
+where
+    T: Send + Sync + Encodable + Decodable + Clone + EventMsg + Debug,
+{
     async fn start(self: Arc<Self>, executor: Arc<smol::Executor<'_>>) -> Result<()> {
         debug!(target: "ircd", "ProtocolEvent::start() [START]");
         self.jobsman.clone().start(executor.clone());
@@ -369,7 +384,10 @@ impl net::ProtocolBase for ProtocolEvent {
     }
 }
 
-impl net::Message for Event {
+impl<T> net::Message for Event<T>
+where
+    T: Send + Sync + Decodable + Encodable + 'static,
+{
     fn name() -> &'static str {
         "event"
     }

src/event_graph/view.rs

@@ -17,6 +17,7 @@
  */
 
 use async_std::sync::{Arc, Mutex};
+use darkfi_serial::{Decodable, Encodable};
 use std::collections::HashMap;
 
 use crate::{
@@ -27,19 +28,24 @@ use crate::{
     Result,
 };
 
-pub type ViewPtr = Arc<Mutex<View>>;
+use super::EventMsg;
 
-pub struct View {
-    pub seen: HashMap<EventId, Event>,
-    pub events_queue: EventsQueuePtr,
+pub type ViewPtr<T> = Arc<Mutex<View<T>>>;
+
+pub struct View<T: Send + Sync> {
+    pub seen: HashMap<EventId, Event<T>>,
+    pub events_queue: EventsQueuePtr<T>,
 }
 
-impl View {
-    pub fn new(events_queue: EventsQueuePtr) -> Self {
+impl<T> View<T>
+where
+    T: Send + Sync + Encodable + Decodable + Clone + EventMsg,
+{
+    pub fn new(events_queue: EventsQueuePtr<T>) -> Self {
         Self { seen: HashMap::new(), events_queue }
     }
 
-    pub async fn process(&mut self) -> Result<Event> {
+    pub async fn process(&mut self) -> Result<Event<T>> {
         // loop {
         let new_event = self.events_queue.fetch().await?;
         Ok(new_event)