Expose event-processing loop from peer_handler.

src/ln/peer_handler.rs

@@ -219,7 +219,6 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 	}
 
 	fn do_read_event(&self, peer_descriptor: &mut Descriptor, data: Vec<u8>) -> Result<bool, PeerHandleError> {
-		let mut upstream_events = Vec::new();
 		let pause_read = {
 			let mut peers = self.peers.lock().unwrap();
 			let (should_insert_node_id, pause_read) = match peers.peers.get_mut(peer_descriptor) {
@@ -463,11 +462,25 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 				None => {}
 			};
 
+			pause_read
+		};
+
+		self.process_events();
+
+		Ok(pause_read)
+	}
+
+	/// Checks for any events generated by our handlers and processes them. May be needed after eg
+	/// calls to ChannelManager::process_pending_htlc_forward.
+	pub fn process_events(&self) {
+		let mut upstream_events = Vec::new();
+		{
 			// TODO: There are some DoS attacks here where you can flood someone's outbound send
 			// buffer by doing things like announcing channels on another node. We should be willing to
 			// drop optional-ish messages when send buffers get full!
 
 			let mut events_generated = self.message_handler.chan_handler.get_and_clear_pending_events();
+			let mut peers = self.peers.lock().unwrap();
 			for event in events_generated.drain(..) {
 				macro_rules! get_peer_for_forwarding {
 					($node_id: expr, $handle_no_such_peer: block) => {
@@ -573,16 +586,12 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 
 				upstream_events.push(event);
 			}
-
-			pause_read
-		};
+		}
 
 		let mut pending_events = self.pending_events.lock().unwrap();
 		for event in upstream_events.drain(..) {
 			pending_events.push(event);
 		}
-
-		Ok(pause_read)
 	}
 
 	/// Indicates that the given socket descriptor's connection is now closed.