message_router: MR,
offers_handler: OMH,
custom_handler: CMH,
+ intercept_messages_for_offline_peers: bool,
+ pending_events: Mutex<Vec<Event>>,
}
/// [`OnionMessage`]s buffered to be sent.
Receive(ParsedOnionMessageContents<T>, Option<[u8; 32]>, Option<BlindedPath>)
}
+
+/// Creates an [`OnionMessage`] with the given `contents` for sending to the destination of
+/// `path`, first calling [`Destination::resolve`] on `path.destination` with the given
+/// [`ReadOnlyNetworkGraph`].
+///
+/// Returns the node id of the peer to send the message to, the message itself, and any addresses
+/// needed to connect to the first node.
+pub fn create_onion_message_resolving_destination<
+ ES: Deref, NS: Deref, NL: Deref, T: OnionMessageContents
+>(
+ entropy_source: &ES, node_signer: &NS, node_id_lookup: &NL,
+ network_graph: &ReadOnlyNetworkGraph, secp_ctx: &Secp256k1<secp256k1::All>,
+ mut path: OnionMessagePath, contents: T, reply_path: Option<BlindedPath>,
+) -> Result<(PublicKey, OnionMessage, Option<Vec<SocketAddress>>), SendError>
+where
+ ES::Target: EntropySource,
+ NS::Target: NodeSigner,
+ NL::Target: NodeIdLookUp,
+{
+ path.destination.resolve(network_graph);
+ create_onion_message(
+ entropy_source, node_signer, node_id_lookup, secp_ctx, path, contents, reply_path,
+ )
+}
+
/// Creates an [`OnionMessage`] with the given `contents` for sending to the destination of
/// `path`.
///
/// Returns the node id of the peer to send the message to, the message itself, and any addresses
-/// need to connect to the first node.
+/// needed to connect to the first node.
+///
+/// Returns [`SendError::UnresolvedIntroductionNode`] if:
+/// - `destination` contains a blinded path with an [`IntroductionNode::DirectedShortChannelId`],
+/// - unless it can be resolved by [`NodeIdLookUp::next_node_id`].
+/// Use [`create_onion_message_resolving_destination`] instead to resolve the introduction node
+/// first with a [`ReadOnlyNetworkGraph`].
pub fn create_onion_message<ES: Deref, NS: Deref, NL: Deref, T: OnionMessageContents>(
entropy_source: &ES, node_signer: &NS, node_id_lookup: &NL,
secp_ctx: &Secp256k1<secp256k1::All>, path: OnionMessagePath, contents: T,
pub fn new(
entropy_source: ES, node_signer: NS, logger: L, node_id_lookup: NL, message_router: MR,
offers_handler: OMH, custom_handler: CMH
+ ) -> Self {
+ Self::new_inner(
+ entropy_source, node_signer, logger, node_id_lookup, message_router,
+ offers_handler, custom_handler, false
+ )
+ }
+
+ ///
+ pub fn new_with_offline_peer_interception(
+ entropy_source: ES, node_signer: NS, logger: L, node_id_lookup: NL,
+ message_router: MR, offers_handler: OMH, custom_handler: CMH
+ ) -> Self {
+ Self::new_inner(
+ entropy_source, node_signer, logger, node_id_lookup, message_router,
+ offers_handler, custom_handler, true
+ )
+ }
+
+ fn new_inner(
+ entropy_source: ES, node_signer: NS, logger: L, node_id_lookup: NL,
+ message_router: MR, offers_handler: OMH, custom_handler: CMH,
+ intercept_messages_for_offline_peers: bool
) -> Self {
let mut secp_ctx = Secp256k1::new();
secp_ctx.seeded_randomize(&entropy_source.get_secure_random_bytes());
message_router,
offers_handler,
custom_handler,
+ intercept_messages_for_offline_peers,
+ pending_events: Mutex::new(Vec::new()),
}
}
}
}
+ /// Forwards an [`OnionMessage`] to `peer_node_id`. Useful if we initialized
+ /// the [`OnionMessenger`] with [`Self::new_with_offline_peer_interception`]
+ /// and want to forward a previously intercepted onion message to a peer that
+ /// has just come online.
+ pub fn forward_onion_message(
+ &self, message: OnionMessage, peer_node_id: &PublicKey
+ ) -> Result<(), SendError> {
+ let mut message_recipients = self.message_recipients.lock().unwrap();
+ if outbound_buffer_full(&peer_node_id, &message_recipients) {
+ return Err(SendError::BufferFull);
+ }
+
+ match message_recipients.entry(*peer_node_id) {
+ hash_map::Entry::Occupied(mut e) if e.get().is_connected() => {
+ e.get_mut().enqueue_message(message);
+ Ok(())
+ },
+ _ => Err(SendError::InvalidFirstHop(*peer_node_id))
+ }
+ }
+
#[cfg(any(test, feature = "_test_utils"))]
pub fn send_onion_message_using_path<T: OnionMessageContents>(
&self, path: OnionMessagePath, contents: T, reply_path: Option<BlindedPath>
}
msgs
}
+
+ fn enqueue_event(&self, event: Event) {
+ const MAX_EVENTS_BUFFER_SIZE: usize = (1 << 10) * 256;
+ let mut pending_events = self.pending_events.lock().unwrap();
+ let total_buffered_bytes: usize = pending_events
+ .iter()
+ .map(|ev| ev.serialized_length())
+ .sum();
+ if total_buffered_bytes >= MAX_EVENTS_BUFFER_SIZE {
+ log_trace!(self.logger, "Dropping event {:?}: buffer full", event);
+ return
+ }
+ pending_events.push(event);
+ }
}
fn outbound_buffer_full(peer_node_id: &PublicKey, buffer: &HashMap<PublicKey, OnionMessageRecipient>) -> bool {
}
}
}
+ let mut events = Vec::new();
+ core::mem::swap(&mut *self.pending_events.lock().unwrap(), &mut events);
+ for ev in events {
+ handler.handle_event(ev);
+ }
}
}
e.get_mut().enqueue_message(onion_message);
log_trace!(logger, "Forwarding an onion message to peer {}", next_node_id);
},
+ _ if self.intercept_messages_for_offline_peers => {
+ self.enqueue_event(
+ Event::OnionMessageIntercepted {
+ peer_node_id: next_node_id, message: onion_message
+ }
+ );
+ },
_ => {
log_trace!(
logger,
.entry(*their_node_id)
.or_insert_with(|| OnionMessageRecipient::ConnectedPeer(VecDeque::new()))
.mark_connected();
+ if self.intercept_messages_for_offline_peers {
+ self.enqueue_event(
+ Event::OnionMessagePeerConnected { peer_node_id: *their_node_id }
+ );
+ }
} else {
self.message_recipients.lock().unwrap().remove(their_node_id);
}