refactor to remove generic message handlers

Simpler model where we have fixed set of protocol message handlers
with shared message and event queues.

Author: johncantrell97

.github/workflows/build.yml

@@ -31,6 +31,8 @@ jobs:
       - name: Pin serde for MSRV
         if: matrix.msrv
         run: cargo update -p serde --precise "1.0.156" --verbose
+      - name: Check documentation
+        run: cargo check --release && cargo doc --release
       - name: Build on Rust ${{ matrix.toolchain }}
         run: cargo build --verbose --color always
       - name: Check formatting

Cargo.toml

@@ -8,11 +8,11 @@ description = "Types and primitives to integrate a spec-compliant LSP with an LD
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-lightning = { version = "0.0.114", features = ["max_level_trace", "std"] }
-lightning-invoice = { version = "0.22" }
-lightning-net-tokio = { version = "0.0.114" }
+lightning = { git = "https://github.com/lightningdevkit/rust-lightning.git", rev = "498f2331459d8031031ef151a44c90d700aa8c7e", features = ["max_level_trace"] }
+lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning.git", rev = "498f2331459d8031031ef151a44c90d700aa8c7e" }
+lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning.git", rev = "498f2331459d8031031ef151a44c90d700aa8c7e" }
 
-bitcoin = "0.29.2"
+bitcoin = "0.29.0"
 
-serde = { version = "1.0", features = ["derive"] }
+serde = { version = "1.0", default-features = false, features = ["derive", "alloc"] }
 serde_json = "1.0"

src/events.rs

@@ -1,10 +1,28 @@
-use bitcoin::secp256k1::PublicKey;
+// This file is Copyright its original authors, visible in version control
+// history.
+//
+// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE
+// or http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your option.
+// You may not use this file except in accordance with one or both of these
+// licenses.
 
-pub trait EventHandler {
-	fn handle_event(&self, event: Event);
-}
+//! Events are returned from various bits in the library which indicate some action must be taken
+//! by the client.
+//!
+//! Because we don't have a built-in runtime, it's up to the client to call events at a time in the
+//! future.
+use bitcoin::secp256k1::PublicKey;
 
+/// An Event which you should probably take some action in response to.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Event {
-	ListProtocols { counterparty_node_id: PublicKey, protocols: Vec<u16> },
+	/// A response to a ListProtocols request that informs the caller
+	/// what protocols a particular node supports.
+	ListProtocols {
+		/// The counterparty's node_id that identifies the node.
+		counterparty_node_id: PublicKey,
+		/// The list of protocol numbers that are supported.
+		protocols: Vec<u16>,
+	},
 }

src/lib.rs

@@ -19,7 +19,9 @@
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
 
 mod channel_request;
-mod events;
+pub mod events;
 mod jit_channel;
 mod transport;
 mod utils;
+
+pub use transport::message_handler::{LSPConfig, LSPManager};

src/transport/message_handler.rs

@@ -1,172 +1,131 @@
-use crate::events::{Event, EventHandler};
-use crate::transport::msgs::{LSPSMessage, Prefix, RawLSPSMessage, LSPS_MESSAGE_TYPE};
+use crate::events::Event;
+use crate::transport::msgs::{LSPSMessage, RawLSPSMessage, LSPS_MESSAGE_TYPE};
+use crate::transport::protocol::LSPS0MessageHandler;
+
 use bitcoin::secp256k1::PublicKey;
+use lightning::ln::features::{InitFeatures, NodeFeatures};
 use lightning::ln::peer_handler::CustomMessageHandler;
 use lightning::ln::wire::CustomMessageReader;
 use lightning::log_info;
+use lightning::sign::EntropySource;
 use lightning::util::logger::Logger;
+use lightning::util::ser::Readable;
 use std::collections::HashMap;
-use std::convert::{TryFrom, TryInto};
+use std::convert::TryFrom;
 use std::io;
 use std::ops::Deref;
 use std::sync::{Arc, Mutex};
 
+const LSPS_FEATURE_BIT: usize = 729;
+
 /// A trait used to implement a specific LSPS protocol
 /// The messages the protocol uses need to be able to be mapped
 /// from and into LSPSMessages.
