Expose event-processing loop from peer_handler.

diff --git a/src/ln/peer_handler.rs b/src/ln/peer_handler.rs
index 3eb3ebf22dafd8eeb80b96661c6e53d7986693f6..a6992eae4ff1fcf62dab20875407e2741dcebec5 100644 (file)
--- a/src/ln/peer_handler.rs
+++ b/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.