Implement Provider Actuation #5
Conversation
|
|
||
| # systemd related dependency, only relevant on linux systems | ||
| [target.'cfg(target_os = "linux")'.dependencies] | ||
| sd-notify = "0.4.1" | ||
|
|
||
| [features] | ||
| default = ["tls"] | ||
| default = ["tls", "dep:futures"] |
There was a problem hiding this comment.
required for val.rs -> provide_actuation
let future_vss_ids = vss_paths.iter().map(|vss_path| broker.get_id_by_path(&vss_path));
let resolved_opt_vss_ids = futures::future::join_all(future_vss_ids).await;
| pub struct ActuationSubscription { | ||
| vss_ids: Vec<i32>, | ||
| actuation_provider: Box<dyn ActuationProvider + Send + Sync + 'static>, | ||
| permissions: Permissions, | ||
| } |
There was a problem hiding this comment.
right now some cleanup operation is missing, e.g. an ActuationProvider disconnects or crashes and can no longer actuate. However, we prevent multiple ActuationProviders to register for the same signal, meaning once the ActuationProvider disconnects, it can no longer re-register to it's signal.
I can not use the actuation_provider to check if it is still available. I maybe could check (e.g. Err during send()) in val.rs -> Provider -> actuate but there I would not have the relation between the ActuationProvider and the ActuationSubscription. Sounds like a lot of overhead and not sure if there isn't a better way
| } | ||
|
|
||
| pub struct ActuationSubscription { | ||
| vss_ids: Vec<i32>, |
There was a problem hiding this comment.
vss_paths are normalized to vss_ids, so it's easier to keep a common list of available / registered actuators!
| if let Err(err) = provide_actuation(&broker, &provided_actuation, response_stream_sender).await { | ||
| debug!("Failed to provide actuation: {}", err) | ||
| } | ||
| break; | ||
| }, | ||
| Some(BatchActuateStreamResponse(_batch_actuate_stream_response)) => { | ||
| if let Err(err) = response_stream_sender.send(Err(tonic::Status::new(tonic::Code::Unimplemented, "Unimplemented"))).await { |
There was a problem hiding this comment.
Handling here is unclear. What is expected to happen when a client (who is not an Actuation Provider) sends the DataBroker a BatchActuateStreamResponse?
| Some(ProvideActuation(provided_actuation)) => { | ||
| if let Err(err) = provide_actuation(&broker, &provided_actuation, response_stream_sender).await { | ||
| debug!("Failed to provide actuation: {}", err) | ||
| } | ||
| break; | ||
| }, |
There was a problem hiding this comment.
Right now Errors are not properly propagated to the Provider - he won't know if registration fails (e.g. invalid signal) except of maybe due to the log. Since we are in a thread here, we can't return the err to the function body.
There was a problem hiding this comment.
Method provide_actuation btw was changed to return an Result<ProvideActuationResponse, tonic::Status> and send it to the sender. However, this still doesn't fix above mentioned problem.
Should we even establish a connection in case the signal does not exist? (just to say 'Yeah, you got a channel now, but you will never receive any responses except of this one, telling you that you provided the wrong signal') Maybe some kind of pre-verification / pre-validation is needed.
Implements the Provider Actuation.
Still missing stuff:
So please don't easily merge.