-pub trait ProtocolMessageHandler {
+pub(crate) trait ProtocolMessageHandler {
 	type ProtocolMessage: TryFrom<LSPSMessage> + Into<LSPSMessage>;
 	const PROTOCOL_NUMBER: Option<u16>;
 
 	fn handle_message(
 		&self, message: Self::ProtocolMessage, counterparty_node_id: &PublicKey,
 	) -> Result<(), lightning::ln::msgs::LightningError>;
-	fn get_and_clear_pending_protocol_messages(&self) -> Vec<(PublicKey, Self::ProtocolMessage)>;
-	fn get_and_clear_pending_protocol_events(&self) -> Vec<Event>;
-	fn get_protocol_number(&self) -> Option<u16> {
-		Self::PROTOCOL_NUMBER
-	}
 }
 
-/// A trait used to implement the mapping from a LSPS transport layer mesage
-/// to a specific protocol message. This enables the ProtocolMessageHandler's
-/// to not need to know about LSPSMessage and only have to deal with the specific
-/// messages related to the protocol that is being implemented.
-pub trait TransportMessageHandler {
-	fn handle_lsps_message(
-		&self, message: LSPSMessage, counterparty_node_id: &PublicKey,
-	) -> Result<(), lightning::ln::msgs::LightningError>;
-	fn get_and_clear_pending_msg(&self) -> Vec<(PublicKey, LSPSMessage)>;
-	fn get_and_clear_pending_events(&self) -> Vec<Event>;
-	fn get_protocol_number(&self) -> Option<u16>;
+/// Configuration for the LSPManager
+/// Allows end-user to configure the behavior of the client
+pub struct LSPConfig {
+	/// whether or not you intend to provide services
+	is_lsp: bool,
 }
 
-impl<T> TransportMessageHandler for T
-where
-	T: ProtocolMessageHandler,
-	LSPSMessage: TryInto<<T as ProtocolMessageHandler>::ProtocolMessage>,
-{
-	fn handle_lsps_message(
-		&self, message: LSPSMessage, counterparty_node_id: &PublicKey,
-	) -> Result<(), lightning::ln::msgs::LightningError> {
-		if let Ok(protocol_message) = message.try_into() {
-			self.handle_message(protocol_message, counterparty_node_id)?;
-		}
-
-		Ok(())
-	}
-
-	fn get_and_clear_pending_msg(&self) -> Vec<(PublicKey, LSPSMessage)> {
-		self.get_and_clear_pending_protocol_messages()
-			.into_iter()
-			.map(|(public_key, protocol_message)| (public_key, protocol_message.into()))
-			.collect()
-	}
-
-	fn get_and_clear_pending_events(&self) -> Vec<Event> {
-		self.get_and_clear_pending_protocol_events()
-	}
-
-	fn get_protocol_number(&self) -> Option<u16> {
-		self.get_protocol_number()
-	}
-}
-
-pub struct LSPManager<L: Deref>
+/// The main interface into LSP functionality
+pub struct LSPManager<L: Deref, ES: Deref>
 where
 	L::Target: Logger,
+	ES::Target: EntropySource,
 {
 	logger: L,
-	pending_messages: Mutex<Vec<(PublicKey, RawLSPSMessage)>>,
+	pending_messages: Arc<Mutex<Vec<(PublicKey, LSPSMessage)>>>,
+	pending_events: Arc<Mutex<Vec<Event>>>,
 	request_id_to_method_map: Mutex<HashMap<String, String>>,
-	message_handlers: Arc<Mutex<HashMap<Prefix, Arc<dyn TransportMessageHandler>>>>,
+	lsps0_message_handler: LSPS0MessageHandler<ES>,
+	config: LSPConfig,
 }
 
-impl<L: Deref> LSPManager<L>
+impl<L: Deref, ES: Deref> LSPManager<L, ES>
 where
 	L::Target: Logger,
+	ES::Target: EntropySource,
 {
-	pub fn new(logger: L) -> Self {
+	/// Constructor for the LSPManager
+	///
+	/// Sets up all required protocol message handlers based on configuration passed in
+	pub fn new(logger: L, entropy_source: ES, config: LSPConfig) -> Self {
+		let pending_messages = Arc::new(Mutex::new(vec![]));
+		let pending_events = Arc::new(Mutex::new(vec![]));
+
+		let lsps0_message_handler = LSPS0MessageHandler::new(
+			entropy_source,
+			vec![],
+			pending_messages.clone(),
+			pending_events.clone(),
+		);
+
 		Self {
 			logger,
-			pending_messages: Mutex::new(Vec::new()),
+			pending_messages,
+			pending_events,
 			request_id_to_method_map: Mutex::new(HashMap::new()),
-			message_handlers: Arc::new(Mutex::new(HashMap::new())),
+			lsps0_message_handler,
+			config,
 		}
 	}
 
-	pub fn get_message_handlers(
-		&self,
-	) -> Arc<Mutex<HashMap<Prefix, Arc<dyn TransportMessageHandler>>>> {
-		self.message_handlers.clone()
-	}
-
-	pub fn register_message_handler(
-		&self, prefix: Prefix, message_handler: Arc<dyn TransportMessageHandler>,
-	) {
-		self.message_handlers.lock().unwrap().insert(prefix, message_handler);
-	}
-
-	pub fn process_pending_events<H: EventHandler>(&self, handler: H) {
-		let message_handlers = self.message_handlers.lock().unwrap();
-
-		for message_handler in message_handlers.values() {
-			let events = message_handler.get_and_clear_pending_events();
-			for event in events {
-				handler.handle_event(event);
-			}
-		}
+	/// Needs to be polled regularly to surface events generated by the various protocols
+	pub fn get_and_clear_pending_events(&self) -> Vec<Event> {
+		self.pending_events.lock().unwrap().drain(..).collect()
 	}
 
 	fn handle_lsps_message(
 		&self, msg: LSPSMessage, sender_node_id: &PublicKey,
 	) -> Result<(), lightning::ln::msgs::LightningError> {
-		if let Some(prefix) = msg.prefix() {
-			let message_handlers = self.message_handlers.lock().unwrap();
-			// TODO: not sure what we are supposed to do when we receive a message we don't have a handler for
-			if let Some(message_handler) = message_handlers.get(&prefix) {
-				message_handler.handle_lsps_message(msg, sender_node_id)?;
-			} else {
-				log_info!(
-					self.logger,
-					"Received a message from {:?} we do not have a handler for: {:?}",
-					sender_node_id,
-					msg
-				);
+		match msg {
+			LSPSMessage::Invalid => {
+				log_info!(self.logger, "Received invalid message from {:?}", sender_node_id);
+			}
+			LSPSMessage::LSPS0(msg) => {
+				self.lsps0_message_handler.handle_message(msg, sender_node_id)?;
 			}
 		}
 		Ok(())
 	}
 
-	fn enqueue_message(&self, node_id: PublicKey, msg: RawLSPSMessage) {
+	fn enqueue_message(&self, node_id: PublicKey, msg: LSPSMessage) {
 		let mut pending_msgs = self.pending_messages.lock().unwrap();
 		pending_msgs.push((node_id, msg));
 	}
 }
 
-impl<L: Deref> CustomMessageReader for LSPManager<L>
+impl<L: Deref, ES: Deref> CustomMessageReader for LSPManager<L, ES>
 where
 	L::Target: Logger,
+	ES::Target: EntropySource,
 {
 	type CustomMessage = RawLSPSMessage;
 
 	fn read<R: io::Read>(
 		&self, message_type: u16, buffer: &mut R,
 	) -> Result<Option<Self::CustomMessage>, lightning::ln::msgs::DecodeError> {
 		match message_type {
-			LSPS_MESSAGE_TYPE => {
-				let mut payload = String::new();
-				buffer.read_to_string(&mut payload)?;
-				Ok(Some(RawLSPSMessage { payload }))
-			}
+			LSPS_MESSAGE_TYPE => Ok(Some(RawLSPSMessage::read(buffer)?)),
 			_ => Ok(None),
 		}
 	}
 }
 
-impl<L: Deref> CustomMessageHandler for LSPManager<L>
+impl<L: Deref, ES: Deref> CustomMessageHandler for LSPManager<L, ES>
 where
 	L::Target: Logger,
+	ES::Target: EntropySource,
 {
 	fn handle_custom_message(
 		&self, msg: Self::CustomMessage, sender_node_id: &PublicKey,
@@ -176,35 +135,47 @@ where
 		match LSPSMessage::from_str_with_id_map(&msg.payload, &mut request_id_to_method_map) {
 			Ok(msg) => self.handle_lsps_message(msg, sender_node_id),
 			Err(_) => {
-				self.enqueue_message(
-					*sender_node_id,
-					RawLSPSMessage {
-						payload: serde_json::to_string(&LSPSMessage::Invalid).unwrap(),
-					},
-				);
+				self.enqueue_message(*sender_node_id, LSPSMessage::Invalid);
 				Ok(())
 			}
 		}
 	}
 
 	fn get_and_clear_pending_msg(&self) -> Vec<(PublicKey, Self::CustomMessage)> {
-		let mut msgs = vec![];
+		let mut request_id_to_method_map = self.request_id_to_method_map.lock().unwrap();
+		self.pending_messages
+			.lock()
+			.unwrap()
+			.drain(..)
+			.map(|(public_key, lsps_message)| {
+				if let Some((request_id, method_name)) = lsps_message.get_request_id_and_method() {
+					request_id_to_method_map.insert(request_id, method_name);
+				}
+				(
+					public_key,
+					RawLSPSMessage { payload: serde_json::to_string(&lsps_message).unwrap() },
+				)
+			})
+			.collect()
+	}
 
-		{
-			let mut pending_messages = self.pending_messages.lock().unwrap();
-			msgs.extend(
-				pending_messages.drain(..).collect::<Vec<(PublicKey, Self::CustomMessage)>>(),
-			);
+	fn provided_node_features(&self) -> NodeFeatures {
+		let mut features = NodeFeatures::empty();
+
+		if self.config.is_lsp {
+			features.set_optional_custom_bit(LSPS_FEATURE_BIT).unwrap();
 		}
 
-		let message_handlers = self.message_handlers.lock().unwrap();
-		for message_handler in message_handlers.values() {
-			let protocol_messages = message_handler.get_and_clear_pending_msg();
-			msgs.extend(protocol_messages.into_iter().map(|(node_id, message)| {
-				(node_id, RawLSPSMessage { payload: serde_json::to_string(&message).unwrap() })
-			}));
+		features
+	}
+
+	fn provided_init_features(&self, _their_node_id: &PublicKey) -> InitFeatures {
+		let mut features = InitFeatures::empty();
+
+		if self.config.is_lsp {
+			features.set_optional_custom_bit(LSPS_FEATURE_BIT).unwrap();
 		}
 
-		msgs
+		features
 	}
 }