A good starting point

Author: bfil

.gitignore

@@ -8,7 +8,7 @@
 *.exe
 
 # Generated by Cargo
-/target/
+target/
 
 # Remove Cargo.lock from gitignore if creating an executable, leave it for libraries
 # More information here http://doc.crates.io/guide.html#cargotoml-vs-cargolock

TODO.md

@@ -0,0 +1,86 @@
+### Database
+
+- Use & Improve Config (Done)
+- Error Handling (Done)
+- Basic Authentication (Done)
+- Validation (Done)
+- Investigate If Tags Should Be Mandatory (Done)
+- Return Event ID On Publish (Done)
+- Consider Own Trait Over ToString/FromString Traits (Unnecessary)
+
+### Config
+
+- Use A Separate Config Struct For Each Components (Done)
+
+### Collection
+
+- Use & Improve Config (Done)
+- Use Subscription Strategy In Config (Done)
+- Error Handling (Done)
+- Improve Subscribe Method (Done)
+- Subscription Routing Strategy (Done)
+- Support Collection-Specific Config (Done)
+- Improve File Reader & File Writer Methods (Done)
+
+### Connection
+
+- Error Handling (Done)
+- Add Close Method (Done)
+
+### Event
+
+- Add Timestamp (Done)
+- Improve Serialization If Necessary (Done)
+- Method To Set ID And Timestamp Should Be Private (Unnecessary)
+
+### Subscription
+
+- Error Handling (Done)
+- Improve API (Done)
+- Add Current Events Queries (Done)
+
+### Writer
+
+- Error Handling (Done)
+- Change API To Disallow Event ID Changes If Necessary (Unnecessary)
+
+### Scanner
+
+- Use Config (Unnecessary)
+- Error Handling (Done)
+
+
+### Net
+
+- Move/Refactor Utility Functions (Done)
+
+### Protocol
+
+- Improve Event Stream (Done)
+
+### Server
+
+- Error Handling (Done)
+- Use Config (Done)
+- Improve Constructor & API (Done)
+
+### Handler
+
+- Error Handling (Done)
+- Improve Commands/Results Serialization (Done)
+- Move Protocol Into Its Own Module (Done)
+
+### Client
+
+- Error Handling (Done)
+- Improve Commands/Results Serialization (Done)
+- Add Close Method (Done)
+
+### General
+
+- Rename Components If Necessary (-)
+- Derive Relevant Traits For All Structs/Enums (-)
+- Implement Display For Relevant Structs/Enums (-)
+- Write Tests (-)
+- Write Docs (-)
+- Split Core/Server/Client Into Separate Crates (-)

exar-client/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "exar-client"
+version = "0.1.0"
+authors = ["Bruno Filippone <[email protected]>"]
+
+[dependencies]
+exar = { version = "0.1", path = "../exar-core" }
+exar-net = { version = "0.1", path = "../exar-net" }

exar-client/src/lib.rs

