WIP

crates/hd-bus/src/lib.rs

@@ -1,7 +1,10 @@
 use std::{
-    collections::VecDeque, sync::{RwLock, mpsc::{Receiver, Sender}},
+    collections::{HashMap, VecDeque}, sync::RwLock, thread::{self, JoinHandle}, time::Duration,
 };
 
+pub type EventId = u64;
+pub type SubscriptionId = u64;
+
 pub enum HError {
     BusLockPoisoned(String),
 }
@@ -15,26 +18,15 @@ pub struct Event {
 pub struct EventBus {
     queue: RwLock<VecDeque<Event>>,
     subscriptions: RwLock<Vec<Subscription>>,
-    processor: Sender<EventProcessRequest>,
 }
 
-pub trait SubscriptionEndpoint {
-    fn deliver(&self, evt: &Event);
-}
-
-struct ConsoleEndpoint;
-impl SubscriptionEndpoint for ConsoleEndpoint {
-    fn deliver(&self, evt: &Event) {
-        let data = String::from_utf8_lossy(&evt.data_utf8);
-        println!("({}, {}, {})", evt.id, evt.topic_name, data);
-    }
+pub struct SubscriptionRequest {
+    topic_pattern: String,
 }
 
 pub struct Subscription {
-    topic_pattern: String,
-    /// from_id is inclusive
-    from_id: u64,
-    endpoint: Box<dyn SubscriptionEndpoint>,
+    id: u64,
+    request: SubscriptionRequest,
 }
 
 impl EventBus {
@@ -54,23 +46,37 @@ impl EventBus {
         Ok(())
     }
 
-    pub fn subscribe(&self, request: Subscription) -> Result<(), HError> {
+    pub fn subscribe(&self, request: SubscriptionRequest) -> Result<(), HError> {
         let mut subscriptions = self
             .subscriptions
             .write()
             .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
-        subscriptions.push(request);
+
+        let last_id = subscriptions.last().map(|s| s.id).unwrap_or(0);
+        let subscription = Subscription {
+            id: last_id + 1,
+            request,
+        };
+        subscriptions.push(subscription);
         Ok(())
     }
 }
 
-struct EventProcessRequest {
-    id: u64,
+struct EventBusProcessor {
+    last_id: u64,
+    subscriptions_queue: RwLock<HashMap<SubscriptionId, EventId>>,
 }
 
-struct EventBusProcessor {
-    last_processed_id: u64,
-    work: Receiver<u64>,
+impl EventBusProcessor {
+    fn start() -> JoinHandle<u64> {
+        thread::spawn(|| {
+            loop {
+                thread::sleep(Duration::from_secs(1));
+            }
+
+            return 0;
+        })
+    }
 }
 
 #[cfg(test)]