use std::sync::{Arc, RwLock};
-use std::time::{SystemTime, UNIX_EPOCH};
use bitcoin::secp256k1::PublicKey;
+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 tokio::sync::mpsc;
+use tokio::sync::mpsc::error::TrySendError;
-use crate::{GossipChainAccess, TestLogger};
-use crate::types::{DetectedGossipMessage, GossipMessage};
+use crate::TestLogger;
+use crate::types::{GossipMessage, GossipChainAccess};
+use crate::verifier::ChainVerifier;
pub(crate) struct GossipCounter {
pub(crate) channel_announcements: u64,
}
pub(crate) struct GossipRouter {
- pub(crate) native_router: Arc<P2PGossipSync<Arc<NetworkGraph<Arc<TestLogger>>>, GossipChainAccess, Arc<TestLogger>>>,
+ native_router: P2PGossipSync<Arc<NetworkGraph<TestLogger>>, GossipChainAccess, TestLogger>,
pub(crate) counter: RwLock<GossipCounter>,
- pub(crate) sender: mpsc::Sender<DetectedGossipMessage>,
+ sender: mpsc::Sender<GossipMessage>,
+}
+
+impl GossipRouter {
+ pub(crate) fn new(network_graph: Arc<NetworkGraph<TestLogger>>, sender: mpsc::Sender<GossipMessage>) -> Self {
+ Self {
+ native_router: P2PGossipSync::new(network_graph, Some(Arc::new(ChainVerifier::new())), TestLogger::new()),
+ counter: RwLock::new(GossipCounter::new()),
+ sender
+ }
+ }
}
impl MessageSendEventsProvider for GossipRouter {
}
fn handle_channel_announcement(&self, msg: &ChannelAnnouncement) -> Result<bool, LightningError> {
- let timestamp_seen = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
-
- 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
- })?;
+ let native_result = self.native_router.handle_channel_announcement(msg);
+ let output_value;
+ {
+ let mut counter = self.counter.write().unwrap();
+ output_value = native_result.map_err(|error| {
+ if error.err.contains("didn't match on-chain script") {
+ counter.channel_announcements_with_mismatched_scripts += 1;
+ }
+ error
+ })?;
+ counter.channel_announcements += 1;
+ }
- counter.channel_announcements += 1;
let gossip_message = GossipMessage::ChannelAnnouncement(msg.clone());
- let detected_gossip_message = DetectedGossipMessage {
- message: gossip_message,
- timestamp_seen: timestamp_seen as u32,
- };
- let sender = self.sender.clone();
- tokio::spawn(async move {
- let _ = sender.send(detected_gossip_message).await;
- });
+ 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<bool, LightningError> {
- let timestamp_seen = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
let output_value = self.native_router.handle_channel_update(msg)?;
- let mut counter = self.counter.write().unwrap();
- counter.channel_updates += 1;
+ self.counter.write().unwrap().channel_updates += 1;
let gossip_message = GossipMessage::ChannelUpdate(msg.clone());
- let detected_gossip_message = DetectedGossipMessage {
- message: gossip_message,
- timestamp_seen: timestamp_seen as u32,
- };
- let sender = self.sender.clone();
- tokio::spawn(async move {
- let _ = sender.send(detected_gossip_message).await;
- });
+
+ 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 get_next_channel_announcements(&self, starting_point: u64, batch_amount: u8) -> Vec<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> {
- self.native_router.get_next_channel_announcements(starting_point, batch_amount)
+ fn get_next_channel_announcement(&self, starting_point: u64) -> Option<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> {
+ self.native_router.get_next_channel_announcement(starting_point)
}
- fn get_next_node_announcements(&self, starting_point: Option<&PublicKey>, batch_amount: u8) -> Vec<NodeAnnouncement> {
- self.native_router.get_next_node_announcements(starting_point, batch_amount)
+ fn get_next_node_announcement(&self, starting_point: Option<&PublicKey>) -> Option<NodeAnnouncement> {
+ self.native_router.get_next_node_announcement(starting_point)
}
- fn peer_connected(&self, their_node_id: &PublicKey, init: &Init) {
+ fn peer_connected(&self, their_node_id: &PublicKey, init: &Init) -> Result<(), ()> {
self.native_router.peer_connected(their_node_id, init)
}
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()
+ }
}
projects / rapid-gossip-sync-server / blobdiff