@@ -0,0 +1,74 @@
+extern crate exar;
+extern crate exar_net;
+
+use exar::*;
+use exar_net::*;
+
+use std::io::{Error as IoError, ErrorKind};
+use std::net::TcpStream;
+use std::sync::mpsc::channel;
+use std::thread;
+
+pub struct Client {
+    stream: Stream
+}
+
+impl Client {
+    pub fn connect(address: &str, collection_name: &str, username: Option<&str>, password: Option<&str>) -> Result<Client, DatabaseError> {
+        match TcpStream::connect(address) {
+            Ok(stream) => {
+                let mut stream = try!(Stream::new(stream));
+                let username = username.map(|u| u.to_owned());
+                let password = password.map(|p| p.to_owned());
+                let connection_message = TcpMessage::Connect(collection_name.to_owned(), username, password);
+                try!(stream.send_message(connection_message));
+                match stream.receive_message() {
+                    Ok(TcpMessage::Connected) => Ok(Client { stream: stream }),
+                    Ok(TcpMessage::Error(error)) => Err(error),
+                    Ok(_) => Err(DatabaseError::ConnectionError),
+                    Err(err) => Err(err)
+                }
+            },
+            Err(err) => Err(DatabaseError::IoError(err))
+        }
+    }
+
+    pub fn publish(&mut self, event: Event) -> Result<usize, DatabaseError> {
+        try!(self.stream.send_message(TcpMessage::Publish(event)));
+        match self.stream.receive_message() {
+            Ok(TcpMessage::Published(event_id)) => Ok(event_id),
+            Ok(TcpMessage::Error(error)) => Err(error),
+            Ok(_) => Err(DatabaseError::IoError(IoError::new(ErrorKind::InvalidData, UnexpectedTcpMessage))),
+            Err(err) => Err(err)
+        }
+    }
+
+    pub fn subscribe(&mut self, query: Query) -> Result<EventStream, DatabaseError> {
+        let subscribe_message = TcpMessage::Subscribe(query.live, query.offset, query.limit, query.tag);
+        self.stream.send_message(subscribe_message).and_then(|_| {
+            let (send, recv) = channel();
+            self.stream.try_clone().and_then(|cloned_stream| {
+                thread::spawn(move || {
+                    for message in cloned_stream.messages() {
+                        match message {
+                            Ok(TcpMessage::Event(event)) => match send.send(event) {
+                                Ok(_) => continue,
+                                Err(err) => println!("Unable to send event to the event stream: {}", err)
+                            },
+                            Ok(TcpMessage::EndOfStream) => (),
+                            Ok(TcpMessage::Error(error)) => println!("Received error from TCP stream: {:?}", error),
+                            Ok(message) => println!("Unexpected TCP message: {:?}", message),
+                            Err(err) => println!("Unable to read TCP message from stream: {:?}", err)
+                        };
+                        break
+                    }
+                });
+                Ok(EventStream::new(recv))
+            })
+        })
+    }
+
+    pub fn close(self) {
+        drop(self)
+    }
+}

exar-core/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "exar"
+version = "0.1.0"
+authors = ["Bruno Filippone <[email protected]>"]
+
+[dependencies]
+rand = "0.3"
+time = "0.1"
+rustc-serialize = { optional = true, version = "0.3" }
+serde = { optional = true, version = "0.6" }
+serde_macros = { optional = true, version = "0.6" }
+
+[features]
+default = ["rustc-serialize"]
+serde-serialization = ["serde", "serde_macros"]

exar-core/src/collection.rs

@@ -0,0 +1,82 @@
+use super::*;
+
+use rand;
+use rand::Rng;
+use std::sync::mpsc::channel;
+
+pub struct Collection {
+    log: Log,
+    scanners: Vec<Scanner>,
+    routing_strategy: RoutingStrategy,
+    writer: Writer
+}
+
+impl Collection {
+    pub fn new(collection_name: &str, config: CollectionConfig) -> Result<Collection, DatabaseError> {
+        let log = Log::new(&config.logs_path, collection_name);
+        Writer::new(log.clone()).and_then(|writer| {
+            let mut scanners = vec![];
+            for _ in 0..config.num_scanners {
+                let scanner = try!(Scanner::new(log.clone()));
+                scanners.push(scanner);
+            }
+            Ok(Collection {
+                log: log,
+                scanners: scanners,
+                routing_strategy: config.routing_strategy.clone(),
+                writer: writer
+            })
+        })
+    }
+
+    pub fn publish(&self, event: Event) -> Result<usize, DatabaseError> {
+        self.writer.store(event)
+    }
+
+    pub fn subscribe(&mut self, query: Query) -> Result<EventStream, DatabaseError> {
+        let (send, recv) = channel();
+        self.apply_strategy(Subscription::new(send, query)).and_then(|updated_strategy| {
+            self.routing_strategy = updated_strategy;
+            Ok(EventStream::new(recv))
+        })
+    }
+
+    pub fn drop(&self) -> Result<(), DatabaseError> {
+        self.drop_scanners();
+        match self.log.remove() {
+            Ok(()) => Ok(()),
+            Err(err) => Err(DatabaseError::IoError(err))
+        }
+    }
+
+    fn drop_scanners(&self) {
+        for scanner in &self.scanners {
+            drop(scanner);
+        }
+    }
+
+    fn apply_strategy(&mut self, subscription: Subscription) -> Result<RoutingStrategy, DatabaseError> {
+        match self.routing_strategy {
+            RoutingStrategy::Random => match rand::thread_rng().choose(&self.scanners) {
+                Some(random_scanner) => {
+                    random_scanner.handle_subscription(subscription).and_then(|_| {
+                        Ok(RoutingStrategy::Random)
+                    })
+                },
+                None => Err(DatabaseError::SubscriptionError)
+            },
+            RoutingStrategy::RoundRobin(ref last_index) => {
+                let mut new_index = 0;
+                if last_index + 1 < self.scanners.len() {
+                    new_index = last_index + 1;
+                }
+                match self.scanners.get(new_index) {
+                    Some(scanner) => scanner.handle_subscription(subscription).and_then(|_| {
+                        Ok(RoutingStrategy::RoundRobin(new_index))
+                    }),
+                    None => Err(DatabaseError::SubscriptionError)
+                }
+            }
+        }
+    }
+}

