Status commit

src/main.rs

@@ -229,7 +229,7 @@ fn main() -> Result<(), Error> {
     // without warning.
     let modlog = log.clone();
     let regs = Registries::new();
-    match modules::init(modlog.new(o!("system" => "modules")), &config, &pool, regs) {
+    match exec.run_until(modules::init(modlog.new(o!("system" => "modules")), config.clone(), pool.clone(), regs)) {
         Ok(()) => {}
         Err(e) => {
             error!(modlog, "Module startup failed: {}", e);

src/modules.rs

@@ -17,9 +17,8 @@ use crate::error::Result;
 use crate::registries::Registries;
 
 // spawner is a type that allows 'tasks' to be spawned on it, running them to completion.
-pub fn init<S: Spawn>(log: Logger, config: &Settings, spawner: &S, registries: Registries) -> Result<()> {
-    let f = Box::new(shelly::run(log.clone(), config.clone(), registries.clone()));
-    spawner.spawn_obj(f.into())?;
+pub async fn init<S: Spawn + Clone + Send>(log: Logger, config: Settings, spawner: S, registries: Registries) -> Result<()> {
+    shelly::run(log.clone(), config.clone(), registries.clone()).await;
 
     Ok(())
 }

src/modules/shelly.rs

@@ -8,7 +8,8 @@ use crate::machine::Status;
 use std::pin::Pin;
 use futures::prelude::*;
 use futures::ready;
-use futures::task::{Poll, Context, Waker};
+use futures::task::{Poll, Context, Waker, Spawn};
+use futures::StreamExt;
 use futures_signals::signal::Signal;
 
 use paho_mqtt as mqtt;
@@ -19,7 +20,7 @@ use paho_mqtt as mqtt;
 pub async fn run(log: Logger, config: Settings, registries: Registries) {
     let shelly = Shelly::new(config).await;
 
-    let r = registries.actuators.register("shelly".to_string(), shelly).await;
+    let r = registries.actuators.register("shelly".to_string(), Box::new(shelly)).await;
 }
 
 /// An actuator for all Shellies connected listening on one MQTT broker
@@ -35,7 +36,7 @@ struct Shelly {
 
 impl Shelly {
     // Can't use Error, it's not Send. fabinfra/fabaccess/bffh#7
-    pub async fn new(config: Settings) -> ActBox {
+    pub async fn new(config: Settings) -> Self {
         let client = mqtt::AsyncClient::new(config.shelly.unwrap().mqtt_url).unwrap();
 
         client.connect(mqtt::ConnectOptions::new()).await.unwrap();
@@ -44,7 +45,7 @@ impl Shelly {
         let signal: Option<StatusSignal> = None;
         let waker = None;
 
-        Box::new(Shelly { signal, waker, name, client })
+        Shelly { signal, waker, name, client }
     }
 }
 

src/registries/actuators.rs

@@ -34,6 +34,20 @@ impl Actuators {
         // TODO: Log an error or something if that name was already taken
         wlock.insert(name, act);
     }
+
+    pub async fn run<S: Spawn>(&mut self) {
+        let mut wlock = self.inner.write().await;
+        for (_name, act) in wlock.into_iter() {
+
+        }
+    }
+
+    pub async fn subscribe(&mut self, name: String, signal: StatusSignal) {
+        let mut wlock = self.inner.write().await;
+        if let Some(act) = wlock.get_mut(&name) {
+            act.subscribe(signal);
+        }
+    }
 }
 
 pub type StatusSignal = Pin<Box<dyn Signal<Item = Status> + Send + Sync>>;