Provide ConnectionNeeded events from PeerManager

OnionMessageHandler may signal to PeerManager that it is buffering onion
messages for non-peer nodes. Have PeerManager query it and provide
ConnectionNeeded events for those nodes. Once the user processes such an
event by connecting to the node, the buffered onion messages will be
processed as usual.

Author: jkczyz

lightning/src/ln/peer_handler.rs

@@ -19,7 +19,7 @@ use bitcoin::blockdata::constants::ChainHash;
 use bitcoin::secp256k1::{self, Secp256k1, SecretKey, PublicKey};
 
 use crate::sign::{KeysManager, NodeSigner, Recipient};
-use crate::events::{MessageSendEvent, MessageSendEventsProvider};
+use crate::events::{Event, EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider};
 use crate::ln::ChannelId;
 use crate::ln::features::{InitFeatures, NodeFeatures};
 use crate::ln::msgs;
@@ -749,6 +749,9 @@ pub struct PeerManager<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: D
 	///        it finishes as another thread tried to start processing events but returned early.
 	event_processing_state: AtomicI32,
 
+	/// Events (as opposed to MessageSendEvents) needing processing.
+	pending_events: Mutex<Vec<Event>>,
+
 	/// Used to track the last value sent in a node_announcement "timestamp" field. We ensure this
 	/// value increases strictly since we don't assume access to a time source.
 	last_node_announcement_serial: AtomicU32,
@@ -918,6 +921,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 			peers: FairRwLock::new(HashMap::new()),
 			node_id_to_descriptor: Mutex::new(HashMap::new()),
 			event_processing_state: AtomicI32::new(0),
+			pending_events: Mutex::new(Vec::new()),
 			ephemeral_key_midstate,
 			peer_counter: AtomicCounter::new(),
 			gossip_processing_backlogged: AtomicBool::new(false),
@@ -2396,6 +2400,13 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 				}
 			}
 		}
+
+		let nodes_to_connect =
+			self.message_handler.onion_message_handler.release_pending_connections();
+		let mut pending_events = self.pending_events.lock().unwrap();
+		for node_announcement in nodes_to_connect {
+			pending_events.push(Event::ConnectionNeeded { node_announcement });
+		}
 	}
 
 	#[allow(dead_code)]
@@ -2482,6 +2493,24 @@ fn is_gossip_msg(type_id: u16) -> bool {
 	}
 }
 
+impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CMH: Deref, NS: Deref>
+EventsProvider for PeerManager<Descriptor, CM, RM, OM, L, CMH, NS>
+where
+	CM::Target: ChannelMessageHandler,
+	RM::Target: RoutingMessageHandler,
+	OM::Target: OnionMessageHandler,
+	L::Target: Logger,
+	CMH::Target: CustomMessageHandler,
+	NS::Target: NodeSigner,
+{
+	fn process_pending_events<H: Deref>(&self, handler: H) where H::Target: EventHandler {
+		let pending_events = core::mem::take(&mut *self.pending_events.lock().unwrap());
+		for event in pending_events {
+			handler.handle_event(event);
+		}
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use crate::sign::{NodeSigner, Recipient};