rewrite admin handler to use loole channels

Cargo.lock

@@ -484,6 +484,7 @@ dependencies = [
  "itertools 0.12.1",
  "jsonwebtoken",
  "log",
+ "loole",
  "lru-cache",
  "nix",
  "num_cpus",
@@ -1473,6 +1474,12 @@ version = "0.4.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "90ed8c1e510134f979dbc4f070f87d4313098b704861a105fe34231c70a3901c"
 
+[[package]]
+name = "loole"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c6725f0feab07fcf90f6de5417c06d7fef976fa6e5912fa9e21cb5e4dc6ae5da"
+
 [[package]]
 name = "lru-cache"
 version = "0.1.2"

Cargo.toml

@@ -70,6 +70,9 @@ cyborgtime = "2.1.1"
 bytes = "1.6.0"
 http = "0.2.12"
 
+# used to replace the channels of the tokio runtime
+loole = "0.3.0"
+
 # standard date and time tools
 [dependencies.chrono]
 version = "0.4.37"

src/service/admin/mod.rs

@@ -23,7 +23,7 @@ use ruma::{
 	EventId, MxcUri, OwnedRoomAliasId, OwnedRoomId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
 };
 use serde_json::value::to_raw_value;
-use tokio::sync::{mpsc, Mutex};
+use tokio::sync::Mutex;
 use tracing::{error, warn};
 
 use super::pdu::PduBuilder;
@@ -91,13 +91,13 @@ pub enum AdminRoomEvent {
 }
 
 pub struct Service {
-	pub sender: mpsc::UnboundedSender<AdminRoomEvent>,
-	receiver: Mutex<mpsc::UnboundedReceiver<AdminRoomEvent>>,
+	pub sender: loole::Sender<AdminRoomEvent>, //mpsc::UnboundedSender<AdminRoomEvent>,
+	receiver: Mutex<loole::Receiver<AdminRoomEvent>>,
 }
 
 impl Service {
 	pub fn build() -> Arc<Self> {
-		let (sender, receiver) = mpsc::unbounded_channel();
+		let (sender, receiver) = loole::unbounded();
 		Arc::new(Self {
 			sender,
 			receiver: Mutex::new(receiver),
@@ -115,7 +115,7 @@ impl Service {
 	}
 
 	async fn handler(&self) -> Result<()> {
-		let mut receiver = self.receiver.lock().await;
+		let receiver = self.receiver.lock().await;
 		// TODO: Use futures when we have long admin commands
 		//let mut futures = FuturesUnordered::new();
 
@@ -125,7 +125,9 @@ impl Service {
 		if let Ok(Some(conduit_room)) = Self::get_admin_room() {
 			loop {
 				tokio::select! {
-					Some(event) = receiver.recv() => {
+						event = receiver.recv_async() => {
+							match event {
+								Ok(event) => {
 									let (mut message_content, reply) = match event {
 										AdminRoomEvent::SendMessage(content) => (content, None),
 										AdminRoomEvent::ProcessMessage(room_message, reply_id) => {
@@ -179,10 +181,17 @@ impl Service {
 											&state_lock)
 										  .await?;
 									}
-
-
 									drop(state_lock);
 							}
+							Err(_) => {
+								// TODO: Handle error, Im too unfamiliar with the codebase to know what to do here
+
+
+								// recv_async returns an error if all senders have been dropped. If the channel is empty, the returned future will yield to the async runtime.
+
+							}
+						}
+					}
 				}
 			}
 		}
@@ -760,13 +769,19 @@ mod test {
 	use super::*;
 
 	#[test]
-	fn get_help_short() { get_help_inner("-h"); }
+	fn get_help_short() {
+		get_help_inner("-h");
+	}
 
 	#[test]
-	fn get_help_long() { get_help_inner("--help"); }
+	fn get_help_long() {
+		get_help_inner("--help");
+	}
 
 	#[test]
-	fn get_help_subcommand() { get_help_inner("help"); }
+	fn get_help_subcommand() {
+		get_help_inner("help");
+	}
 
 	fn get_help_inner(input: &str) {
 		let error = AdminCommand::try_parse_from(["argv[0] doesn't matter", input])