diff --git a/src/admin/handler.rs b/src/admin/handler.rs
index e283def55772ec01fbf2fb94ebf9fd73e6c03cb2..85270db2ccc5d3e30575bd857cfa36c95c291b9e 100644
--- a/src/admin/handler.rs
+++ b/src/admin/handler.rs
@@ -12,8 +12,8 @@
extern crate conduit_service as service;
use conduit::Result;
-use service::admin::HandlerResult;
-pub(crate) use service::admin::{AdminEvent, Service};
+pub(crate) use service::admin::{Command, Service};
+use service::admin::{CommandResult, HandlerResult};
use self::{fsck::FsckCommand, tester::TesterCommands};
use crate::{
@@ -68,25 +68,18 @@ pub(crate) enum AdminCommand {
}
#[must_use]
-pub fn handle(event: AdminEvent) -> HandlerResult { Box::pin(handle_event(event)) }
+pub fn handle(command: Command) -> HandlerResult { Box::pin(handle_command(command)) }
#[tracing::instrument(skip_all, name = "admin")]
-async fn handle_event(event: AdminEvent) -> Result<AdminEvent> { Ok(AdminEvent::Reply(process_event(event).await)) }
-
-async fn process_event(event: AdminEvent) -> Option<RoomMessageEventContent> {
- let (mut message_content, reply_id) = match event {
- AdminEvent::Command(room_message, reply_id) => (Box::pin(process_admin_message(room_message)).await, reply_id),
- AdminEvent::Notice(content) => (content, None),
- AdminEvent::Reply(_) => return None,
- };
-
- message_content.relates_to = reply_id.map(|reply_id| Reply {
+async fn handle_command(command: Command) -> CommandResult {
+ let mut content = process_admin_message(command.command).await;
+ content.relates_to = command.reply_id.map(|event_id| Reply {
in_reply_to: InReplyTo {
- event_id: reply_id.into(),
+ event_id,
},
});
- Some(message_content)
+ Ok(Some(content))
}
// Parse and process a message from the admin room
diff --git a/src/service/admin/mod.rs b/src/service/admin/mod.rs
index dca3ac8d73bde65639dec7c3a27ebdbe9e741b4b..ddc3606f45e49ee182745046d0c3776f11ce7736 100644
--- a/src/service/admin/mod.rs
+++ b/src/service/admin/mod.rs
@@ -7,12 +7,13 @@
use conduit::{utils::mutex_map, Error, Result};
pub use create::create_admin_room;
pub use grant::make_user_admin;
+use loole::{Receiver, Sender};
use ruma::{
events::{
room::message::{Relation, RoomMessageEventContent},
TimelineEventType,
},
- EventId, OwnedRoomId, RoomId, UserId,
+ OwnedEventId, OwnedRoomId, RoomId, UserId,
};
use serde_json::value::to_raw_value;
use tokio::{sync::Mutex, task::JoinHandle};
@@ -20,12 +21,16 @@
use crate::{pdu::PduBuilder, services, user_is_local, PduEvent};
-pub type HandlerResult = Pin<Box<dyn Future<Output = Result<AdminEvent, Error>> + Send>>;
-pub type Handler = fn(AdminEvent) -> HandlerResult;
+const COMMAND_QUEUE_LIMIT: usize = 512;
+
+pub type CommandOutput = Option<RoomMessageEventContent>;
+pub type CommandResult = Result<CommandOutput, Error>;
+pub type HandlerResult = Pin<Box<dyn Future<Output = CommandResult> + Send>>;
+pub type Handler = fn(Command) -> HandlerResult;
pub struct Service {
- sender: loole::Sender<AdminEvent>,
- receiver: Mutex<loole::Receiver<AdminEvent>>,
+ sender: Sender<Command>,
+ receiver: Mutex<Receiver<Command>>,
handler_join: Mutex<Option<JoinHandle<()>>>,
pub handle: Mutex<Option<Handler>>,
#[cfg(feature = "console")]
@@ -33,16 +38,15 @@ pub struct Service {
}
#[derive(Debug)]
-pub enum AdminEvent {
- Command(String, Option<Arc<EventId>>),
- Reply(Option<RoomMessageEventContent>),
- Notice(RoomMessageEventContent),
+pub struct Command {
+ pub command: String,
+ pub reply_id: Option<OwnedEventId>,
}
impl Service {
#[must_use]
pub fn build() -> Arc<Self> {
- let (sender, receiver) = loole::unbounded();
+ let (sender, receiver) = loole::bounded(COMMAND_QUEUE_LIMIT);
Arc::new(Self {
sender,
receiver: Mutex::new(receiver),
@@ -53,6 +57,18 @@ pub fn build() -> Arc<Self> {
})
}
+ pub async fn start_handler(self: &Arc<Self>) {
+ let self_ = Arc::clone(self);
+ let handle = services().server.runtime().spawn(async move {
+ self_
+ .handler()
+ .await
+ .expect("Failed to initialize admin room handler");
+ });
+
+ _ = self.handler_join.lock().await.insert(handle);
+ }
+
pub fn interrupt(&self) {
#[cfg(feature = "console")]
self.console.interrupt();
@@ -75,16 +91,39 @@ pub async fn close(&self) {
}
}
- pub async fn start_handler(self: &Arc<Self>) {
- let self_ = Arc::clone(self);
- let handle = services().server.runtime().spawn(async move {
- self_
- .handler()
- .await
- .expect("Failed to initialize admin room handler");
- });
+ pub async fn send_text(&self, body: &str) {
+ self.send_message(RoomMessageEventContent::text_plain(body))
+ .await;
+ }
- _ = self.handler_join.lock().await.insert(handle);
+ pub async fn send_message(&self, message_content: RoomMessageEventContent) {
+ if let Ok(Some(room_id)) = Self::get_admin_room() {
+ let user_id = &services().globals.server_user;
+ respond_to_room(message_content, &room_id, user_id).await;
+ }
+ }
+
+ pub async fn command(&self, command: String, reply_id: Option<OwnedEventId>) {
+ self.send(Command {
+ command,
+ reply_id,
+ })
+ .await;
+ }
+
+ pub async fn command_in_place(
+ &self, command: String, reply_id: Option<OwnedEventId>,
+ ) -> Result<Option<RoomMessageEventContent>> {
+ self.process_command(Command {
+ command,
+ reply_id,
+ })
+ .await
+ }
+
+ async fn send(&self, message: Command) {
+ debug_assert!(!self.sender.is_closed(), "channel closed");
+ self.sender.send_async(message).await.expect("message sent");
}
async fn handler(self: &Arc<Self>) -> Result<()> {
@@ -93,8 +132,8 @@ async fn handler(self: &Arc<Self>) -> Result<()> {
loop {
debug_assert!(!receiver.is_closed(), "channel closed");
tokio::select! {
- event = receiver.recv_async() => match event {
- Ok(event) => self.receive(event).await,
+ command = receiver.recv_async() => match command {
+ Ok(command) => self.handle_command(command).await,
Err(_) => return Ok(()),
},
sig = signals.recv() => match sig {
@@ -105,55 +144,27 @@ async fn handler(self: &Arc<Self>) -> Result<()> {
}
}
- pub async fn send_text(&self, body: &str) {
- self.send_message(RoomMessageEventContent::text_plain(body))
- .await;
- }
-
- pub async fn send_message(&self, message_content: RoomMessageEventContent) {
- self.send(AdminEvent::Notice(message_content)).await;
- }
-
- pub async fn command(&self, command: String, event_id: Option<Arc<EventId>>) {
- self.send(AdminEvent::Command(command, event_id)).await;
- }
-
- pub async fn command_in_place(
- &self, command: String, event_id: Option<Arc<EventId>>,
- ) -> Result<Option<RoomMessageEventContent>> {
- match self.handle(AdminEvent::Command(command, event_id)).await? {
- AdminEvent::Reply(content) => Ok(content),
- _ => Ok(None),
+ async fn handle_signal(&self, #[allow(unused_variables)] sig: &'static str) {
+ #[cfg(feature = "console")]
+ if sig == "SIGINT" && services().server.running() {
+ self.console.start().await;
}
}
- async fn send(&self, message: AdminEvent) {
- debug_assert!(!self.sender.is_full(), "channel full");
- debug_assert!(!self.sender.is_closed(), "channel closed");
- self.sender.send(message).expect("message sent");
- }
-
- async fn receive(&self, event: AdminEvent) {
- if let Ok(AdminEvent::Reply(content)) = self.handle(event).await {
- handle_response(content).await;
+ async fn handle_command(&self, command: Command) {
+ if let Ok(Some(output)) = self.process_command(command).await {
+ handle_response(output).await;
}
}
- async fn handle(&self, event: AdminEvent) -> Result<AdminEvent, Error> {
+ async fn process_command(&self, command: Command) -> CommandResult {
if let Some(handle) = self.handle.lock().await.as_ref() {
- handle(event).await
+ handle(command).await
} else {
Err(Error::Err("Admin module is not loaded.".into()))
}
}
- async fn handle_signal(&self, #[allow(unused_variables)] sig: &'static str) {
- #[cfg(feature = "console")]
- if sig == "SIGINT" && services().server.running() {
- self.console.start().await;
- }
- }
-
/// Checks whether a given user is an admin of this server
pub async fn user_is_admin(&self, user_id: &UserId) -> Result<bool> {
if let Ok(Some(admin_room)) = Self::get_admin_room() {
@@ -186,26 +197,28 @@ pub fn get_admin_room() -> Result<Option<OwnedRoomId>> {
}
}
-async fn handle_response(content: Option<RoomMessageEventContent>) {
- if let Some(content) = content.as_ref() {
- if let Some(Relation::Reply {
- in_reply_to,
- }) = content.relates_to.as_ref()
- {
- if let Ok(Some(pdu)) = services().rooms.timeline.get_pdu(&in_reply_to.event_id) {
- let response_sender = if is_admin_room(&pdu.room_id) {
- &services().globals.server_user
- } else {
- &pdu.sender
- };
-
- respond_to_room(content, &pdu.room_id, response_sender).await;
- }
- }
- }
+async fn handle_response(content: RoomMessageEventContent) {
+ let Some(Relation::Reply {
+ in_reply_to,
+ }) = content.relates_to.as_ref()
+ else {
+ return;
+ };
+
+ let Ok(Some(pdu)) = services().rooms.timeline.get_pdu(&in_reply_to.event_id) else {
+ return;
+ };
+
+ let response_sender = if is_admin_room(&pdu.room_id) {
+ &services().globals.server_user
+ } else {
+ &pdu.sender
+ };
+
+ respond_to_room(content, &pdu.room_id, response_sender).await;
}
-async fn respond_to_room(content: &RoomMessageEventContent, room_id: &RoomId, user_id: &UserId) {
+async fn respond_to_room(content: RoomMessageEventContent, room_id: &RoomId, user_id: &UserId) {
assert!(
services()
.admin
@@ -218,7 +231,7 @@ async fn respond_to_room(content: &RoomMessageEventContent, room_id: &RoomId, us
let state_lock = services().globals.roomid_mutex_state.lock(room_id).await;
let response_pdu = PduBuilder {
event_type: TimelineEventType::RoomMessage,
- content: to_raw_value(content).expect("event is valid, we just created it"),
+ content: to_raw_value(&content).expect("event is valid, we just created it"),
unsigned: None,
state_key: None,
redacts: None,
diff --git a/src/service/rooms/timeline/mod.rs b/src/service/rooms/timeline/mod.rs
index e8774cdc723911a8a002dd39fc7890f9570049bf..ed3693349d12dd45403b28b953b6c4072f49f364 100644
--- a/src/service/rooms/timeline/mod.rs
+++ b/src/service/rooms/timeline/mod.rs
@@ -475,7 +475,7 @@ pub async fn append_pdu(
if admin::is_admin_command(pdu, &body).await {
services()
.admin
- .command(body, Some(pdu.event_id.clone()))
+ .command(body, Some((*pdu.event_id).into()))
.await;
}
}