simplify message handler types
This commit is contained in:
@@ -4,7 +4,7 @@
|
|||||||
//! writing the message terminated with CRLF, and reading the response until CRLF.
|
//! writing the message terminated with CRLF, and reading the response until CRLF.
|
||||||
|
|
||||||
use conf::Conf;
|
use conf::Conf;
|
||||||
use signal_gateway::MessageHandlerResult;
|
use signal_gateway::{AdminMessageResponse, MessageHandlerResult};
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
use tokio::{
|
use tokio::{
|
||||||
io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
|
io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
|
||||||
@@ -74,5 +74,5 @@ async fn handle_message(config: &AdminNetcatConfig, message: String) -> MessageH
|
|||||||
.trim_end_matches(['\r', '\n'])
|
.trim_end_matches(['\r', '\n'])
|
||||||
.to_owned();
|
.to_owned();
|
||||||
|
|
||||||
Ok((text, vec![]))
|
Ok(AdminMessageResponse::new(text))
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ use crate::{
|
|||||||
alertmanager::AlertPost,
|
alertmanager::AlertPost,
|
||||||
jsonrpc::{Envelope, RpcClient, RpcClientError, SignalMessage, connect_tcp},
|
jsonrpc::{Envelope, RpcClient, RpcClientError, SignalMessage, connect_tcp},
|
||||||
log_message::{LogMessage, Origin},
|
log_message::{LogMessage, Origin},
|
||||||
|
message_handler::{AdminMessageResponse, MessageHandler, MessageHandlerResult},
|
||||||
prometheus::{Prometheus, PrometheusConfig},
|
prometheus::{Prometheus, PrometheusConfig},
|
||||||
};
|
};
|
||||||
use chrono::Utc;
|
use chrono::Utc;
|
||||||
@@ -11,10 +12,7 @@ use http::{Method, Request, Response, StatusCode};
|
|||||||
use http_body::Body;
|
use http_body::Body;
|
||||||
use http_body_util::BodyExt;
|
use http_body_util::BodyExt;
|
||||||
use prometheus_http_client::{AlertStatus, ExtractLabels};
|
use prometheus_http_client::{AlertStatus, ExtractLabels};
|
||||||
use std::{
|
use std::{collections::HashMap, fmt::Write, net::SocketAddr, path::PathBuf};
|
||||||
collections::HashMap, error::Error, fmt::Write, future::Future, net::SocketAddr,
|
|
||||||
path::PathBuf, pin::Pin,
|
|
||||||
};
|
|
||||||
use tokio::{
|
use tokio::{
|
||||||
join,
|
join,
|
||||||
sync::{
|
sync::{
|
||||||
@@ -27,15 +25,6 @@ use tokio_util::bytes::Buf;
|
|||||||
use tokio_util::sync::CancellationToken;
|
use tokio_util::sync::CancellationToken;
|
||||||
use tracing::{debug, error, info, warn};
|
use tracing::{debug, error, info, warn};
|
||||||
|
|
||||||
/// Response from a message handler: either success with text and optional attachments,
|
|
||||||
/// or an error with status code and message.
|
|
||||||
pub type MessageHandlerResult = Result<(String, Vec<PathBuf>), (u16, Box<dyn Error + Send + Sync>)>;
|
|
||||||
|
|
||||||
/// Handler function for admin messages that don't start with `/`.
|
|
||||||
/// Takes the message text and returns a response.
|
|
||||||
pub type MessageHandler =
|
|
||||||
Box<dyn Fn(String) -> Pin<Box<dyn Future<Output = MessageHandlerResult> + Send>> + Send + Sync>;
|
|
||||||
|
|
||||||
mod circular_buffer;
|
mod circular_buffer;
|
||||||
mod log_handler;
|
mod log_handler;
|
||||||
use log_handler::{LogHandler, LogHandlerConfig};
|
use log_handler::{LogHandler, LogHandlerConfig};
|
||||||
@@ -291,20 +280,18 @@ impl Gateway {
|
|||||||
})
|
})
|
||||||
);
|
);
|
||||||
|
|
||||||
let (message, attachments) = resp.unwrap_or_else(
|
let resp = resp.unwrap_or_else(|(code, msg)| {
|
||||||
|(code, msg)| {
|
let text = format!("{code}: {msg}");
|
||||||
let text = format!("{code}: {msg}");
|
error!("Message handler error: {text}");
|
||||||
error!("Message handler error: {text}");
|
AdminMessageResponse::new(text)
|
||||||
(text, vec![])
|
});
|
||||||
}
|
|
||||||
);
|
|
||||||
|
|
||||||
let attachments = attachments.into_iter().map(|p| p.to_str().expect("attachments must have utf8 paths").to_owned()).collect();
|
let attachments = resp.attachments.into_iter().map(|p| p.to_str().expect("attachments must have utf8 paths").to_owned()).collect();
|
||||||
|
|
||||||
SignalMessage {
|
SignalMessage {
|
||||||
sender: self.config.signal_account.clone(),
|
sender: self.config.signal_account.clone(),
|
||||||
recipient: vec![msg.envelope.source_uuid.clone()],
|
recipient: vec![msg.envelope.source_uuid.clone()],
|
||||||
message,
|
message: resp.text,
|
||||||
attachments,
|
attachments,
|
||||||
}.send(signal_cli).await?;
|
}.send(signal_cli).await?;
|
||||||
}
|
}
|
||||||
@@ -316,10 +303,7 @@ impl Gateway {
|
|||||||
|
|
||||||
// Returns Err in case of a timeout or handler error
|
// Returns Err in case of a timeout or handler error
|
||||||
// Returns Ok when success or error text is generated
|
// Returns Ok when success or error text is generated
|
||||||
async fn handle_signal_admin_message(
|
async fn handle_signal_admin_message(&self, msg: &Envelope) -> MessageHandlerResult {
|
||||||
&self,
|
|
||||||
msg: &Envelope,
|
|
||||||
) -> Result<(String, Vec<PathBuf>), (u16, Box<dyn Error + Send + Sync>)> {
|
|
||||||
let data = msg.data_message.as_ref().unwrap();
|
let data = msg.data_message.as_ref().unwrap();
|
||||||
|
|
||||||
// Admin messages starting with / are handled by gateway
|
// Admin messages starting with / are handled by gateway
|
||||||
@@ -398,15 +382,12 @@ impl Gateway {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_gateway_command(
|
async fn handle_gateway_command(&self, cmd: GatewayCommand) -> MessageHandlerResult {
|
||||||
&self,
|
|
||||||
cmd: GatewayCommand,
|
|
||||||
) -> Result<(String, Vec<PathBuf>), (u16, Box<dyn Error + Send + Sync>)> {
|
|
||||||
match cmd {
|
match cmd {
|
||||||
GatewayCommand::Log { filter } => {
|
GatewayCommand::Log { filter } => {
|
||||||
let handlers = self.log_handlers.read().await;
|
let handlers = self.log_handlers.read().await;
|
||||||
if handlers.is_empty() {
|
if handlers.is_empty() {
|
||||||
return Ok(("No log sources registered yet".to_string(), vec![]));
|
return Ok(AdminMessageResponse::new("No log sources registered yet"));
|
||||||
}
|
}
|
||||||
let mut text = String::new();
|
let mut text = String::new();
|
||||||
for (origin, handler) in handlers.iter() {
|
for (origin, handler) in handlers.iter() {
|
||||||
@@ -421,9 +402,9 @@ impl Gateway {
|
|||||||
text.push('\n');
|
text.push('\n');
|
||||||
}
|
}
|
||||||
if text.is_empty() {
|
if text.is_empty() {
|
||||||
return Ok(("No matching log sources".to_string(), vec![]));
|
return Ok(AdminMessageResponse::new("No matching log sources"));
|
||||||
}
|
}
|
||||||
Ok((text, vec![]))
|
Ok(AdminMessageResponse::new(text))
|
||||||
}
|
}
|
||||||
GatewayCommand::Query { query } => {
|
GatewayCommand::Query { query } => {
|
||||||
let prometheus = self
|
let prometheus = self
|
||||||
@@ -455,7 +436,7 @@ impl Gateway {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok((text, vec![]))
|
Ok(AdminMessageResponse::new(text))
|
||||||
}
|
}
|
||||||
Err(err) => Err((500, err)),
|
Err(err) => Err((500, err)),
|
||||||
}
|
}
|
||||||
@@ -469,7 +450,7 @@ impl Gateway {
|
|||||||
|
|
||||||
prometheus.purge_old_plots();
|
prometheus.purge_old_plots();
|
||||||
match prometheus.create_oneoff_plot(query.clone(), duration).await {
|
match prometheus.create_oneoff_plot(query.clone(), duration).await {
|
||||||
Ok(filename) => Ok((query, vec![filename])),
|
Ok(filename) => Ok(AdminMessageResponse::new(query).with_attachment(filename)),
|
||||||
Err(err) => Err((500, err)),
|
Err(err) => Err((500, err)),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -493,7 +474,7 @@ impl Gateway {
|
|||||||
if text.is_empty() {
|
if text.is_empty() {
|
||||||
text = "no matches".into();
|
text = "no matches".into();
|
||||||
}
|
}
|
||||||
Ok((text, vec![]))
|
Ok(AdminMessageResponse::new(text))
|
||||||
}
|
}
|
||||||
Err(err) => Err((500, err)),
|
Err(err) => Err((500, err)),
|
||||||
}
|
}
|
||||||
@@ -514,7 +495,7 @@ impl Gateway {
|
|||||||
if text.is_empty() {
|
if text.is_empty() {
|
||||||
text = "no matches".into();
|
text = "no matches".into();
|
||||||
}
|
}
|
||||||
Ok((text, vec![]))
|
Ok(AdminMessageResponse::new(text))
|
||||||
}
|
}
|
||||||
Err(err) => Err((500, err)),
|
Err(err) => Err((500, err)),
|
||||||
}
|
}
|
||||||
@@ -554,7 +535,7 @@ impl Gateway {
|
|||||||
if text.is_empty() {
|
if text.is_empty() {
|
||||||
text = "no alerts".into();
|
text = "no alerts".into();
|
||||||
}
|
}
|
||||||
Ok((text, vec![]))
|
Ok(AdminMessageResponse::new(text))
|
||||||
}
|
}
|
||||||
Err(err) => Err((500, err)),
|
Err(err) => Err((500, err)),
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
pub mod alertmanager;
|
pub mod alertmanager;
|
||||||
pub mod gateway;
|
pub mod gateway;
|
||||||
|
pub mod message_handler;
|
||||||
|
|
||||||
pub(crate) mod human_duration;
|
pub(crate) mod human_duration;
|
||||||
pub(crate) mod jsonrpc;
|
pub(crate) mod jsonrpc;
|
||||||
@@ -7,5 +8,6 @@ pub(crate) mod log_message;
|
|||||||
pub(crate) mod prometheus;
|
pub(crate) mod prometheus;
|
||||||
pub(crate) mod transports;
|
pub(crate) mod transports;
|
||||||
|
|
||||||
pub use gateway::{Gateway, GatewayConfig, MessageHandler, MessageHandlerResult};
|
pub use gateway::{Gateway, GatewayConfig};
|
||||||
pub use log_message::{Level, LogMessage, LogMessageBuilder};
|
pub use log_message::{Level, LogMessage, LogMessageBuilder};
|
||||||
|
pub use message_handler::{AdminMessageResponse, MessageHandler, MessageHandlerResult};
|
||||||
|
|||||||
@@ -0,0 +1,88 @@
|
|||||||
|
//! Message handler types for admin messages not handled by the gateway.
|
||||||
|
|
||||||
|
use std::{
|
||||||
|
error::Error,
|
||||||
|
future::Future,
|
||||||
|
path::PathBuf,
|
||||||
|
pin::Pin,
|
||||||
|
};
|
||||||
|
|
||||||
|
/// Response to an admin message.
|
||||||
|
#[non_exhaustive]
|
||||||
|
#[derive(Clone, Debug, Default)]
|
||||||
|
pub struct AdminMessageResponse {
|
||||||
|
/// The text response to send back to the admin.
|
||||||
|
pub text: String,
|
||||||
|
/// Optional file attachments to include with the response.
|
||||||
|
pub attachments: Vec<PathBuf>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl AdminMessageResponse {
|
||||||
|
/// Create a new response with the given text.
|
||||||
|
pub fn new(text: impl Into<String>) -> Self {
|
||||||
|
Self {
|
||||||
|
text: text.into(),
|
||||||
|
attachments: Vec::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Create a builder for constructing a response.
|
||||||
|
pub fn builder() -> AdminMessageResponseBuilder {
|
||||||
|
AdminMessageResponseBuilder::default()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Add an attachment to the response.
|
||||||
|
pub fn with_attachment(mut self, path: impl Into<PathBuf>) -> Self {
|
||||||
|
self.attachments.push(path.into());
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Add multiple attachments to the response.
|
||||||
|
pub fn with_attachments(mut self, paths: impl IntoIterator<Item = impl Into<PathBuf>>) -> Self {
|
||||||
|
self.attachments.extend(paths.into_iter().map(Into::into));
|
||||||
|
self
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Builder for constructing an [`AdminMessageResponse`].
|
||||||
|
#[derive(Clone, Debug, Default)]
|
||||||
|
pub struct AdminMessageResponseBuilder {
|
||||||
|
text: String,
|
||||||
|
attachments: Vec<PathBuf>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl AdminMessageResponseBuilder {
|
||||||
|
/// Set the text response.
|
||||||
|
pub fn text(mut self, text: impl Into<String>) -> Self {
|
||||||
|
self.text = text.into();
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Add an attachment.
|
||||||
|
pub fn attachment(mut self, path: impl Into<PathBuf>) -> Self {
|
||||||
|
self.attachments.push(path.into());
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Add multiple attachments.
|
||||||
|
pub fn attachments(mut self, paths: impl IntoIterator<Item = impl Into<PathBuf>>) -> Self {
|
||||||
|
self.attachments.extend(paths.into_iter().map(Into::into));
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Build the response.
|
||||||
|
pub fn build(self) -> AdminMessageResponse {
|
||||||
|
AdminMessageResponse {
|
||||||
|
text: self.text,
|
||||||
|
attachments: self.attachments,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Result type for message handler responses.
|
||||||
|
pub type MessageHandlerResult = Result<AdminMessageResponse, (u16, Box<dyn Error + Send + Sync>)>;
|
||||||
|
|
||||||
|
/// Handler function for admin messages that don't start with `/`.
|
||||||
|
/// Takes the message text and returns a response.
|
||||||
|
pub type MessageHandler =
|
||||||
|
Box<dyn Fn(String) -> Pin<Box<dyn Future<Output = MessageHandlerResult> + Send>> + Send + Sync>;
|
||||||
Reference in New Issue
Block a user