X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fdownloader.rs;h=ac11ec7091739f62290274c9637af2910b3484e4;hb=8be01c021ee8a748454ac65cc6a87337e41a2ead;hp=72e55c5673970c6481be403e1c001d8564b3c374;hpb=ebbd62f41d8150b31167c1f5a60efc080f2b2961;p=rapid-gossip-sync-server diff --git a/src/downloader.rs b/src/downloader.rs index 72e55c5..ac11ec7 100644 --- a/src/downloader.rs +++ b/src/downloader.rs @@ -1,14 +1,15 @@ use std::sync::{Arc, RwLock}; use bitcoin::secp256k1::PublicKey; +use lightning::events::{MessageSendEvent, MessageSendEventsProvider}; +use lightning::ln::features::{InitFeatures, NodeFeatures}; use lightning::ln::msgs::{ChannelAnnouncement, ChannelUpdate, Init, LightningError, NodeAnnouncement, QueryChannelRange, QueryShortChannelIds, ReplyChannelRange, ReplyShortChannelIdsEnd, RoutingMessageHandler}; -use lightning::routing::gossip::{NetworkGraph, P2PGossipSync}; -use lightning::util::events::{MessageSendEvent, MessageSendEventsProvider}; +use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync}; use tokio::sync::mpsc; use tokio::sync::mpsc::error::TrySendError; use crate::TestLogger; -use crate::types::{GossipMessage, GossipChainAccess}; +use crate::types::{GossipMessage, GossipChainAccess, GossipPeerManager}; use crate::verifier::ChainVerifier; pub(crate) struct GossipCounter { @@ -30,63 +31,48 @@ impl GossipCounter { } pub(crate) struct GossipRouter { - native_router: P2PGossipSync>>, GossipChainAccess, Arc>, + native_router: P2PGossipSync>, GossipChainAccess, TestLogger>, pub(crate) counter: RwLock, sender: mpsc::Sender, + verifier: Arc, + outbound_gossiper: Arc>, GossipChainAccess, TestLogger>>, } impl GossipRouter { - pub(crate) fn new(network_graph: Arc>>, sender: mpsc::Sender) -> Self { + pub(crate) fn new(network_graph: Arc>, sender: mpsc::Sender) -> Self { + let outbound_gossiper = Arc::new(P2PGossipSync::new(Arc::clone(&network_graph), None, TestLogger::new())); + let verifier = Arc::new(ChainVerifier::new(Arc::clone(&network_graph), Arc::clone(&outbound_gossiper))); Self { - native_router: P2PGossipSync::new(network_graph, Some(Arc::new(ChainVerifier::new())), - Arc::new(TestLogger::new())), + native_router: P2PGossipSync::new(network_graph, Some(Arc::clone(&verifier)), TestLogger::new()), + outbound_gossiper, counter: RwLock::new(GossipCounter::new()), - sender + sender, + verifier, } } -} -impl MessageSendEventsProvider for GossipRouter { - fn get_and_clear_pending_msg_events(&self) -> Vec { - self.native_router.get_and_clear_pending_msg_events() + pub(crate) fn set_pm(&self, peer_handler: GossipPeerManager) { + self.verifier.set_ph(peer_handler); } -} -impl RoutingMessageHandler for GossipRouter { - fn handle_node_announcement(&self, msg: &NodeAnnouncement) -> Result { - self.native_router.handle_node_announcement(msg) - } - - fn handle_channel_announcement(&self, msg: &ChannelAnnouncement) -> Result { - let mut counter = self.counter.write().unwrap(); - - let output_value = self.native_router.handle_channel_announcement(msg).map_err(|error| { - let error_string = format!("{:?}", error); - if error_string.contains("announced on an unknown chain"){ - return error; - } - counter.channel_announcements_with_mismatched_scripts += 1; - error - })?; + fn new_channel_announcement(&self, msg: ChannelAnnouncement) { + { + let mut counter = self.counter.write().unwrap(); + counter.channel_announcements += 1; + } - counter.channel_announcements += 1; - let gossip_message = GossipMessage::ChannelAnnouncement(msg.clone()); + let gossip_message = GossipMessage::ChannelAnnouncement(msg); if let Err(err) = self.sender.try_send(gossip_message) { let gossip_message = match err { TrySendError::Full(msg)|TrySendError::Closed(msg) => msg }; tokio::task::block_in_place(move || { tokio::runtime::Handle::current().block_on(async move { self.sender.send(gossip_message).await.unwrap(); })}); } - - Ok(output_value) } - fn handle_channel_update(&self, msg: &ChannelUpdate) -> Result { - let output_value = self.native_router.handle_channel_update(msg)?; - - let mut counter = self.counter.write().unwrap(); - counter.channel_updates += 1; - let gossip_message = GossipMessage::ChannelUpdate(msg.clone()); + fn new_channel_update(&self, msg: ChannelUpdate) { + self.counter.write().unwrap().channel_updates += 1; + let gossip_message = GossipMessage::ChannelUpdate(msg); if let Err(err) = self.sender.try_send(gossip_message) { let gossip_message = match err { TrySendError::Full(msg)|TrySendError::Closed(msg) => msg }; @@ -94,20 +80,59 @@ impl RoutingMessageHandler for GossipRouter { self.sender.send(gossip_message).await.unwrap(); })}); } + } +} + +impl MessageSendEventsProvider for GossipRouter { + fn get_and_clear_pending_msg_events(&self) -> Vec { + let gossip_evs = self.outbound_gossiper.get_and_clear_pending_msg_events(); + for ev in gossip_evs { + match ev { + MessageSendEvent::BroadcastChannelAnnouncement { msg, .. } => { + self.new_channel_announcement(msg); + }, + MessageSendEvent::BroadcastNodeAnnouncement { .. } => {}, + MessageSendEvent::BroadcastChannelUpdate { msg } => { + self.new_channel_update(msg); + }, + _ => { unreachable!() }, + } + } + self.native_router.get_and_clear_pending_msg_events() + } +} - Ok(output_value) +impl RoutingMessageHandler for GossipRouter { + fn handle_node_announcement(&self, msg: &NodeAnnouncement) -> Result { + self.native_router.handle_node_announcement(msg) } - fn get_next_channel_announcements(&self, starting_point: u64, batch_amount: u8) -> Vec<(ChannelAnnouncement, Option, Option)> { - self.native_router.get_next_channel_announcements(starting_point, batch_amount) + fn handle_channel_announcement(&self, msg: &ChannelAnnouncement) -> Result { + let res = self.native_router.handle_channel_announcement(msg)?; + self.new_channel_announcement(msg.clone()); + Ok(res) } - fn get_next_node_announcements(&self, starting_point: Option<&PublicKey>, batch_amount: u8) -> Vec { - self.native_router.get_next_node_announcements(starting_point, batch_amount) + fn handle_channel_update(&self, msg: &ChannelUpdate) -> Result { + let res = self.native_router.handle_channel_update(msg)?; + self.new_channel_update(msg.clone()); + Ok(res) } - fn peer_connected(&self, their_node_id: &PublicKey, init: &Init) { - self.native_router.peer_connected(their_node_id, init) + fn processing_queue_high(&self) -> bool { + self.native_router.processing_queue_high() + } + + fn get_next_channel_announcement(&self, starting_point: u64) -> Option<(ChannelAnnouncement, Option, Option)> { + self.native_router.get_next_channel_announcement(starting_point) + } + + fn get_next_node_announcement(&self, starting_point: Option<&NodeId>) -> Option { + self.native_router.get_next_node_announcement(starting_point) + } + + fn peer_connected(&self, their_node_id: &PublicKey, init: &Init, inbound: bool) -> Result<(), ()> { + self.native_router.peer_connected(their_node_id, init, inbound) } fn handle_reply_channel_range(&self, their_node_id: &PublicKey, msg: ReplyChannelRange) -> Result<(), LightningError> { @@ -125,4 +150,12 @@ impl RoutingMessageHandler for GossipRouter { fn handle_query_short_channel_ids(&self, their_node_id: &PublicKey, msg: QueryShortChannelIds) -> Result<(), LightningError> { self.native_router.handle_query_short_channel_ids(their_node_id, msg) } + + fn provided_init_features(&self, their_node_id: &PublicKey) -> InitFeatures { + self.native_router.provided_init_features(their_node_id) + } + + fn provided_node_features(&self) -> NodeFeatures { + self.native_router.provided_node_features() + } }