use std::fs::OpenOptions;
use std::io::{BufWriter, Write};
+use std::ops::Deref;
use std::sync::Arc;
use std::time::{Duration, Instant};
+use lightning::log_info;
use lightning::routing::gossip::NetworkGraph;
+use lightning::util::logger::Logger;
use lightning::util::ser::Writeable;
use tokio::sync::mpsc;
use tokio_postgres::NoTls;
-use crate::{config, TestLogger};
+use crate::config;
use crate::types::GossipMessage;
-pub(crate) struct GossipPersister {
+const POSTGRES_INSERT_TIMEOUT: Duration = Duration::from_secs(15);
+
+pub(crate) struct GossipPersister<L: Deref> where L::Target: Logger {
gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
- network_graph: Arc<NetworkGraph<TestLogger>>,
+ network_graph: Arc<NetworkGraph<L>>,
+ logger: L
}
-impl GossipPersister {
- pub fn new(network_graph: Arc<NetworkGraph<TestLogger>>) -> (Self, mpsc::Sender<GossipMessage>) {
+impl<L: Deref> GossipPersister<L> where L::Target: Logger {
+ pub fn new(network_graph: Arc<NetworkGraph<L>>, logger: L) -> (Self, mpsc::Sender<GossipMessage>) {
let (gossip_persistence_sender, gossip_persistence_receiver) =
mpsc::channel::<GossipMessage>(100);
(GossipPersister {
gossip_persistence_receiver,
- network_graph
+ network_graph,
+ logger
}, gossip_persistence_sender)
}
i += 1; // count the persisted gossip messages
if latest_persistence_log.elapsed().as_secs() >= 60 {
- println!("Persisting gossip message #{}", i);
+ log_info!(self.logger, "Persisting gossip message #{}", i);
latest_persistence_log = Instant::now();
}
let mut announcement_signed = Vec::new();
announcement.write(&mut announcement_signed).unwrap();
- let result = client
+ tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
.execute("INSERT INTO channel_announcements (\
short_channel_id, \
announcement_signed \
) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[
&scid,
&announcement_signed
- ]).await;
- if result.is_err() {
- panic!("error: {}", result.err().unwrap());
- }
+ ])).await.unwrap().unwrap();
}
GossipMessage::ChannelUpdate(update) => {
let scid = update.contents.short_channel_id as i64;
let mut update_signed = Vec::new();
update.write(&mut update_signed).unwrap();
- let result = client
+ tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
.execute("INSERT INTO channel_updates (\
short_channel_id, \
timestamp, \
&fee_proportional_millionths,
&htlc_maximum_msat,
&update_signed
- ]).await;
- if result.is_err() {
- panic!("error: {}", result.err().unwrap());
- }
+ ])).await.unwrap().unwrap();
}
}
}
}
fn persist_network_graph(&self) {
- println!("Caching network graph…");
+ log_info!(self.logger, "Caching network graph…");
let cache_path = config::network_graph_cache_path();
let file = OpenOptions::new()
.create(true)
let mut writer = BufWriter::new(file);
self.network_graph.write(&mut writer).unwrap();
writer.flush().unwrap();
- println!("Cached network graph!");
+ log_info!(self.logger, "Cached network graph!");
}
}