Process PeerManager events in BackgroundProcessor

PeerManager now implements EventsProvider as it generates
ConnectionNeeded events for its OnionMessageHandler. Add processing to
BackgroundProcessor so that the user's EventHandler is called.

Author: jkczyz

lightning-background-processor/src/lib.rs

@@ -272,8 +272,8 @@ fn update_scorer<'a, S: 'static + Deref<Target = SC> + Send + Sync, SC: 'a + Wri
 macro_rules! define_run_body {
 	($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
 	 $channel_manager: ident, $process_channel_manager_events: expr,
-	 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
-	 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
+	 $gossip_sync: ident, $peer_manager: ident, $process_peer_manager_events: expr, $logger: ident,
+	 $scorer: ident, $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
 	 $check_slow_await: expr)
 	=> { {
 		log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
@@ -291,6 +291,7 @@ macro_rules! define_run_body {
 		loop {
 			$process_channel_manager_events;
 			$process_chain_monitor_events;
+			$process_peer_manager_events;
 
 			// Note that the PeerManager::process_events may block on ChannelManager's locks,
 			// hence it comes last here. When the ChannelManager finishes whatever it's doing,
@@ -653,7 +654,8 @@ where
 	define_run_body!(persister,
 		chain_monitor, chain_monitor.process_pending_events_async(async_event_handler).await,
 		channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
-		gossip_sync, peer_manager, logger, scorer, should_break, {
+		gossip_sync, peer_manager, peer_manager.process_pending_events_async(async_event_handler).await,
+		logger, scorer, should_break, {
 			let fut = Selector {
 				a: channel_manager.get_event_or_persistence_needed_future(),
 				b: chain_monitor.get_update_future(),
@@ -742,7 +744,7 @@ impl BackgroundProcessor {
 		CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
 		PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
 		RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
-		APM: APeerManager + Send + Sync,
+		APM: APeerManager + EventsProvider + Send + Sync,
 		PM: 'static + Deref<Target = APM> + Send + Sync,
 		S: 'static + Deref<Target = SC> + Send + Sync,
 		SC: for <'b> WriteableScore<'b>,
@@ -784,7 +786,8 @@ impl BackgroundProcessor {
 			};
 			define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
 				channel_manager, channel_manager.process_pending_events(&event_handler),
-				gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
+				gossip_sync, peer_manager, peer_manager.process_pending_events(&event_handler),
+				logger, scorer, stop_thread.load(Ordering::Acquire),
 				{ Sleeper::from_two_futures(
 					channel_manager.get_event_or_persistence_needed_future(),
 					chain_monitor.get_update_future()