Better initialization

src/actor.rs

@@ -11,11 +11,13 @@ use futures::channel::mpsc;
 use futures_signals::signal::{Signal, MutableSignalCloned, MutableSignal, Mutable};
 
 use crate::db::machine::MachineState;
-use crate::config::Settings;
+use crate::config::Config;
 use crate::error::Result;
-
 use crate::network::ActorMap;
 
+use paho_mqtt::AsyncClient;
+use slog::Logger;
+
 pub trait Actuator {
     fn apply(&mut self, state: MachineState) -> BoxFuture<'static, ()>;
 }
@@ -104,12 +106,34 @@ impl Actuator for Dummy {
     }
 }
 
-pub fn load() -> Result<(ActorMap, Vec<Actor>)> {
+pub fn load(log: &Logger, client: &AsyncClient, config: &Config) -> Result<(ActorMap, Vec<Actor>)> {
-    let d = Box::new(Dummy);
-    let (tx, a) = Actor::wrap(d);
-
     let mut map = HashMap::new();
-    map.insert("Dummy".to_string(), tx);
 
-    Ok(( map, vec![a] ))
+    let actuators = config.actors.iter()
+        .map(|(k,v)| (k, load_single(log, client, k, &v.name, &v.params)))
+        .filter_map(|(k, n)| match n {
+            None => None,
+            Some(a) => Some((k, a))
+        });
+
+    let mut v = Vec::new();
+    for (name, actuator) in actuators {
+        let (tx, a) = Actor::wrap(actuator);
+        map.insert(name.clone(), tx);
+        v.push(a);
+    }
+
+
+    Ok(( map, v ))
+}
+
+fn load_single(log: &Logger, client: &AsyncClient, name: &String, module_name: &String, params: &HashMap<String, String>) -> Option<Box<dyn Actuator + Sync + Send>> {
+    use crate::modules::*;
+
+    match module_name.as_ref() {
+        "Shelly" => {
+            Some(Box::new(Shelly::new(log, name.clone(), client.clone())))
+        }
+        _ => None,
+    }
 }

src/config.rs

@@ -12,7 +12,9 @@ use crate::machine::MachineDescription;
 use crate::db::machine::MachineIdentifier;
 
 pub fn read(path: &Path) -> Result<Config> {
-    serde_dhall::from_file(path).parse().map_err(Into::into)
+    serde_dhall::from_file(path)
+        .parse()
+        .map_err(Into::into)
 }
 
 #[deprecated]
@@ -29,8 +31,13 @@ pub struct Config {
     /// Machine descriptions to load
     pub machines: HashMap<MachineIdentifier, MachineDescription>,
 
-    /// Modules to load and their configuration options
+    /// Actors to load and their configuration options
-    pub modules: HashMap<String, HashMap<String, String>>,
+    pub actors: HashMap<String, ModuleConfig>,
+
+    /// Initiators to load and their configuration options
+    pub initiators: HashMap<String, ModuleConfig>,
+
+    pub mqtt_url: String,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -39,13 +46,38 @@ pub struct Listen {
     pub port: Option<u16>,
 }
 
-impl Default for Settings {
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ModuleConfig {
+    pub name: String,
+    #[serde(skip_serializing_if = "HashMap::is_empty")]
+    pub params: HashMap<String, String>
+}
+
+impl Default for Config {
     fn default() -> Self {
-        let modules: HashMap::<String, HashMap<String, String>> = HashMap::new();
+        let mut actors: HashMap::<String, ModuleConfig> = HashMap::new();
+        let mut initiators: HashMap::<String, ModuleConfig> = HashMap::new();
+
+        actors.insert("Actor".to_string(), ModuleConfig {
+            name: "Shelly".to_string(),
+            params: HashMap::new(),
+        });
+        initiators.insert("Initiator".to_string(), ModuleConfig {
+            name: "TCP-Listen".to_string(),
+            params: HashMap::new(),
+        });
+
         Config {
-            listens: Box::new([]),
+            listens: Box::new([
+                Listen {
+                    address: "localhost".to_string(),
+                    port: Some(DEFAULT_PORT),
+                }
+            ]),
             machines: HashMap::new(),
-            modules: modules,
+            actors: actors,
+            initiators: initiators,
+            mqtt_url: "tcp://localhost:1883".to_string(),
         }
     }
 }

src/main.rs

@@ -46,6 +46,8 @@ use error::Error;
 
 use slog::Logger;
 
+use paho_mqtt::AsyncClient;
+
 fn main() {
     use clap::{crate_version, crate_description, crate_name};
 
@@ -81,14 +83,14 @@ fn main() {
     // Check for the --print-default option first because we don't need to do anything else in that
     // case.
     if matches.is_present("print default") {
-        let config = config::Settings::default();
+        let config = config::Config::default();
-        let encoded = toml::to_vec(&config).unwrap();
+        let encoded = serde_dhall::serialize(&config).to_string().unwrap();
 
         // Direct writing to fd 1 is faster but also prevents any print-formatting that could
         // invalidate the generated TOML
         let stdout = io::stdout();
         let mut handle = stdout.lock();
-        handle.write_all(&encoded).unwrap();
+        handle.write_all(&encoded.as_bytes()).unwrap();
 
         // Early return to exit.
         return;
@@ -123,7 +125,7 @@ fn maybe(matches: clap::ArgMatches, log: Arc<Logger>) -> Result<(), Error> {
     // If no `config` option is given use a preset default.
     let configpath = matches.value_of("config").unwrap_or("/etc/bffh/config.toml");
     let config = config::read(&PathBuf::from_str(configpath).unwrap())?;
-
+    debug!(log, "Loaded Config: {:?}", config);
 
     if matches.is_present("dump") {
         error!(log, "Dumping is currently not implemented");
@@ -134,8 +136,13 @@ fn maybe(matches: clap::ArgMatches, log: Arc<Logger>) -> Result<(), Error> {
     } else {
         let ex = Executor::new();
 
+        let mqtt = AsyncClient::new(config.mqtt_url.clone())?;
+        let tok = mqtt.connect(paho_mqtt::ConnectOptions::new());
+
+        smol::block_on(tok);
+
         let machines = machine::load(&config)?;
-        let (mut actor_map, actors) = actor::load()?;
+        let (mut actor_map, actors) = actor::load(&log, &mqtt, &config)?;
         let (mut init_map, initiators) = initiator::load()?;
 
         let network = network::Network::new(machines, actor_map, init_map);
@@ -143,9 +150,6 @@ fn maybe(matches: clap::ArgMatches, log: Arc<Logger>) -> Result<(), Error> {
 
         // TODO HERE: restore connections between initiators, machines, actors
 
-        // TODO HERE: Spawn all actors & inits
-
-        // Like so
         let actor_tasks = actors.into_iter().map(|actor| ex.spawn(actor));
         let init_tasks = initiators.into_iter().map(|init| ex.spawn(init));
 

src/modules.rs

@@ -8,6 +8,7 @@
 use slog::Logger;
 
 mod shelly;
+pub use shelly::Shelly;
 
 use futures::prelude::*;
 use futures::task::Spawn;

src/modules/shelly.rs

@@ -23,7 +23,7 @@ use paho_mqtt as mqtt;
 /// This actuator will toggle the shellie with the given `name`.
 /// If you need to toggle shellies on multiple brokers you need multiple instanced of this
 /// actuator with different clients.
-struct Shelly {
+pub struct Shelly {
     log: Logger,
     name: String,
     client: mqtt::AsyncClient,

src/network.rs

@@ -39,6 +39,7 @@ impl fmt::Display for Error {
 /// Main signal network
 ///
 /// Network as per FRP, not the one with packages and frames
+// TODO De/Serialize established connection on startup/shutdown.
 pub struct Network {
     inits: InitMap,