Add Notifier to OnionMessenger

[tnull]

Closes #3190

We add a Notifier to OnionMessenger that allows us to notify the background processor of new events being ready to be handled.

[codecov]

Codecov Report

Attention: Patch coverage is 80.85106% with 9 lines in your changes missing coverage. Please review.

Project coverage is 90.95%. Comparing base (9ce3dd5) to head (2dd8c2b).
Report is 80 commits behind head on main.

Files	Patch %	Lines
lightning-background-processor/src/lib.rs	71.87%	9 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3194      +/-   ##
==========================================
+ Coverage   89.78%   90.95%   +1.17%     
==========================================
  Files         121      122       +1     
  Lines      100932   109280    +8348     
  Branches   100932   109280    +8348     
==========================================
+ Hits        90619    99395    +8776     
+ Misses       7635     7191     -444     
- Partials     2678     2694      +16     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[TheBlueMatt]

Need to update the non-async BP too.

[tnull]

Need to update the non-async BP too.

Wat, we still support that? :P

(Ah, thanks, good catch!)

[tnull]

Let me know if I can squash fixups

[TheBlueMatt]

Feel free, IMO.

[tnull]

Squashed without further changes (ignore the intermittent force-pushes..):

> git diff-tree -U2 e3ed93b2d b3d45ccfd
>

[tnull]

Force-pushed with the following changes:

> git diff-tree -U2 b3d45ccfd 7b8200a47
diff --git a/lightning/src/onion_message/messenger.rs b/lightning/src/onion_message/messenger.rs
index 30ad572fd..ec1682bed 100644
--- a/lightning/src/onion_message/messenger.rs
+++ b/lightning/src/onion_message/messenger.rs
@@ -1250,8 +1250,5 @@ where
                        },
                        hash_map::Entry::Occupied(mut e) => {
-                               let notify = e.get_mut().enqueue_message(onion_message);
-                               if notify {
-                                       self.event_notifier.notify();
-                               }
+                               e.get_mut().enqueue_message(onion_message);
                                if e.get().is_connected() {
                                        Ok(SendSuccess::Buffered)

[tnull]

Force-pushed with additional changes (actually reverting prior changes):

> git diff-tree -U2 7b8200a47 2dd8c2b3d
diff --git a/lightning/src/onion_message/messenger.rs b/lightning/src/onion_message/messenger.rs
index ec1682bed..590886721 100644
--- a/lightning/src/onion_message/messenger.rs
+++ b/lightning/src/onion_message/messenger.rs
@@ -295,17 +295,11 @@ impl OnionMessageRecipient {
        }

-       // Returns whether changes were made that are pending event processing
-       fn enqueue_message(&mut self, message: OnionMessage) -> bool {
-               let mut pending_event_processing = false;
+       fn enqueue_message(&mut self, message: OnionMessage) {
                let pending_messages = match self {
                        OnionMessageRecipient::ConnectedPeer(pending_messages) => pending_messages,
-                       OnionMessageRecipient::PendingConnection(pending_messages, _, _) => {
-                               pending_event_processing = true;
-                               pending_messages
-                       }
+                       OnionMessageRecipient::PendingConnection(pending_messages, _, _) => pending_messages,
                };

                pending_messages.push_back(message);
-               pending_event_processing
        }

@@ -1241,9 +1235,7 @@ where
                                None => Err(SendError::InvalidFirstHop(first_node_id)),
                                Some(addresses) => {
-                                       let notify = e.insert(OnionMessageRecipient::pending_connection(addresses))
+                                       e.insert(OnionMessageRecipient::pending_connection(addresses))
                                                .enqueue_message(onion_message);
-                                       if notify {
-                                               self.event_notifier.notify();
-                                       }
+                                       self.event_notifier.notify();
                                        Ok(SendSuccess::BufferedAwaitingConnection(first_node_id))
                                },

[TheBlueMatt]

LGTM