Introduce MessageQueue trait to DRY up enqueing and PM triggering

Author: tnull

src/lib.rs

@@ -44,7 +44,10 @@ pub mod lsps0;
 pub mod lsps1;
 pub mod lsps2;
 mod manager;
+pub mod message_queue;
 mod sync;
+#[cfg(test)]
+mod tests;
 mod utils;
 
 pub use manager::{LiquidityClientConfig, LiquidityManager, LiquidityServiceConfig};

src/lsps0/client.rs

@@ -7,11 +7,11 @@
 use crate::events::{Event, EventQueue};
 use crate::lsps0::event::LSPS0ClientEvent;
 use crate::lsps0::msgs::{
-	LSPS0Message, LSPS0Request, LSPS0Response, LSPSMessage, ListProtocolsRequest,
-	ListProtocolsResponse, ProtocolMessageHandler, ResponseError,
+	LSPS0Message, LSPS0Request, LSPS0Response, ListProtocolsRequest, ListProtocolsResponse,
+	ProtocolMessageHandler, ResponseError,
 };
-use crate::prelude::Vec;
-use crate::sync::{Arc, Mutex};
+use crate::message_queue::MessageQueue;
+use crate::sync::Arc;
 use crate::utils;
 
 use lightning::ln::msgs::{ErrorAction, LightningError};
@@ -23,23 +23,24 @@ use bitcoin::secp256k1::PublicKey;
 use core::ops::Deref;
 
 /// A message handler capable of sending and handling LSPS0 messages.
-pub struct LSPS0ClientHandler<ES: Deref>
+pub struct LSPS0ClientHandler<ES: Deref, MQ: Deref>
 where
 	ES::Target: EntropySource,
+	MQ::Target: MessageQueue,
 {
 	entropy_source: ES,
-	pending_messages: Arc<Mutex<Vec<(PublicKey, LSPSMessage)>>>,
+	pending_messages: MQ,
 	pending_events: Arc<EventQueue>,
 }
 
-impl<ES: Deref> LSPS0ClientHandler<ES>
+impl<ES: Deref, MQ: Deref> LSPS0ClientHandler<ES, MQ>
 where
 	ES::Target: EntropySource,
