fabaccess-bffh/bffhd/initiators/mod.rs

181 lines
5.0 KiB
Rust
Raw Normal View History

2022-06-07 14:05:46 +02:00
use crate::initiators::dummy::Dummy;
2022-06-18 16:52:22 +02:00
use crate::initiators::process::Process;
2022-06-07 14:05:46 +02:00
use crate::resources::modules::fabaccess::Status;
use crate::session::SessionHandle;
use crate::{
AuthenticationHandle, Config, MachineState, Resource, ResourcesHandle, SessionManager,
};
use async_compat::CompatExt;
use executor::prelude::Executor;
use futures_util::ready;
use miette::IntoDiagnostic;
use rumqttc::ConnectReturnCode::Success;
use rumqttc::{AsyncClient, ConnectionError, Event, Incoming, MqttOptions};
use std::collections::HashMap;
use std::fmt::Display;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
2022-06-07 14:05:46 +02:00
use std::time::Duration;
use tracing::Span;
2022-06-07 14:05:46 +02:00
use url::Url;
2021-12-17 16:43:31 +01:00
2022-06-07 14:05:46 +02:00
mod dummy;
2022-06-18 16:52:22 +02:00
mod process;
2021-12-17 16:43:31 +01:00
2022-06-07 14:05:46 +02:00
pub trait Initiator: Future<Output = ()> {
fn new(params: &HashMap<String, String>, callbacks: InitiatorCallbacks) -> miette::Result<Self>
where
Self: Sized;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<()> {
<Self as Future>::poll(self, cx)
}
2021-12-17 16:43:31 +01:00
}
#[derive(Clone)]
2022-06-07 14:05:46 +02:00
pub struct InitiatorCallbacks {
span: Span,
2022-06-07 14:05:46 +02:00
resource: Resource,
sessions: SessionManager,
}
2022-06-07 14:05:46 +02:00
impl InitiatorCallbacks {
pub fn new(span: Span, resource: Resource, sessions: SessionManager) -> Self {
Self {
span,
resource,
sessions,
}
2021-12-17 16:43:31 +01:00
}
2022-06-07 14:05:46 +02:00
pub async fn try_update(&mut self, session: SessionHandle, status: Status) {
self.resource.try_update(session, status).await
}
pub fn set_status(&mut self, status: Status) {
self.resource.set_status(status)
}
2022-06-07 14:05:46 +02:00
pub fn open_session(&self, uid: &str) -> Option<SessionHandle> {
2022-11-01 10:47:51 +01:00
self.sessions.try_open(&self.span, uid)
2022-06-07 14:05:46 +02:00
}
}
2022-06-07 14:05:46 +02:00
pub struct InitiatorDriver {
span: Span,
2022-06-07 14:05:46 +02:00
name: String,
initiator: Box<dyn Initiator + Unpin + Send>,
}
2022-06-07 14:05:46 +02:00
impl InitiatorDriver {
pub fn new<I>(
span: Span,
2022-06-07 14:05:46 +02:00
name: String,
params: &HashMap<String, String>,
resource: Resource,
sessions: SessionManager,
) -> miette::Result<Self>
where
I: 'static + Initiator + Unpin + Send,
{
let callbacks = InitiatorCallbacks::new(span.clone(), resource, sessions);
2022-06-07 14:05:46 +02:00
let initiator = Box::new(I::new(params, callbacks)?);
Ok(Self {
span,
name,
initiator,
})
}
}
2022-06-07 14:05:46 +02:00
impl Future for InitiatorDriver {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
2022-06-07 14:05:46 +02:00
let _guard = tracing::info_span!("initiator poll", initiator=%self.name);
tracing::trace!(initiator=%self.name, "polling initiator");
2022-06-07 14:05:46 +02:00
ready!(Pin::new(&mut self.initiator).poll(cx));
2022-06-18 16:52:22 +02:00
tracing::warn!(initiator=%self.name, "initiator module ran to completion!");
Poll::Ready(())
}
2022-06-07 14:05:46 +02:00
}
pub fn load(
executor: Executor,
config: &Config,
resources: ResourcesHandle,
sessions: SessionManager,
authentication: AuthenticationHandle,
) -> miette::Result<()> {
let span = tracing::info_span!("loading initiators");
let _guard = span.enter();
let mut initiator_map: HashMap<String, Resource> = config
.init_connections
.iter()
.filter_map(|(k, v)| {
if let Some(resource) = resources.get_by_id(v) {
Some((k.clone(), resource.clone()))
} else {
tracing::error!(initiator=%k, machine=%v,
"Machine configured for initiator not found!");
None
}
})
.collect();
for (name, cfg) in config.initiators.iter() {
if let Some(resource) = initiator_map.remove(name) {
if let Some(driver) = load_single(name, &cfg.module, &cfg.params, resource, &sessions) {
tracing::debug!(module_name=%cfg.module, %name, "starting initiator task");
executor.spawn(driver);
} else {
tracing::error!(module_name=%cfg.module, %name, "Initiator module could not be configured");
}
} else {
tracing::warn!(actor=%name, ?config, "Initiator has no machine configured. Skipping!");
}
}
Ok(())
}
fn load_single(
name: &String,
module_name: &String,
params: &HashMap<String, String>,
resource: Resource,
sessions: &SessionManager,
) -> Option<InitiatorDriver> {
let span = tracing::info_span!(
"initiator",
name = %name,
module = %module_name,
);
2022-06-07 14:05:46 +02:00
tracing::info!(%name, %module_name, ?params, "Loading initiator");
let o = match module_name.as_ref() {
"Dummy" => Some(InitiatorDriver::new::<Dummy>(
span,
2022-06-07 14:05:46 +02:00
name.clone(),
params,
resource,
sessions.clone(),
)),
2022-06-18 16:52:22 +02:00
"Process" => Some(InitiatorDriver::new::<Process>(
span,
2022-06-18 16:52:22 +02:00
name.clone(),
params,
resource,
sessions.clone(),
)),
2022-06-07 14:05:46 +02:00
_ => None,
};
o.transpose().unwrap_or_else(|error| {
tracing::error!(%error, "failed to configure initiator");
None
})
}