Merge pull request #50 from PoorRican/update-testing

Major update testing and improve error handling

src/action/subscribers/pid.rs

@@ -1,14 +1,15 @@
 use crate::action::{ThresholdMonitor, PublisherInstance, SubscriberStrategy};
 use crate::helpers::Deferred;
-use crate::io::{IOEvent, IOType, OutputType, };
+use crate::io::{DeviceType, IOEvent, IOType};
 
 /// Subscriber routine to actively maintain an arbitrary threshold using PID
 pub struct PIDMonitor {
     name: String,
     threshold: IOType,
     publisher: Option<Deferred<PublisherInstance>>,
 
-    output: Deferred<OutputType>,
+    // TODO: check that device is output
+    _output: Deferred<DeviceType>,
 }
 
 impl ThresholdMonitor for PIDMonitor {
@@ -21,7 +22,7 @@ impl SubscriberStrategy for PIDMonitor {
     fn name(&self) -> String {
         self.name.clone()
     }
-    fn evaluate(&mut self, data: &IOEvent) -> Option<IOEvent> {
+    fn evaluate(&mut self, _data: &IOEvent) -> Option<IOEvent> {
         todo!()
         // maintain PID
     }

src/action/subscribers/threshold.rs

@@ -1,5 +1,5 @@
 use std::sync::{Arc, Mutex};
-use crate::action::{BaseCommandFactory, Command, SimpleNotifier, PublisherInstance, SubscriberStrategy, SubscriberType};
+use crate::action::{BaseCommandFactory, PublisherInstance, SubscriberStrategy, SubscriberType};
 use crate::helpers::{Deferrable, Deferred};
 use crate::io::{IOEvent, IOType};
 
@@ -57,16 +57,12 @@ impl SubscriberStrategy for ThresholdNotifier {
 
     fn evaluate(&mut self, event: &IOEvent) -> Option<IOEvent> {
         let value = event.data.value;
-        let exceed = match &self.trigger {
+        let exceeded = match &self.trigger {
             &Comparison::GT => value >= self.threshold,
             &Comparison::LT => value <= self.threshold,
         };
-        if exceed {
-            // insert command here
-            let msg = format!("{} exceeded {}", value, self.threshold);
-            let command = SimpleNotifier::new(msg);
-            // Some(event.invert(1.0))  // re-enable this when dynamic IOTypes have been implemented
-            command.execute()
+        if exceeded {
+            (self.factory)(value, self.threshold).execute()
         } else {
             None
         }

src/action/types.rs

@@ -1,15 +1,7 @@
 //! Type aliases for functions and closures to assist `ActionBuilder`.
 //! These aliases allow for strongly structuring the dynamic initialization of subscriber/command instances.
-use crate::action::{Command, Comparison, PublisherInstance, SubscriberType, CommandType};
-use crate::helpers::Deferred;
+use crate::action::CommandType;
 use crate::io::IOType;
 
 // Command Factories
-pub type BaseCommandFactory = fn(IOType, IOType) -> CommandType;
-
-// **********************
-// Subscriber Factories *
-// **********************
-
-/// Type alias for a function or closure that returns a `ThresholdNotifier` instance
-pub type ThresholdNotifierFactory = fn(String, IOType, Comparison, BaseCommandFactory) -> Deferred<SubscriberType>;
+pub type BaseCommandFactory = fn(IOType, IOType) -> CommandType;

src/builders.rs

@@ -1,48 +1,5 @@
 mod action;
+mod device_log;
 
 pub use action::*;
-
-use std::sync::{Arc, Mutex};
-use crate::action::{BaseCommandFactory, Comparison, ThresholdNotifier, SimpleNotifier, Publisher,
-                    PublisherInstance};
-use crate::helpers::{Deferrable, Deferred};
-use crate::io::{Device, GenericInput, IdType, InputType, IOKind, IOType};
-use crate::settings::Settings;
-use crate::storage::OwnedLog;
-
-/// Init input and `OwnedLog`, then set owner on log. Return deferred log and deferred input.
-pub fn input_log_builder(
-    name: &str,
-    id: &IdType,
-    kind: &Option<IOKind>,
-    settings: Option<Arc<Settings>>,
-) -> (Deferred<OwnedLog>, Deferred<InputType>) {
-    let log = Arc::new(Mutex::new(OwnedLog::new(*id, settings)));
-    let input = GenericInput::new(name.to_string(), *id, *kind, log.clone());
-
-    let wrapped = input.deferred();
-    log.lock().unwrap().set_owner(wrapped.clone());
-
-    (log, wrapped)
-}
-
-pub fn pubsub_builder(input: Deferred<InputType>, name: String, threshold: IOType, trigger: Comparison,
-                      factory: BaseCommandFactory) {
-    let binding = PublisherInstance::default();
-    let publisher = binding.deferred();
-
-    // attempt to add publisher. Existing publisher is not overwritten.
-    let _ = input.try_lock().unwrap().add_publisher(publisher.clone());
-
-    let notifier = ThresholdNotifier::new(
-        name.clone(),
-        threshold,
-        trigger,
-        factory,
-    );
-    let deferred = notifier.deferred();
-    let mut binding = publisher.try_lock().unwrap();
-    binding.subscribe(deferred);
-
-    println!("Initialized and setup up subscriber: {}", name);
-}
+pub use device_log::*;

src/builders/action.rs

@@ -1,7 +1,9 @@
-use crate::action::{BaseCommandFactory, Comparison, Publisher, PublisherInstance, SubscriberType,
-                    ThresholdNotifier, ThresholdNotifierFactory};
+use std::ops::DerefMut;
+use crate::action::{BaseCommandFactory, Comparison, Publisher, PublisherInstance,
+                    ThresholdNotifier};
+use crate::errors::{ErrorKind, Error, Result};
 use crate::helpers::{Deferrable, Deferred};
-use crate::io::{IOType, InputType};
+use crate::io::{DeferredDevice, DeviceType, IOType, DeviceWrapper};
 
 /// Assist the user in dynamically initializing a single publisher for a single input.
 /// Since an abstract input only uses a single publisher, helper functions help build
@@ -10,14 +12,23 @@ use crate::io::{IOType, InputType};
 /// # Notes
 /// Return types should be checked here, if anywhere.
 pub struct ActionBuilder {
-    input: Deferred<InputType>,
+    // TODO: check that device is input
+    input: DeferredDevice,
     publisher: Deferred<PublisherInstance>,
     // TODO: add reference to `PollGroup`
 }
 impl ActionBuilder {
-    pub fn new(input: Deferred<InputType>) -> Self {
+    /// Create a new builder for a given device
+    /// This starts the process of adding pubs/subscribers
+    /// `Err` is returned if passed device is not input.
+    /// # Args
+    /// - device: Device to add pub/subs. Should be Input
+    pub fn new(device: DeferredDevice) -> Result<Self> {
+        if device.is_output() {
+            return Err(Error::new(ErrorKind::DeviceError, "Passed device is output. Expected input."))
+        }
         let publisher = Self::build_publisher();
-        Self { input, publisher }
+        Ok(Self { input: device, publisher })
     }
 
     /// Initialize and return a deferred `PublisherInstance`
@@ -27,16 +38,18 @@ impl ActionBuilder {
         // TODO: add publisher to `PollGroup`
     }
 
-    /// Silently add to publisher.
-    /// Existing publisher is not overwritten, however any returned error is muted.
+    /// Silently add publisher to input device.
+    /// Existing publisher is not overwritten as any returned error is ignored.
     /// Future updates will return a reference to the existing publisher. However, this shouldn't be
     /// necessary for instances built with `ActionBuilder`.
-    fn check_publisher(&mut self) {
+    fn check_publisher(&self) {
         let mut binding = self.input.lock().unwrap();
-        if binding.has_publisher() == false {
-            let publisher: Deferred<PublisherInstance> = self.publisher.clone();
+        if let DeviceType::Input(inner) = binding.deref_mut() {
+            if inner.has_publisher() == false {
+                let publisher: Deferred<PublisherInstance> = self.publisher.clone();
 
-            binding.add_publisher(publisher).unwrap()
+                inner.add_publisher(publisher).unwrap()
+            }
         }
     }
 
@@ -50,14 +63,14 @@ impl ActionBuilder {
     ) {
         self.check_publisher();
 
-        let subscriber = ThresholdNotifier::new(name.to_string(), threshold, trigger, factory);
-        let deferred = subscriber.deferred();
+        let _subscriber = ThresholdNotifier::new(name.to_string(), threshold, trigger, factory);
+        let subscriber = _subscriber.deferred();
 
         // add subscriber to publisher
-        self.publisher.lock().unwrap().subscribe(deferred.clone());
+        self.publisher.lock().unwrap().subscribe(subscriber.clone());
 
         // add reverse reference to publisher from subscriber
-        deferred
+        subscriber
             .lock()
             .unwrap()
             .add_publisher(self.publisher.clone());

src/builders/device_log.rs

@@ -0,0 +1,46 @@
+use crate::helpers::{Deferrable, Deferred};
+use crate::io::{
+    DeferredDevice, Device, DeviceType, GenericInput, GenericOutput, IODirection, IOKind, IdType,
+};
+use crate::settings::Settings;
+use crate::storage::OwnedLog;
+use std::sync::{Arc, Mutex, Weak};
+
+pub struct DeviceLogBuilder {
+    device: DeferredDevice,
+    log: Deferred<OwnedLog>,
+}
+
+impl DeviceLogBuilder {
+    pub fn new(
+        name: &str,
+        id: &IdType,
+        kind: &Option<IOKind>,
+        direction: &IODirection,
+        settings: Option<Arc<Settings>>,
+    ) -> Self {
+        let log: Deferred<OwnedLog> = Arc::new(Mutex::new(OwnedLog::new(*id, settings)));
+        let device = match direction {
+            IODirection::Output => {
+                let output = GenericOutput::new(name.to_string(), *id, *kind, log.clone());
+                DeviceType::Output(output)
+            }
+            IODirection::Input => {
+                let input = GenericInput::new(name.to_string(), *id, *kind, log.clone());
+                DeviceType::Input(input)
+            }
+        };
+
+        let wrapped = device.deferred();
+        let downgraded: Weak<Mutex<DeviceType>> = Arc::downgrade(&wrapped.clone());
+        log.lock().unwrap().set_owner(downgraded);
+        Self {
+            device: wrapped,
+            log,
+        }
+    }
+
+    pub fn get(&self) -> (DeferredDevice, Deferred<OwnedLog>) {
+        (self.device.clone(), self.log.clone())
+    }
+}

src/errors.rs

@@ -8,7 +8,10 @@ pub enum ErrorKind {
     ContainerError,
     ContainerEmpty,
     ContainerNotEmpty,
+
     SerializationError,
+
+    DeviceError,
 }
 
 #[derive(Debug)]
@@ -31,6 +34,7 @@ impl fmt::Display for Error {
             ErrorKind::ContainerEmpty => "Container is empty",
             ErrorKind::ContainerNotEmpty => "Container is not empty",
             ErrorKind::SerializationError => "Error during serialization",
+            ErrorKind::DeviceError => "Wrong type of device passed",
         };
 
         write!(f, "{}: {}", pretext, self.message)

src/helpers.rs

@@ -18,7 +18,8 @@ pub fn writable_or_create(path: String) -> File {
 
 /// Check a sequence of `Result`
 /// This used to check the returned outputs of recursive or parallel operations.
-pub fn check_results(results: &[Result<()>]) -> Result<()> {
+/// This does not crash the program but instead prints any errors via `dbg!`.
+pub fn check_results<T>(results: &[Result<T>]) -> Result<()> {
     for result in results {
         match result {
             Err(e) => dbg!(e),

src/io/device.rs

@@ -1,7 +1,9 @@
-/// Provide Low-level Device Functionality
+//! Provide Low-level Device Functionality
+use std::fmt::Formatter;
+use chrono::{DateTime, Utc};
 use crate::helpers::Deferred;
 use crate::io::metadata::DeviceMetadata;
-use crate::io::{IODirection, IOKind, IdType};
+use crate::io::{IODirection, IOKind, IdType, IOType, IOEvent};
 use crate::storage::OwnedLog;
 
 /// Defines a minimum interface for interacting with GPIO devices.
@@ -43,4 +45,21 @@ pub trait Device {
     fn kind(&self) -> IOKind {
         self.metadata().kind
     }
+
+    /// Generate an `IOEvent` instance from provided value or `::rx()`
+    fn generate_event(&self, dt: DateTime<Utc>, value: Option<IOType>) -> IOEvent;
 }
+
+impl std::fmt::Debug for dyn Device {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "Device: {} - {{ name: {}, id: {}, kind: {}}}",
+            self.direction(),
+            self.name(),
+            self.id(),
+            self.metadata().kind
+        )
+    }
+}
+

src/io/event.rs

@@ -3,10 +3,9 @@ use serde::{Deserialize, Serialize};
 
 use crate::io::types::{IOData, IOType, IdTraits};
 use crate::io::{Device, IODirection, IdType};
-use crate::storage::{Container, Containerized, LogType};
 
 /// Encapsulates `IOData` alongside of timestamp and device data
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize, Clone, Copy)]
 pub struct IOEvent {
     pub id: IdType,
     pub timestamp: DateTime<Utc>,
@@ -48,25 +47,6 @@ impl IOEvent {
             data,
         }
     }
-
-    /// Invert a copy of existing `IOEvent` and inject a new value.
-    /// This should be used for converting an `IOEvent` from input to output.
-    pub fn invert(&self, value: IOType) -> Self {
-        let mut inverted = self.clone();
-        inverted.data.value = value;
-        inverted.direction = match inverted.direction {
-            IODirection::Input => IODirection::Output,
-            IODirection::Output => IODirection::Input,
-        };
-        inverted
-    }
 }
 
 impl IdTraits for DateTime<Utc> {}
-
-/// Return a new instance of `Container` with for storing `IOEvent` which are accessed by `DateTime<Utc>` as keys
-impl Containerized<IOEvent, DateTime<Utc>> for IOEvent {
-    fn container() -> LogType {
-        Container::<IOEvent, DateTime<Utc>>::new()
-    }
-}