exar-core/src/config.rs

@@ -0,0 +1,46 @@
+use super::*;
+
+use std::collections::BTreeMap;
+
+#[cfg_attr(feature = "rustc-serialize", derive(RustcEncodable, RustcDecodable))]
+#[cfg_attr(feature = "serde-serialization", derive(Serialize, Deserialize))]
+#[derive(Clone, Debug)]
+pub struct DatabaseConfig  {
+    pub logs_path: String,
+    pub num_scanners: u8,
+    pub routing_strategy: RoutingStrategy,
+    pub collections: BTreeMap<String, CollectionConfig>
+}
+
+impl Default for DatabaseConfig {
+    fn default() -> DatabaseConfig {
+        DatabaseConfig {
+            logs_path: "".to_owned(),
+            num_scanners: 2,
+            routing_strategy: RoutingStrategy::default(),
+            collections: BTreeMap::new()
+        }
+    }
+}
+
+impl DatabaseConfig {
+    pub fn get_collection_config(&self, collection_name: &str) -> CollectionConfig {
+        match self.collections.get(collection_name) {
+            Some(collection_config) => collection_config.clone(),
+            None => CollectionConfig {
+                logs_path: self.logs_path.clone(),
+                num_scanners: self.num_scanners,
+                routing_strategy: self.routing_strategy.clone()
+            }
+        }
+    }
+}
+
+#[cfg_attr(feature = "rustc-serialize", derive(RustcEncodable, RustcDecodable))]
+#[cfg_attr(feature = "serde-serialization", derive(Serialize, Deserialize))]
+#[derive(Clone, Debug)]
+pub struct CollectionConfig {
+    pub logs_path: String,
+    pub num_scanners: u8,
+    pub routing_strategy: RoutingStrategy
+}

exar-core/src/connection.rs

@@ -0,0 +1,28 @@
+use super::*;
+
+use std::sync::{Arc, Mutex};
+
+#[derive(Clone)]
+pub struct Connection {
+    collection: Arc<Mutex<Collection>>
+}
+
+impl Connection {
+    pub fn new(collection: Arc<Mutex<Collection>>) -> Connection {
+        Connection {
+            collection: collection
+        }
+    }
+
+    pub fn publish(&self, event: Event) -> Result<usize, DatabaseError> {
+        self.collection.lock().unwrap().publish(event)
+    }
+
+    pub fn subscribe(&self, query: Query) -> Result<EventStream, DatabaseError> {
+        self.collection.lock().unwrap().subscribe(query)
+    }
+
+    pub fn close(self) {
+        drop(self)
+    }
+}