+	MQ::Target: MessageQueue,
 {
 	/// Returns a new instance of [`LSPS0ClientHandler`].
 	pub(crate) fn new(
-		entropy_source: ES, pending_messages: Arc<Mutex<Vec<(PublicKey, LSPSMessage)>>>,
-		pending_events: Arc<EventQueue>,
+		entropy_source: ES, pending_messages: MQ, pending_events: Arc<EventQueue>,
 	) -> Self {
 		Self { entropy_source, pending_messages, pending_events }
 	}
@@ -49,17 +50,13 @@ where
 	/// Please refer to the [LSPS0
 	/// specifcation](https://github.com/BitcoinAndLightningLayerSpecs/lsp/tree/main/LSPS0#lsps-specification-support-query)
 	/// for more information.
-	pub fn list_protocols(&self, counterparty_node_id: PublicKey) {
+	pub fn list_protocols(&self, counterparty_node_id: &PublicKey) {
 		let msg = LSPS0Message::Request(
 			utils::generate_request_id(&self.entropy_source),
 			LSPS0Request::ListProtocols(ListProtocolsRequest {}),
 		);
 
-		self.enqueue_message(counterparty_node_id, msg);
-	}
-
-	fn enqueue_message(&self, counterparty_node_id: PublicKey, message: LSPS0Message) {
-		self.pending_messages.lock().unwrap().push((counterparty_node_id, message.into()));
+		self.pending_messages.enqueue(counterparty_node_id, msg.into());
 	}
 
 	fn handle_response(
@@ -88,9 +85,10 @@ where
 	}
 }
 
-impl<ES: Deref> ProtocolMessageHandler for LSPS0ClientHandler<ES>
+impl<ES: Deref, MQ: Deref> ProtocolMessageHandler for LSPS0ClientHandler<ES, MQ>
 where
 	ES::Target: EntropySource,
+	MQ::Target: MessageQueue,
 {
 	type ProtocolMessage = LSPS0Message;
 	const PROTOCOL_NUMBER: Option<u16> = None;
@@ -119,20 +117,14 @@ mod tests {
 	use alloc::string::ToString;
 	use alloc::sync::Arc;
 
-	use crate::lsps0::msgs::RequestId;
+	use crate::lsps0::msgs::{LSPSMessage, RequestId};
+	use crate::tests::utils::{TestEntropy, TestMessageQueue};
 
 	use super::*;
 
-	struct TestEntropy {}
-	impl EntropySource for TestEntropy {
-		fn get_secure_random_bytes(&self) -> [u8; 32] {
-			[0; 32]
-		}
-	}
-
 	#[test]
 	fn test_list_protocols() {
-		let pending_messages = Arc::new(Mutex::new(vec![]));
+		let pending_messages = Arc::new(TestMessageQueue::new());
 		let entropy_source = Arc::new(TestEntropy {});
 		let event_queue = Arc::new(EventQueue::new());
 
@@ -147,8 +139,8 @@ mod tests {
 		)
 		.unwrap();
 
-		lsps0_handler.list_protocols(counterparty_node_id);
-		let pending_messages = pending_messages.lock().unwrap();
+		lsps0_handler.list_protocols(&counterparty_node_id);
+		let pending_messages = pending_messages.get_and_clear_pending_msgs();
 
 		assert_eq!(pending_messages.len(), 1);
 

src/lsps0/service.rs

@@ -14,35 +14,37 @@
 //! information.
 
 use crate::lsps0::msgs::{
-	LSPS0Message, LSPS0Request, LSPS0Response, LSPSMessage, ListProtocolsResponse,
-	ProtocolMessageHandler, RequestId,
+	LSPS0Message, LSPS0Request, LSPS0Response, ListProtocolsResponse, ProtocolMessageHandler,
+	RequestId,
 };
+use crate::message_queue::MessageQueue;
 use crate::prelude::Vec;
-use crate::sync::{Arc, Mutex};
 
 use lightning::ln::msgs::{ErrorAction, LightningError};
 use lightning::util::logger::Level;
 
 use bitcoin::secp256k1::PublicKey;
 
+use core::ops::Deref;
+
 /// The main server-side object allowing to send and receive LSPS0 messages.
-pub struct LSPS0ServiceHandler {
-	pending_messages: Arc<Mutex<Vec<(PublicKey, LSPSMessage)>>>,
+pub struct LSPS0ServiceHandler<MQ: Deref>
+where
+	MQ::Target: MessageQueue,
+{
+	pending_messages: MQ,
 	protocols: Vec<u16>,
 }
 
-impl LSPS0ServiceHandler {
+impl<MQ: Deref> LSPS0ServiceHandler<MQ>
+where
+	MQ::Target: MessageQueue,
+{
 	/// Returns a new instance of [`LSPS0ServiceHandler`].
-	pub(crate) fn new(
-		protocols: Vec<u16>, pending_messages: Arc<Mutex<Vec<(PublicKey, LSPSMessage)>>>,
-	) -> Self {
+	pub(crate) fn new(protocols: Vec<u16>, pending_messages: MQ) -> Self {
 		Self { protocols, pending_messages }
 	}
 
-	fn enqueue_message(&self, counterparty_node_id: PublicKey, message: LSPS0Message) {
-		self.pending_messages.lock().unwrap().push((counterparty_node_id, message.into()));
-	}
-
 	fn handle_request(
 		&self, request_id: RequestId, request: LSPS0Request, counterparty_node_id: &PublicKey,
 	) -> Result<(), lightning::ln::msgs::LightningError> {
@@ -54,14 +56,17 @@ impl LSPS0ServiceHandler {
 						protocols: self.protocols.clone(),
 					}),
 				);
-				self.enqueue_message(*counterparty_node_id, msg);
+				self.pending_messages.enqueue(counterparty_node_id, msg.into());
 				Ok(())
 			}
 		}
 	}
 }
 
-impl ProtocolMessageHandler for LSPS0ServiceHandler {
+impl<MQ: Deref> ProtocolMessageHandler for LSPS0ServiceHandler<MQ>
+where
+	MQ::Target: MessageQueue,
+{
 	type ProtocolMessage = LSPS0Message;
 	const PROTOCOL_NUMBER: Option<u16> = None;
 
@@ -86,7 +91,8 @@ impl ProtocolMessageHandler for LSPS0ServiceHandler {
 #[cfg(test)]
 mod tests {
 
-	use crate::lsps0::msgs::ListProtocolsRequest;
+	use crate::lsps0::msgs::{LSPSMessage, ListProtocolsRequest};
+	use crate::tests::utils::TestMessageQueue;
 	use crate::utils;
 	use alloc::string::ToString;
 	use alloc::sync::Arc;
@@ -96,7 +102,7 @@ mod tests {
 	#[test]
 	fn test_handle_list_protocols_request() {
 		let protocols: Vec<u16> = vec![];
-		let pending_messages = Arc::new(Mutex::new(vec![]));
+		let pending_messages = Arc::new(TestMessageQueue::new());
 
 		let lsps0_handler = Arc::new(LSPS0ServiceHandler::new(protocols, pending_messages.clone()));
 
@@ -110,7 +116,7 @@ mod tests {
 		.unwrap();
 
 		lsps0_handler.handle_message(list_protocols_request, &counterparty_node_id).unwrap();
-		let pending_messages = pending_messages.lock().unwrap();
+		let pending_messages = pending_messages.get_and_clear_pending_msgs();
 
 		assert_eq!(pending_messages.len(), 1);