105 lines
3.3 KiB
Rust
105 lines
3.3 KiB
Rust
use std::sync::Weak;
|
|
|
|
use nxmesh_proto::{ConfigUpdate, MasterMessage, command::Command, master_message::Payload};
|
|
|
|
use crate::service::master_handler::{MasterHandlerError, MessageResult};
|
|
|
|
#[async_trait::async_trait]
|
|
pub trait MasterMessageHandler: Send + Sync + 'static {
|
|
async fn handle_master_message(
|
|
&self,
|
|
agent_id: &str,
|
|
message: MasterMessage,
|
|
) -> MessageResult<()>;
|
|
}
|
|
|
|
#[async_trait::async_trait]
|
|
pub trait OnConfigUpdateHandler: Send + Sync + 'static {
|
|
// Handle the config update message from master, write the config content to files, validate the new config and reload nginx
|
|
async fn on_config_update(
|
|
&self,
|
|
agent_id: &str,
|
|
timestamp: i64,
|
|
message_id: &str,
|
|
config_info: ConfigUpdate,
|
|
) -> MessageResult<()>;
|
|
}
|
|
|
|
#[async_trait::async_trait]
|
|
pub trait OnCommandHandler: Send + Sync + 'static {
|
|
// Handle the command message from master, execute the command and return the result
|
|
async fn on_command(
|
|
&self,
|
|
agent_id: &str,
|
|
timestamp: i64,
|
|
message_id: &str,
|
|
command: Command,
|
|
) -> MessageResult<()>;
|
|
}
|
|
|
|
pub struct HandlerImpl<OCUH, OCH>
|
|
where
|
|
OCUH: OnConfigUpdateHandler + ?Sized,
|
|
OCH: OnCommandHandler + ?Sized,
|
|
{
|
|
on_config_update_handler: Weak<OCUH>,
|
|
on_command_handler: Weak<OCH>,
|
|
}
|
|
|
|
impl<OCUH, OCH> HandlerImpl<OCUH, OCH>
|
|
where
|
|
OCUH: OnConfigUpdateHandler + ?Sized,
|
|
OCH: OnCommandHandler + ?Sized,
|
|
{
|
|
pub fn new(on_config_update_handler: Weak<OCUH>, on_command_handler: Weak<OCH>) -> Self {
|
|
Self {
|
|
on_config_update_handler,
|
|
on_command_handler,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[async_trait::async_trait]
|
|
impl<OCUH, OCH> MasterMessageHandler for HandlerImpl<OCUH, OCH>
|
|
where
|
|
OCUH: OnConfigUpdateHandler + ?Sized,
|
|
OCH: OnCommandHandler + ?Sized,
|
|
{
|
|
async fn handle_master_message(
|
|
&self,
|
|
agent_id: &str,
|
|
message: MasterMessage,
|
|
) -> MessageResult<()> {
|
|
match message.payload {
|
|
Some(Payload::ConfigUpdate(config_info)) => {
|
|
let on_config_update_handler =
|
|
self.on_config_update_handler.upgrade().ok_or_else(|| {
|
|
MasterHandlerError::MessageHandlingError(
|
|
"Failed to upgrade weak reference to config update handler".to_string(),
|
|
)
|
|
})?;
|
|
on_config_update_handler
|
|
.on_config_update(
|
|
agent_id,
|
|
message.timestamp,
|
|
&message.message_id,
|
|
config_info,
|
|
)
|
|
.await
|
|
}
|
|
Some(_) => {
|
|
// We should never receive other types of messages from the master, but we should handle it anyway
|
|
Err(MasterHandlerError::MessageHandlingError(
|
|
"Received unsupported master message type".to_string(),
|
|
))
|
|
}
|
|
None => {
|
|
// This should never happen as the master should always send a valid message, but we should handle it anyway
|
|
return Err(MasterHandlerError::MessageHandlingError(
|
|
"Received master message with empty payload".to_string(),
|
|
));
|
|
}
|
|
}
|
|
}
|
|
}
|