Core functions

crates/hd-bus/src/lib.rs

@@ -1,6 +1,4 @@
-use std::{
-    collections::{HashMap, VecDeque}, sync::RwLock, thread::{self, JoinHandle}, time::Duration,
-};
+use std::sync::{Arc, RwLock};
 
 pub type EventId = u64;
 pub type SubscriptionId = u64;
@@ -10,75 +8,173 @@ pub enum HError {
 }
 
 pub struct Event {
-    id: u64,
-    topic_name: String,
-    data_utf8: Vec<u8>,
+    id: EventId,
+    pub data: EventData,
 }
 
+pub struct EventData {
+    pub subject: String,
+    pub data_utf8: Vec<u8>,
+}
+
+// Add Event              event_bus scope W
+// Add subscription       event_bus scope W
+// Poll subscription      event_bus scope R + subscription scope R
+// Ack subscription       event_bus scope R + subscription scope W
+// Remove subscription    event_bus scope W
+// Cleanup Event          event_bus scope W
 pub struct EventBus {
-    queue: RwLock<VecDeque<Event>>,
-    subscriptions: RwLock<Vec<Subscription>>,
+    inner: Arc<RwLock<EventBusInner>>,
 }
 
-pub struct SubscriptionRequest {
-    topic_pattern: String,
+struct EventBusInner {
+    next_event_id: EventId,
+    events: Vec<Arc<Event>>,
+    subscriptions: RwLock<SubscriptionsInner>,
+}
+
+impl EventBusInner {
+    fn new() -> Self {
+        let subscriptions = SubscriptionsInner {
+            next_subscription_id: 0,
+            subscriptions: Vec::new(),
+        };
+
+        Self {
+            next_event_id: 0,
+            events: Vec::new(),
+            subscriptions: RwLock::new(subscriptions),
+        }
+    }
+}
+
+struct SubscriptionsInner {
+    next_subscription_id: SubscriptionId,
+    subscriptions: Vec<Subscription>,
 }
 
 pub struct Subscription {
-    id: u64,
-    request: SubscriptionRequest,
+    id: SubscriptionId,
+    // Return events exclusive of the cursor on a poll
+    cursor: EventId,
+}
+
+pub struct PollResult {
+    pub events: Vec<Arc<Event>>,
+    pub cursor_end: EventId,
 }
 
 impl EventBus {
     pub fn new() -> Self {
         Self {
-            queue: RwLock::new(VecDeque::new()),
-            subscriptions: RwLock::new(Vec::new()),
+            inner: Arc::new(RwLock::new(EventBusInner::new())),
         }
     }
 
-    pub fn publish(&self, evt: Event) -> Result<(), HError> {
-        let mut queue = self
-            .queue
+    pub fn publish(&self, evt: EventData) -> Result<(), HError> {
+        let mut bus = self
+            .inner
             .write()
             .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
-        queue.push_back(evt);
+
+        let next_event_id = &mut bus.next_event_id;
+        let event = Event {
+            id: *next_event_id,
+            data: evt,
+        };
+
+        *next_event_id += 1;
+
+        let events = &mut bus.events;
+        events.push(Arc::new(event));
         Ok(())
     }
 
-    pub fn subscribe(&self, request: SubscriptionRequest) -> Result<(), HError> {
-        let mut subscriptions = self
+    pub fn subscribe(&self) -> Result<SubscriptionId, HError> {
+        let bus = self
+            .inner
+            .write()
+            .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+
+        let subscriptions = &mut bus
             .subscriptions
             .write()
             .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
 
-        let last_id = subscriptions.last().map(|s| s.id).unwrap_or(0);
+        let next_subscription_id = &mut subscriptions.next_subscription_id;
         let subscription = Subscription {
-            id: last_id + 1,
-            request,
+            id: *next_subscription_id,
+            cursor: 0,
         };
-        subscriptions.push(subscription);
+
+        *next_subscription_id += 1;
+
+        let ret = subscription.id;
+        subscriptions.subscriptions.push(subscription);
+        Ok(ret)
+    }
+
+    pub fn poll(&self, id: SubscriptionId, max: usize) -> Result<PollResult, HError> {
+        let bus = self
+            .inner
+            .read()
+            .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+
+        let mut cursor = 0;
+
+        {
+            let subscriptions = &mut bus
+                .subscriptions
+                .read()
+                .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+
+            for sub in &subscriptions.subscriptions {
+                if sub.id == id {
+                    cursor = sub.cursor;
+                    break;
+                }
+            }
+        }
+
+        let mut events = Vec::new();
+        let mut cursor_end = cursor;
+        for evt in &bus.events {
+            if evt.id >= cursor {
+                events.push(evt.clone());
+                cursor_end = evt.id;
+                if events.len() >= max {
+                    break;
+                }
+            }
+        }
+
+        Ok(PollResult { events, cursor_end })
+    }
+
+    pub fn ack(&self, id: SubscriptionId, cursor: EventId) -> Result<(), HError> {
+        let bus = self
+            .inner
+            .read()
+            .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+
+        {
+            let subscriptions = &mut bus
+                .subscriptions
+                .write()
+                .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+
+            for sub in &mut subscriptions.subscriptions {
+                if sub.id == id {
+                    sub.cursor = cursor;
+                    break;
+                }
+            }
+        }
+
         Ok(())
     }
 }
 
-struct EventBusProcessor {
-    last_id: u64,
-    subscriptions_queue: RwLock<HashMap<SubscriptionId, EventId>>,
-}
-
-impl EventBusProcessor {
-    fn start() -> JoinHandle<u64> {
-        thread::spawn(|| {
-            loop {
-                thread::sleep(Duration::from_secs(1));
-            }
-
-            return 0;
-        })
-    }
-}
-
 #[cfg(test)]
 mod tests {
     #[test]