blob
}
-async fn serialize_delta<L: Deref>(network_graph: Arc<NetworkGraph<L>>, last_sync_timestamp: u32, logger: L) -> SerializedResponse where L::Target: Logger {
+async fn serialize_delta<L: Deref + Clone>(network_graph: Arc<NetworkGraph<L>>, last_sync_timestamp: u32, logger: L) -> SerializedResponse where L::Target: Logger {
let (client, connection) = lookup::connect_to_db().await;
network_graph.remove_stale_channels_and_tracking();
};
let mut delta_set = DeltaSet::new();
- lookup::fetch_channel_announcements(&mut delta_set, network_graph, &client, last_sync_timestamp).await;
+ lookup::fetch_channel_announcements(&mut delta_set, network_graph, &client, last_sync_timestamp, logger.clone()).await;
log_info!(logger, "announcement channel count: {}", delta_set.len());
- lookup::fetch_channel_updates(&mut delta_set, &client, last_sync_timestamp).await;
+ lookup::fetch_channel_updates(&mut delta_set, &client, last_sync_timestamp, logger.clone()).await;
log_info!(logger, "update-fetched channel count: {}", delta_set.len());
- lookup::filter_delta_set(&mut delta_set);
+ lookup::filter_delta_set(&mut delta_set, logger.clone());
log_info!(logger, "update-filtered channel count: {}", delta_set.len());
let serialization_details = serialization::serialize_delta_set(delta_set, last_sync_timestamp);
use tokio_postgres::tls::NoTlsStream;
use futures::StreamExt;
+use lightning::log_info;
use lightning::util::logger::Logger;
use crate::config;
/// whether they had been seen before.
/// Also include all announcements for which the first update was announced
/// after `last_sync_timestamp`
-pub(super) async fn fetch_channel_announcements<L: Deref>(delta_set: &mut DeltaSet, network_graph: Arc<NetworkGraph<L>>, client: &Client, last_sync_timestamp: u32) where L::Target: Logger {
- println!("Obtaining channel ids from network graph");
+pub(super) async fn fetch_channel_announcements<L: Deref>(delta_set: &mut DeltaSet, network_graph: Arc<NetworkGraph<L>>, client: &Client, last_sync_timestamp: u32, logger: L) where L::Target: Logger {
+ log_info!(logger, "Obtaining channel ids from network graph");
let last_sync_timestamp_object = SystemTime::UNIX_EPOCH.add(Duration::from_secs(last_sync_timestamp as u64));
let channel_ids = {
let read_only_graph = network_graph.read_only();
- println!("Retrieved read-only network graph copy");
+ log_info!(logger, "Retrieved read-only network graph copy");
let channel_iterator = read_only_graph.channels().unordered_iter();
channel_iterator
.filter(|c| c.1.announcement_message.is_some())
.collect::<Vec<_>>()
};
- println!("Obtaining corresponding database entries");
+ log_info!(logger, "Obtaining corresponding database entries");
// get all the channel announcements that are currently in the network graph
let announcement_rows = client.query_raw("SELECT announcement_signed, seen FROM channel_announcements WHERE short_channel_id = any($1) ORDER BY short_channel_id ASC", [&channel_ids]).await.unwrap();
let mut pinned_rows = Box::pin(announcement_rows);
{
// THIS STEP IS USED TO DETERMINE IF A CHANNEL SHOULD BE OMITTED FROM THE DELTA
- println!("Annotating channel announcements whose oldest channel update in a given direction occurred after the last sync");
+ log_info!(logger, "Annotating channel announcements whose oldest channel update in a given direction occurred after the last sync");
// Steps:
// — Obtain all updates, distinct by (scid, direction), ordered by seen DESC // to find the oldest update in a given direction
// — From those updates, select distinct by (scid), ordered by seen DESC (to obtain the newer one per direction)
{
// THIS STEP IS USED TO DETERMINE IF A REMINDER UPDATE SHOULD BE SENT
- println!("Annotating channel announcements whose latest channel update in a given direction occurred more than six days ago");
+ log_info!(logger, "Annotating channel announcements whose latest channel update in a given direction occurred more than six days ago");
// Steps:
// — Obtain all updates, distinct by (scid, direction), ordered by seen DESC
// — From those updates, select distinct by (scid), ordered by seen ASC (to obtain the older one per direction)
}
}
-pub(super) async fn fetch_channel_updates(delta_set: &mut DeltaSet, client: &Client, last_sync_timestamp: u32) {
+pub(super) async fn fetch_channel_updates<L: Deref>(delta_set: &mut DeltaSet, client: &Client, last_sync_timestamp: u32, logger: L) where L::Target: Logger {
let start = Instant::now();
let last_sync_timestamp_object = SystemTime::UNIX_EPOCH.add(Duration::from_secs(last_sync_timestamp as u64));
", [last_sync_timestamp_object]).await.unwrap();
let mut pinned_rows = Box::pin(reference_rows);
- println!("Fetched reference rows in {:?}", start.elapsed());
+ log_info!(logger, "Fetched reference rows in {:?}", start.elapsed());
let mut last_seen_update_ids: Vec<i32> = Vec::new();
let mut non_intermediate_ids: HashSet<i32> = HashSet::new();
reference_row_count += 1;
}
- println!("Processed {} reference rows (delta size: {}) in {:?}",
+ log_info!(logger, "Processed {} reference rows (delta size: {}) in {:?}",
reference_row_count, delta_set.len(), start.elapsed());
// get all the intermediate channel updates
WHERE seen >= $1
", [last_sync_timestamp_object]).await.unwrap();
let mut pinned_updates = Box::pin(intermediate_updates);
- println!("Fetched intermediate rows in {:?}", start.elapsed());
+ log_info!(logger, "Fetched intermediate rows in {:?}", start.elapsed());
let mut previous_scid = u64::MAX;
let mut previously_seen_directions = (false, false);
}
}
}
- println!("Processed intermediate rows ({}) (delta size: {}): {:?}", intermediate_update_count, delta_set.len(), start.elapsed());
+ log_info!(logger, "Processed intermediate rows ({}) (delta size: {}): {:?}", intermediate_update_count, delta_set.len(), start.elapsed());
}
-pub(super) fn filter_delta_set(delta_set: &mut DeltaSet) {
+pub(super) fn filter_delta_set<L: Deref>(delta_set: &mut DeltaSet, logger: L) where L::Target: Logger {
let original_length = delta_set.len();
let keys: Vec<u64> = delta_set.keys().cloned().collect();
for k in keys {
let new_length = delta_set.len();
if original_length != new_length {
- println!("length modified!");
+ log_info!(logger, "length modified!");
}
}