diff --git a/src/access.rs b/src/access.rs index bdcee5a..cef7470 100644 --- a/src/access.rs +++ b/src/access.rs @@ -19,9 +19,9 @@ use crate::config::Settings; use crate::error::Result; // FIXME: fabinfra/fabaccess/bffh#3 -type UserIdentifier = u64; -type RoleIdentifier = u64; -type PermIdentifier = u64; +pub type UserIdentifier = u64; +pub type RoleIdentifier = u64; +pub type PermIdentifier = u64; pub struct PermissionsProvider { log: Logger, diff --git a/src/machine.rs b/src/machine.rs index d7fb90e..3cd8a62 100644 --- a/src/machine.rs +++ b/src/machine.rs @@ -1,6 +1,9 @@ +use std::str::FromStr; use std::collections::HashMap; +use std::fs; use std::fs::File; use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; use slog::Logger; @@ -11,19 +14,25 @@ use smol::lock::RwLock; use crate::error::Result; use crate::config::Settings; +use crate::access; use capnp::Error; use uuid::Uuid; -use lmdb::{Transaction, RwTransaction}; +use lmdb::{Transaction, RwTransaction, Cursor}; use smol::channel::{Receiver, Sender}; use futures_signals::signal::*; +use crate::registries::StatusSignal; + +pub type ID = Uuid; + /// Status of a Machine #[derive(Clone, Copy, PartialEq, Eq, Debug, Serialize, Deserialize)] +#[repr(u8)] pub enum Status { /// Not currently used by anybody Free, @@ -33,17 +42,6 @@ pub enum Status { Blocked, } -pub struct MachinesProvider { - log: Logger, - mdb: MachineDB, -} - -impl MachinesProvider { - pub fn new(log: Logger, mdb: MachineDB) -> Self { - Self { log, mdb } - } -} - #[derive(Clone)] pub struct Machines { inner: Arc>, @@ -98,11 +96,16 @@ impl MachineManager { /// machine, checking that the user who wants the machine (de)activated has the required /// permissions. pub struct Machine { + /// Computer-readable identifier for this machine + // Implicit in database since it's the key. + #[serde(skip)] + id: ID, + /// The human-readable name of the machine. Does not need to be unique name: String, /// The required permission to use this machine. - perm: String, + perm: access::PermIdentifier, /// The state of the machine as bffh thinks the machine *should* be in. /// @@ -112,8 +115,9 @@ pub struct Machine { } impl Machine { - pub fn new(name: String, perm: String) -> Machine { + pub fn new(id: Uuid, name: String, perm: access::PermIdentifier) -> Machine { Machine { + id: id, name: name, perm: perm, state: Mutable::new(Status::Free), @@ -128,18 +132,41 @@ impl Machine { /// dedupe ensures that if state is changed but only changes to the value it had beforehand /// (could for example happen if the machine changes current user but stays activated) no /// update is sent. - pub fn signal(&self) -> impl Signal { - self.state.signal().dedupe() + pub fn signal(&self) -> StatusSignal { + Box::pin(self.state.signal().dedupe()) + } + + /// Requests to use a machine. Returns `true` if successful. + /// + /// This will update the internal state of the machine, notifying connected actors, if any. + pub fn request_use + ( &mut self + , txn: &T + , pp: &access::PermissionsProvider + , who: access::UserIdentifier + ) -> Result + { + if pp.check(txn, who, self.perm)? { + self.state.set(Status::Occupied); + return Ok(true); + } else { + return Ok(false); + } + } + + pub fn set_state(&mut self, state: Status) { + self.state.set(state) } } -pub struct MachineDB { +pub struct MachinesProvider { + log: Logger, db: lmdb::Database, } -impl MachineDB { - pub fn new(db: lmdb::Database) -> Self { - Self { db } +impl MachinesProvider { + pub fn new(log: Logger, db: lmdb::Database) -> Self { + Self { log, db } } pub fn get_machine(&self, txn: &T, uuid: Uuid) @@ -147,7 +174,10 @@ impl MachineDB { { match txn.get(self.db, &uuid.as_bytes()) { Ok(bytes) => { - Ok(Some(flexbuffers::from_slice(bytes)?)) + let mut machine: Machine = flexbuffers::from_slice(bytes)?; + machine.id = uuid; + + Ok(Some(machine)) }, Err(lmdb::Error::NotFound) => { Ok(None) }, Err(e) => { Err(e.into()) }, @@ -162,8 +192,75 @@ impl MachineDB { Ok(()) } + + pub fn load_db(&mut self, txn: &mut RwTransaction, mut path: PathBuf) -> Result<()> { + path.push("machines"); + for entry in std::fs::read_dir(path)? { + let entry = entry?; + let path = entry.path(); + if path.is_file() { + // will only ever be none if the path has no file name and then how is it a file?! + let machID_str = path + .file_stem().expect("Found a file with no filename?") + .to_str().expect("Found an OsStr that isn't valid Unicode. Fix your OS!"); + let machID = match uuid::Uuid::from_str(machID_str) { + Ok(i) => i, + Err(e) => { + warn!(self.log, "File {} had a invalid name. Expected an u64 in [0-9a-z] hex with optional file ending: {}. Skipping!", path.display(), e); + continue; + } + }; + let s = match fs::read_to_string(path.as_path()) { + Ok(s) => s, + Err(e) => { + warn!(self.log, "Failed to open file {}: {}, skipping!" + , path.display() + , e); + continue; + } + }; + let mach: Machine = match toml::from_str(&s) { + Ok(r) => r, + Err(e) => { + warn!(self.log, "Failed to parse mach at path {}: {}, skipping!" + , path.display() + , e); + continue; + } + }; + self.put_machine(txn, machID, mach)?; + debug!(self.log, "Loaded machine {}", machID); + } else { + warn!(self.log, "Path {} is not a file, skipping!", path.display()); + } + } + + Ok(()) + } + + pub fn dump_db(&self, txn: &T, mut path: PathBuf) -> Result<()> { + path.push("machines"); + let mut mach_cursor = txn.open_ro_cursor(self.db)?; + for buf in mach_cursor.iter_start() { + let (kbuf, vbuf) = buf?; + let machID = uuid::Uuid::from_slice(kbuf).unwrap(); + let mach: Machine = flexbuffers::from_slice(vbuf)?; + let filename = format!("{}.yml", machID.to_hyphenated().to_string()); + path.set_file_name(filename); + let mut fp = std::fs::File::create(&path)?; + let out = toml::to_vec(&mach)?; + fp.write_all(&out)?; + } + + Ok(()) + } } -pub async fn init(log: Logger, config: &Settings) -> Result { - unimplemented!() +pub fn init(log: Logger, config: &Settings, env: &lmdb::Environment) -> Result { + let mut flags = lmdb::DatabaseFlags::empty(); + flags.set(lmdb::DatabaseFlags::INTEGER_KEY, true); + let machdb = env.create_db(Some("machines"), flags)?; + debug!(&log, "Opened machine db successfully."); + + Ok(MachinesProvider::new(log, machdb)) } diff --git a/src/main.rs b/src/main.rs index 25bc642..1572ae2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -132,7 +132,7 @@ fn main() -> Result<(), Error> { // Start loading the machine database, authentication system and permission system // All of those get a custom logger so the source of a log message can be better traced and // filtered - let machinedb_f = machine::init(log.new(o!("system" => "machines")), &config); + let mdb = machine::init(log.new(o!("system" => "machines")), &config, &env); let pdb = access::init(log.new(o!("system" => "permissions")), &config, &env); let authentication_f = auth::init(log.new(o!("system" => "authentication")), config.clone()); @@ -148,7 +148,8 @@ fn main() -> Result<(), Error> { let mut txn = env.begin_rw_txn()?; let path = path.to_path_buf(); - pdb?.load_db(&mut txn, path)?; + pdb?.load_db(&mut txn, path.clone())?; + mdb?.load_db(&mut txn, path)?; txn.commit(); } else { error!(log, "You must provide a directory path to load from"); @@ -165,9 +166,9 @@ fn main() -> Result<(), Error> { let txn = env.begin_ro_txn()?; let path = path.to_path_buf(); - pdb?.dump_db(&txn, path)?; + pdb?.dump_db(&txn, path.clone())?; + mdb?.dump_db(&txn, path)?; } else { - error!(log, "You must provide a directory path to dump into"); } @@ -197,16 +198,16 @@ fn main() -> Result<(), Error> { } }).collect(); - let (mach, auth) = exec.run_until(async { - // Rull all futures to completion in parallel. - // This will block until all three are done starting up. - join!(machinedb_f, authentication_f) - }); + //let (mach, auth) = exec.run_until(async { + // // Rull all futures to completion in parallel. + // // This will block until all three are done starting up. + // join!(machinedb_f, authentication_f) + //}); // Error out if any of the subsystems failed to start. - let mach = mach?; + let mdb = mdb?; let pdb = pdb?; - let auth = auth?; + //let auth = auth?; // Since the below closures will happen at a much later time we need to make sure all pointers // are still valid. Thus, Arc. @@ -228,8 +229,8 @@ fn main() -> Result<(), Error> { // FIXME: implement notification so the modules can shut down cleanly instead of being killed // without warning. let modlog = log.clone(); - let regs = Registries::new(); - match modules::init(modlog.new(o!("system" => "modules")), &config, &local_spawn, regs) { + let mut regs = Registries::new(); + match exec.run_until(modules::init(modlog.new(o!("system" => "modules")), config.clone(), pool.clone(), regs.clone())) { Ok(()) => {} Err(e) => { error!(modlog, "Module startup failed: {}", e); diff --git a/src/modules.rs b/src/modules.rs index 8faaf35..9f8a21e 100644 --- a/src/modules.rs +++ b/src/modules.rs @@ -10,16 +10,15 @@ use slog::Logger; mod shelly; use futures::prelude::*; -use futures::task::LocalSpawn; +use futures::task::Spawn; use crate::config::Settings; 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(log: Logger, config: &Settings, spawner: &S, registries: Registries) -> Result<()> { - let f = Box::new(shelly::run(log.clone(), config.clone(), registries.clone())); - spawner.spawn_local_obj(f.into())?; +pub async fn init(log: Logger, config: Settings, spawner: S, registries: Registries) -> Result<()> { + shelly::run(log.clone(), config.clone(), registries.clone(), spawner.clone()).await; Ok(()) } diff --git a/src/modules/shelly.rs b/src/modules/shelly.rs index 732782c..91ebee6 100644 --- a/src/modules/shelly.rs +++ b/src/modules/shelly.rs @@ -1,30 +1,32 @@ use slog::Logger; use crate::config::Settings; -use crate::registries::{Registries, Actuator, ActBox}; +use crate::registries::{Registries, Actuator, ActBox, StatusSignal}; use crate::error::Result; +use crate::machine::Status; use std::pin::Pin; use futures::prelude::*; +use futures::channel::mpsc; use futures::ready; -use futures::task::{Poll, Context}; +use futures::task::{Poll, Context, Waker, Spawn, FutureObj}; +use futures::StreamExt; +use futures_signals::signal::Signal; use paho_mqtt as mqtt; // TODO: Late config parsing. Right now the config is validated at the very startup in its // entirety. This works reasonably enough for this static modules here but if we do dynamic loading // via dlopen(), lua API, python API etc it will not. -pub async fn run(log: Logger, config: Settings, registries: Registries) { - let shelly_r = Shelly::new(config).await; - if let Err(e) = shelly_r { - error!(log, "Shelly module errored: {}", e); - return; - } +pub async fn run(log: Logger, config: Settings, registries: Registries, spawner: S) { + let (tx, rx) = mpsc::channel(1); + let mut shelly = Shelly::new(log, config, rx).await; + + let r = registries.actuators.register("shelly".to_string(), tx).await; + + let f = shelly.for_each(|f| f); + spawner.spawn_obj(FutureObj::from(Box::pin(f))); - let r = registries.actuators.register( - "shelly".to_string(), - shelly_r.unwrap() - ).await; } /// An actuator for all Shellies connected listening on one MQTT broker @@ -32,31 +34,87 @@ pub async fn run(log: Logger, config: Settings, registries: Registries) { /// This actuator can power toggle an arbitrariy named shelly on the broker it is connected to. If /// you need to toggle shellies on multiple brokers you need multiple instanced of this actuator. struct Shelly { + log: Logger, + sigchan: mpsc::Receiver, + signal: Option, + waker: Option, + name: String, client: mqtt::AsyncClient, } impl Shelly { - pub async fn new(config: Settings) -> Result { - let client = mqtt::AsyncClient::new(config.shelly.unwrap().mqtt_url)?; + // Can't use Error, it's not Send. fabinfra/fabaccess/bffh#7 + pub async fn new(log: Logger, config: Settings, sigchan: mpsc::Receiver) -> Self { + let client = mqtt::AsyncClient::new(config.shelly.unwrap().mqtt_url).unwrap(); - client.connect(mqtt::ConnectOptions::new()).await?; + let o = client.connect(mqtt::ConnectOptions::new()).await.unwrap(); + println!("{:?}", o); - Ok(Box::new(Shelly { client }) as ActBox) + let name = "test".to_string(); + let signal: Option = None; + let waker = None; + + Shelly { log, sigchan, signal, waker, name, client } } } -#[async_trait] impl Actuator for Shelly { - async fn power_on(&mut self, name: String) { - let topic = format!("shellies/{}/relay/0/command", name); - let msg = mqtt::Message::new(topic, "on", 0); - self.client.publish(msg).map(|_| ()).await - } - - async fn power_off(&mut self, name: String) { - let topic = format!("shellies/{}/relay/0/command", name); - let msg = mqtt::Message::new(topic, "off", 0); - self.client.publish(msg).map(|_| ()).await + fn subscribe(&mut self, signal: StatusSignal) { + self.signal.replace(signal); + if let Some(waker) = self.waker.take() { + waker.wake(); + } + } +} + +impl Stream for Shelly { + type Item = future::BoxFuture<'static, ()>; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + let unpin = Pin::into_inner(self); + + info!(unpin.log, "tick {}", unpin.signal.is_some()); + + if let Poll::Ready(v) = Stream::poll_next(Pin::new(&mut unpin.sigchan), cx) { + if let Some(s) = v { + // We have received a new signal to use + unpin.signal.replace(s); + // We use `if let` instead of .and_then because we want the waker to be dropped + // afterwards. It's only there to ensure the future is called when a signal is + // installed the first time + // TODO probably don't need that here because we're polling it either way directly + // afterwards, eh? + if let Some(waker) = unpin.waker.take() { + waker.wake(); + } + } else { + info!(unpin.log, "bye"); + // This means that the sending end was dropped, so we shut down + unpin.signal.take(); + unpin.waker.take(); + return Poll::Ready(None); + } + } + + if let Some(ref mut s) = unpin.signal { + if let Some(status) = ready!(Signal::poll_change(Pin::new(s), cx)) { + info!(unpin.log, "Machine Status changed: {:?}", status); + let topic = format!("shellies/{}/relay/0/command", unpin.name); + let pl = match status { + Status::Free | Status::Blocked => "off", + Status::Occupied => "on", + }; + let msg = mqtt::Message::new(topic, pl, 0); + let f = unpin.client.publish(msg).map(|_| ()); + + return Poll::Ready(Some(Box::pin(f))); + } + } else { + info!(unpin.log, "I ain't got no signal son"); + unpin.waker.replace(cx.waker().clone()); + } + + Poll::Pending } } diff --git a/src/network.rs b/src/network.rs index 5e4559d..a93b8b0 100644 --- a/src/network.rs +++ b/src/network.rs @@ -1,5 +1,8 @@ use futures_signals::signal::Signal; +use crate::machine; +use crate::access; + struct Network { } @@ -24,10 +27,12 @@ impl Network { } } +/// The internal bffh event type +/// +/// Everything that BFFH considers an event is contained in an instance of this. +#[derive(PartialEq, Eq, Clone, PartialOrd, Ord, Debug)] enum Event { - -} - -trait Filter { - fn filter(&self, f: Fn(&S) -> bool); + /// An user wants to use a machine + // TODO: Define /what/ an user wants to do with said machine? + MachineRequest(machine::ID, access::UserIdentifier), } diff --git a/src/registries.rs b/src/registries.rs index e5d218c..0975bab 100644 --- a/src/registries.rs +++ b/src/registries.rs @@ -1,6 +1,8 @@ mod actuators; +mod sensors; -pub use actuators::{Actuator, ActBox}; +pub use actuators::{Actuator, ActBox, StatusSignal}; +pub use sensors::{Sensor, SensBox}; #[derive(Clone)] /// BFFH registries @@ -9,12 +11,14 @@ pub use actuators::{Actuator, ActBox}; /// reference, not clone the registries pub struct Registries { pub actuators: actuators::Actuators, + pub sensors: sensors::Sensors, } impl Registries { pub fn new() -> Self { Registries { - actuators: actuators::Actuators::new() + actuators: actuators::Actuators::new(), + sensors: sensors::Sensors::new(), } } } diff --git a/src/registries/actuators.rs b/src/registries/actuators.rs index 890fc1e..2ffa81f 100644 --- a/src/registries/actuators.rs +++ b/src/registries/actuators.rs @@ -1,7 +1,16 @@ +use slog::Logger; + use std::sync::Arc; use smol::lock::RwLock; +use std::pin::Pin; +use futures::ready; use futures::prelude::*; +use futures::channel::mpsc; +use futures::task::{Context, Poll, Spawn}; +use futures_signals::signal::Signal; + +use crate::machine::Status; use std::collections::HashMap; @@ -10,11 +19,9 @@ pub struct Actuators { inner: Arc>, } -unsafe impl Send for Actuators { } +pub type ActBox = Box; -pub type ActBox = Box; - -type Inner = HashMap; +type Inner = HashMap>; impl Actuators { pub fn new() -> Self { @@ -23,31 +30,53 @@ impl Actuators { } } - pub async fn register(&self, name: String, act: ActBox) { + pub async fn register(&self, name: String, tx: mpsc::Sender) { let mut wlock = self.inner.write().await; // TODO: Log an error or something if that name was already taken - wlock.insert(name, act); + wlock.insert(name, tx); + } + + pub async fn subscribe(&mut self, name: String, signal: StatusSignal) { + let mut wlock = self.inner.write().await; + if let Some(tx) = wlock.get_mut(&name) { + tx.send(signal).await; + } } } +pub type StatusSignal = Pin + Send + Sync>>; -#[async_trait] -pub trait Actuator { - // TODO: Is it smarter to pass a (reference to?) a machine instead of 'name'? Do we need to - // pass basically arbitrary parameters to the Actuator? - async fn power_on(&mut self, name: String); - async fn power_off(&mut self, name: String); +pub trait Actuator: Stream> { + fn subscribe(&mut self, signal: StatusSignal); } // This is merely a proof that Actuator *can* be implemented on a finite, known type. Yay for type // systems with halting problems. -struct Dummy; -#[async_trait] +struct Dummy { + log: Logger, + sigchan: mpsc::Receiver, + signal: Option, +} + impl Actuator for Dummy { - async fn power_on(&mut self, _name: String) { - return - } - async fn power_off(&mut self, _name: String) { - return + fn subscribe(&mut self, signal: StatusSignal) { + self.signal.replace(signal); + } +} + +impl Stream for Dummy { + type Item = future::BoxFuture<'static, ()>; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + let unpin = Pin::into_inner(self); + if let Some(ref mut s) = unpin.signal { + let status = ready!(Signal::poll_change(Pin::new(s), cx)); + + info!(unpin.log, "Dummy actuator would set status to {:?}, but is a Dummy", status); + + Poll::Ready(Some(Box::pin(futures::future::ready(())))) + } else { + Poll::Pending + } } } diff --git a/src/registries/sensors.rs b/src/registries/sensors.rs new file mode 100644 index 0000000..2b7819e --- /dev/null +++ b/src/registries/sensors.rs @@ -0,0 +1,73 @@ +use std::pin::Pin; +use futures::task::{Context, Poll}; +use futures::{Future, Stream}; +use futures::future::BoxFuture; + +use std::sync::Arc; +use smol::lock::RwLock; +use std::collections::HashMap; + +#[derive(Clone)] +pub struct Sensors { + inner: Arc>, +} + +impl Sensors { + pub fn new() -> Self { + Sensors { + inner: Arc::new(RwLock::new(Inner::new())), + } + } +} + +pub type SensBox = Box; +type Inner = HashMap; + + +// Implementing Sensors. +// +// Given the coroutine/task split stays as it is - Sensor input to machine update being one, +// machine update signal to actor doing thing being another, a Sensor implementation would send a +// Stream of futures - each future being an atomic Machine update. +#[async_trait] +/// BFFH Sensor +/// +/// A sensor is anything that can forward an intent of an user to do something to bffh. +/// This may be a card reader connected to a machine, a website allowing users to select a machine +/// they want to use or something like QRHello +pub trait Sensor: Stream> { + /// Setup the Sensor. + /// + /// After this async function completes the Stream implementation should be able to generate + /// futures when polled. + /// Implementations can rely on this function being polled to completeion before the stream + /// is polled. + // TODO Is this sensible vs just having module-specific setup fns? + async fn setup(&mut self); + + /// Shutdown the sensor gracefully + /// + /// Implementations can rely on that the stream will not be polled after this function has been + /// called. + async fn shutdown(&mut self); +} + +struct Dummy; +#[async_trait] +impl Sensor for Dummy { + async fn setup(&mut self) { + return; + } + + async fn shutdown(&mut self) { + return; + } +} + +impl Stream for Dummy { + type Item = BoxFuture<'static, ()>; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + Poll::Ready(Some(Box::pin(futures::future::ready(())))) + } +}