(4/4) Merge upstream/master

lightning-net-tokio/src/lib.rs

@@ -37,7 +37,7 @@
 //! type ChainWatchInterface = dyn lightning::chain::chaininterface::ChainWatchInterface;
 //! type ChannelMonitor = lightning::ln::channelmonitor::SimpleManyChannelMonitor<lightning::chain::transaction::OutPoint, lightning::chain::keysinterface::InMemoryChannelKeys, Arc<TxBroadcaster>, Arc<FeeEstimator>, Arc<Logger>, Arc<ChainWatchInterface>>;
 //! type ChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<ChannelMonitor, TxBroadcaster, FeeEstimator, Logger>;
-//! type PeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChannelMonitor, TxBroadcaster, FeeEstimator, ChainWatchInterface, Logger>;
+//! type PeerManager = lightning::ln::peers::handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChannelMonitor, TxBroadcaster, FeeEstimator, ChainWatchInterface, Logger>;
 //!
 //! // Connect to node with pubkey their_node_id at addr:
 //! async fn connect_to_node(peer_manager: PeerManager, channel_monitor: Arc<ChannelMonitor>, channel_manager: ChannelManager, their_node_id: PublicKey, addr: SocketAddr) {
@@ -77,8 +77,8 @@ use tokio::{io, time};
 use tokio::sync::mpsc;
 use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
-use lightning::ln::peer_handler;
-use lightning::ln::peer_handler::SocketDescriptor as LnSocketTrait;
+use lightning::ln::peers::handler;
+use lightning::ln::peers::handler::SocketDescriptor as LnSocketTrait;
 use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler};
 use lightning::util::logger::Logger;
 
@@ -133,7 +133,7 @@ impl Connection {
 			_ => panic!()
 		}
 	}
-	async fn schedule_read<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
+	async fn schedule_read<CMH, RMH, L>(peer_manager: Arc<handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
 			CMH: ChannelMessageHandler + 'static,
 			RMH: RoutingMessageHandler + 'static,
 			L: Logger + 'static + ?Sized {
@@ -189,17 +189,20 @@ impl Connection {
 					Ok(len) => {
 						prepare_read_write_call!();
 						let read_res = peer_manager.read_event(&mut our_descriptor, &buf[0..len]);
-						let mut us_lock = us.lock().unwrap();
-						match read_res {
-							Ok(pause_read) => {
-								if pause_read {
-									us_lock.read_paused = true;
-								}
-								Self::event_trigger(&mut us_lock);
-							},
-							Err(e) => shutdown_socket!(e, Disconnect::CloseConnection),
+						{
+							let mut us_lock = us.lock().unwrap();
+							match read_res {
+								Ok(pause_read) => {
+									if pause_read {
+										us_lock.read_paused = true;
+									}
+									Self::event_trigger(&mut us_lock);
+								},
+								Err(e) => shutdown_socket!(e, Disconnect::CloseConnection),
+							}
+							us_lock.block_disconnect_socket = false;
 						}
-						us_lock.block_disconnect_socket = false;
+						peer_manager.process_events();
 					},
 					Err(e) => shutdown_socket!(e, Disconnect::PeerDisconnected),
 				},
@@ -246,7 +249,7 @@ impl Connection {
 /// not need to poll the provided future in order to make progress.
 ///
 /// See the module-level documentation for how to handle the event_notify mpsc::Sender.
-pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
+pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
 		CMH: ChannelMessageHandler + 'static,
 		RMH: RoutingMessageHandler + 'static,
 		L: Logger + 'static + ?Sized {
@@ -288,7 +291,7 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
 /// not need to poll the provided future in order to make progress.
 ///
 /// See the module-level documentation for how to handle the event_notify mpsc::Sender.
-pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
+pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
 		CMH: ChannelMessageHandler + 'static,
 		RMH: RoutingMessageHandler + 'static,
 		L: Logger + 'static + ?Sized {
@@ -360,7 +363,7 @@ pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<S
 /// make progress.
 ///
 /// See the module-level documentation for how to handle the event_notify mpsc::Sender.
-pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, addr: SocketAddr) -> Option<impl std::future::Future<Output=()>> where
+pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, addr: SocketAddr) -> Option<impl std::future::Future<Output=()>> where
 		CMH: ChannelMessageHandler + 'static,
 		RMH: RoutingMessageHandler + 'static,
 		L: Logger + 'static + ?Sized {
@@ -411,7 +414,7 @@ impl SocketDescriptor {
 		Self { conn, id }
 	}
 }
-impl peer_handler::SocketDescriptor for SocketDescriptor {
+impl handler::SocketDescriptor for SocketDescriptor {
 	fn send_data(&mut self, data: &[u8], resume_read: bool) -> usize {
 		// To send data, we take a lock on our Connection to access the WriteHalf of the TcpStream,
 		// writing to it if there's room in the kernel buffer, or otherwise create a new Waker with
@@ -503,7 +506,7 @@ impl Hash for SocketDescriptor {
 mod tests {
 	use lightning::ln::features::*;
 	use lightning::ln::msgs::*;
-	use lightning::ln::peer_handler::{MessageHandler, PeerManager};
+	use lightning::ln::peers::handler::{MessageHandler, PeerManager};
 	use lightning::util::events::*;
 	use bitcoin::secp256k1::{Secp256k1, SecretKey, PublicKey};
 

lightning/src/ln/features.rs

@@ -491,6 +491,10 @@ impl<T: sealed::StaticRemoteKey> Features<T> {
 	pub(crate) fn requires_static_remote_key(&self) -> bool {
 		<T as sealed::StaticRemoteKey>::requires_feature(&self.flags)
 	}
+	#[cfg(test)]
+	pub(crate) fn clear_requires_static_remote_key(&mut self) {
+		<T as sealed::StaticRemoteKey>::clear_bits(&mut self.flags)
+	}
 }
 
 impl<T: sealed::InitialRoutingSync> Features<T> {

lightning/src/ln/mod.rs

@@ -21,16 +21,11 @@
 pub mod channelmanager;
 pub mod channelmonitor;
 pub mod msgs;
-pub mod peer_handler;
+pub mod peers;
 pub mod chan_utils;
 pub mod features;
 pub(crate) mod onchaintx;
 
-#[cfg(feature = "fuzztarget")]
-pub mod peer_channel_encryptor;
-#[cfg(not(feature = "fuzztarget"))]
-pub(crate) mod peer_channel_encryptor;
-
 mod channel;
 mod onion_utils;
 mod wire;
@@ -50,5 +45,3 @@ mod functional_tests;
 mod chanmon_update_fail_tests;
 #[cfg(test)]
 mod reorg_tests;
-
-pub use self::peer_channel_encryptor::LN_MAX_MSG_LEN;

lightning/src/ln/msgs.rs

@@ -672,6 +672,7 @@ pub enum ErrorAction {
 }
 
 /// An Err type for failure to process messages.
+#[derive(Clone)]
 pub struct LightningError {
 	/// A human-readable message describing the error
 	pub err: String,