2 use std::sync::{Arc, RwLock};
4 use bitcoin::secp256k1::PublicKey;
5 use lightning::events::{MessageSendEvent, MessageSendEventsProvider};
6 use lightning::ln::features::{InitFeatures, NodeFeatures};
7 use lightning::ln::msgs::{ChannelAnnouncement, ChannelUpdate, Init, LightningError, NodeAnnouncement, QueryChannelRange, QueryShortChannelIds, ReplyChannelRange, ReplyShortChannelIdsEnd, RoutingMessageHandler};
8 use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
9 use lightning::util::logger::Logger;
10 use tokio::sync::mpsc;
11 use tokio::sync::mpsc::error::TrySendError;
13 use crate::types::{GossipMessage, GossipChainAccess, GossipPeerManager};
14 use crate::verifier::ChainVerifier;
16 pub(crate) struct GossipCounter {
17 pub(crate) node_announcements: u64,
18 pub(crate) channel_announcements: u64,
19 pub(crate) channel_updates: u64,
20 pub(crate) channel_updates_without_htlc_max_msats: u64,
21 pub(crate) channel_announcements_with_mismatched_scripts: u64
25 pub(crate) fn new() -> Self {
27 node_announcements: 0,
28 channel_announcements: 0,
30 channel_updates_without_htlc_max_msats: 0,
31 channel_announcements_with_mismatched_scripts: 0,
36 pub(crate) struct GossipRouter<L: Deref + Clone + Send + Sync + 'static> where L::Target: Logger {
37 native_router: P2PGossipSync<Arc<NetworkGraph<L>>, GossipChainAccess<L>, L>,
38 pub(crate) counter: RwLock<GossipCounter>,
39 sender: mpsc::Sender<GossipMessage>,
40 verifier: Arc<ChainVerifier<L>>,
41 outbound_gossiper: Arc<P2PGossipSync<Arc<NetworkGraph<L>>, GossipChainAccess<L>, L>>,
44 impl<L: Deref + Clone + Send + Sync> GossipRouter<L> where L::Target: Logger {
45 pub(crate) fn new(network_graph: Arc<NetworkGraph<L>>, sender: mpsc::Sender<GossipMessage>, logger: L) -> Self {
46 let outbound_gossiper = Arc::new(P2PGossipSync::new(Arc::clone(&network_graph), None, logger.clone()));
47 let verifier = Arc::new(ChainVerifier::new(Arc::clone(&network_graph), Arc::clone(&outbound_gossiper), logger.clone()));
49 native_router: P2PGossipSync::new(network_graph, Some(Arc::clone(&verifier)), logger.clone()),
51 counter: RwLock::new(GossipCounter::new()),
57 pub(crate) fn set_pm(&self, peer_handler: GossipPeerManager<L>) {
58 self.verifier.set_ph(peer_handler);
61 fn new_channel_announcement(&self, msg: ChannelAnnouncement) {
63 let mut counter = self.counter.write().unwrap();
64 counter.channel_announcements += 1;
67 let gossip_message = GossipMessage::ChannelAnnouncement(msg, None);
68 if let Err(err) = self.sender.try_send(gossip_message) {
69 let gossip_message = match err { TrySendError::Full(msg)|TrySendError::Closed(msg) => msg };
70 tokio::task::block_in_place(move || { tokio::runtime::Handle::current().block_on(async move {
71 self.sender.send(gossip_message).await.unwrap();
76 fn new_node_announcement(&self, msg: NodeAnnouncement) {
78 let mut counter = self.counter.write().unwrap();
79 counter.node_announcements += 1;
82 let gossip_message = GossipMessage::NodeAnnouncement(msg, None);
83 if let Err(err) = self.sender.try_send(gossip_message) {
84 let gossip_message = match err { TrySendError::Full(msg)|TrySendError::Closed(msg) => msg };
85 tokio::task::block_in_place(move || { tokio::runtime::Handle::current().block_on(async move {
86 self.sender.send(gossip_message).await.unwrap();
91 fn new_channel_update(&self, msg: ChannelUpdate) {
92 self.counter.write().unwrap().channel_updates += 1;
93 let gossip_message = GossipMessage::ChannelUpdate(msg, None);
95 if let Err(err) = self.sender.try_send(gossip_message) {
96 let gossip_message = match err { TrySendError::Full(msg)|TrySendError::Closed(msg) => msg };
97 tokio::task::block_in_place(move || { tokio::runtime::Handle::current().block_on(async move {
98 self.sender.send(gossip_message).await.unwrap();
104 impl<L: Deref + Clone + Send + Sync> MessageSendEventsProvider for GossipRouter<L> where L::Target: Logger {
105 fn get_and_clear_pending_msg_events(&self) -> Vec<MessageSendEvent> {
106 let gossip_evs = self.outbound_gossiper.get_and_clear_pending_msg_events();
107 for ev in gossip_evs {
109 MessageSendEvent::BroadcastChannelAnnouncement { msg, .. } => {
110 self.new_channel_announcement(msg);
112 MessageSendEvent::BroadcastNodeAnnouncement { msg } => {
113 self.new_node_announcement(msg);
115 MessageSendEvent::BroadcastChannelUpdate { msg } => {
116 self.new_channel_update(msg);
118 _ => { unreachable!() },
121 self.native_router.get_and_clear_pending_msg_events()
125 impl<L: Deref + Clone + Send + Sync> RoutingMessageHandler for GossipRouter<L> where L::Target: Logger {
126 fn handle_node_announcement(&self, msg: &NodeAnnouncement) -> Result<bool, LightningError> {
127 let res = self.native_router.handle_node_announcement(msg)?;
128 self.new_node_announcement(msg.clone());
132 fn handle_channel_announcement(&self, msg: &ChannelAnnouncement) -> Result<bool, LightningError> {
133 let res = self.native_router.handle_channel_announcement(msg)?;
134 self.new_channel_announcement(msg.clone());
138 fn handle_channel_update(&self, msg: &ChannelUpdate) -> Result<bool, LightningError> {
139 let res = self.native_router.handle_channel_update(msg)?;
140 self.new_channel_update(msg.clone());
144 fn processing_queue_high(&self) -> bool {
145 self.native_router.processing_queue_high()
148 fn get_next_channel_announcement(&self, starting_point: u64) -> Option<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> {
149 self.native_router.get_next_channel_announcement(starting_point)
152 fn get_next_node_announcement(&self, starting_point: Option<&NodeId>) -> Option<NodeAnnouncement> {
153 self.native_router.get_next_node_announcement(starting_point)
156 fn peer_connected(&self, their_node_id: &PublicKey, init: &Init, inbound: bool) -> Result<(), ()> {
157 self.native_router.peer_connected(their_node_id, init, inbound)
160 fn handle_reply_channel_range(&self, their_node_id: &PublicKey, msg: ReplyChannelRange) -> Result<(), LightningError> {
161 self.native_router.handle_reply_channel_range(their_node_id, msg)
164 fn handle_reply_short_channel_ids_end(&self, their_node_id: &PublicKey, msg: ReplyShortChannelIdsEnd) -> Result<(), LightningError> {
165 self.native_router.handle_reply_short_channel_ids_end(their_node_id, msg)
168 fn handle_query_channel_range(&self, their_node_id: &PublicKey, msg: QueryChannelRange) -> Result<(), LightningError> {
169 self.native_router.handle_query_channel_range(their_node_id, msg)
172 fn handle_query_short_channel_ids(&self, their_node_id: &PublicKey, msg: QueryShortChannelIds) -> Result<(), LightningError> {
173 self.native_router.handle_query_short_channel_ids(their_node_id, msg)
176 fn provided_init_features(&self, their_node_id: &PublicKey) -> InitFeatures {
177 self.native_router.provided_init_features(their_node_id)
180 fn provided_node_features(&self) -> NodeFeatures {
181 self.native_router.provided_node_features()