X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fpersistence.rs;h=04c6b9a0c4076cab725b90b5a4ec769718d183a0;hb=74dd28d4e2c75a4dbb15b3c5045bddfeb2bd9c53;hp=8e299ee202e48dbf0d700e4c278462e132b7338b;hpb=3f77a1de51c39809b45b6065d1d533a2e7080a7f;p=rapid-gossip-sync-server

diff --git a/src/persistence.rs b/src/persistence.rs
index 8e299ee..04c6b9a 100644
--- a/src/persistence.rs
+++ b/src/persistence.rs
@@ -1,7 +1,6 @@
 use std::fs::OpenOptions;
 use std::io::{BufWriter, Write};
 use std::ops::Deref;
-use std::mem;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use lightning::log_info;
@@ -38,10 +37,10 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger {
 	}
 
 	pub(crate) async fn persist_gossip(&mut self) {
-		let mut client = crate::connect_to_db().await;
+		{ // initialize the database
+			// this client instance is only used once
+			let mut client = crate::connect_to_db().await;
 
-		{
-			// initialize the database
 			let initialization = client
 				.execute(config::db_config_table_creation_query(), &[])
 				.await;
@@ -70,21 +69,20 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger {
 				panic!("db init error: {}", initialization_error);
 			}
 
-			let initialization = client
-				.execute(config::db_announcement_table_creation_query(), &[])
-				.await;
-			if let Err(initialization_error) = initialization {
-				panic!("db init error: {}", initialization_error);
-			}
+			let table_creation_queries = [
+				config::db_announcement_table_creation_query(),
+				config::db_channel_update_table_creation_query(),
+				config::db_channel_update_table_creation_query(),
+				config::db_node_announcement_table_creation_query()
+			];
 
-			let initialization = client
-				.execute(
-					config::db_channel_update_table_creation_query(),
-					&[],
-				)
-				.await;
-			if let Err(initialization_error) = initialization {
-				panic!("db init error: {}", initialization_error);
+			for current_table_creation_query in table_creation_queries {
+				let initialization = client
+					.execute(current_table_creation_query, &[])
+					.await;
+				if let Err(initialization_error) = initialization {
+					panic!("db init error: {}", initialization_error);
+				}
 			}
 
 			let initialization = client
@@ -122,8 +120,71 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger {
 			insert_limiter.acquire().await.unwrap().forget();
 
 			let limiter_ref = Arc::clone(&insert_limiter);
+			let client = {
+				let mut connections_set = connections_cache.lock().await;
+				let client = if connections_set.is_empty() {
+					crate::connect_to_db().await
+				} else {
+					connections_set.pop().unwrap()
+				};
+				client
+			};
+
 			let connections_cache_ref = Arc::clone(&connections_cache);
 			match gossip_message {
+				GossipMessage::NodeAnnouncement(announcement, seen_override) => {
+					let public_key_hex = announcement.contents.node_id.to_string();
+
+					let mut announcement_signed = Vec::new();
+					announcement.write(&mut announcement_signed).unwrap();
+
+					let features = announcement.contents.features.encode();
+					let timestamp = announcement.contents.timestamp as i64;
+
+					let mut serialized_addresses = Vec::new();
+					announcement.contents.addresses.write(&mut serialized_addresses).unwrap();
+
+					let _task = self.tokio_runtime.spawn(async move {
+						if cfg!(test) && seen_override.is_some() {
+							tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
+								.execute("INSERT INTO node_announcements (\
+								public_key, \
+								features, \
+								socket_addresses, \
+								timestamp, \
+								announcement_signed, \
+								seen \
+							) VALUES ($1, $2, $3, $4, $5, TO_TIMESTAMP($6))", &[
+									&public_key_hex,
+									&features,
+									&serialized_addresses,
+									&timestamp,
+									&announcement_signed,
+									&(seen_override.unwrap() as f64)
+								])).await.unwrap().unwrap();
+						} else {
+							tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
+								.execute("INSERT INTO node_announcements (\
+								public_key, \
+								features, \
+								socket_addresses, \
+								timestamp, \
+								announcement_signed \
+							) VALUES ($1, $2, $3, $4, $5)", &[
+									&public_key_hex,
+									&features,
+									&serialized_addresses,
+									&timestamp,
+									&announcement_signed,
+								])).await.unwrap().unwrap();
+						}
+						let mut connections_set = connections_cache_ref.lock().await;
+						connections_set.push(client);
+						limiter_ref.add_permits(1);
+					});
+					#[cfg(test)]
+					tasks_spawned.push(_task);
+				},
 				GossipMessage::ChannelAnnouncement(announcement, seen_override) => {
 					let scid = announcement.contents.short_channel_id as i64;
 
@@ -132,16 +193,6 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger {
 					announcement.write(&mut announcement_signed).unwrap();
 
 					let _task = self.tokio_runtime.spawn(async move {
-						let client;
-						{
-							let mut connections_set = connections_cache_ref.lock().await;
-							if connections_set.is_empty() {
-								mem::drop(connections_set);
-								client = crate::connect_to_db().await;
-							} else {
-								client = connections_set.pop().unwrap();
-							}
-						}
 						if cfg!(test) && seen_override.is_some() {
 							tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
 								.execute("INSERT INTO channel_announcements (\
@@ -224,16 +275,6 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger {
 					let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64;
 
 					let _task = self.tokio_runtime.spawn(async move {
-						let client;
-						{
-							let mut connections_set = connections_cache_ref.lock().await;
-							if connections_set.is_empty() {
-								mem::drop(connections_set);
-								client = crate::connect_to_db().await;
-							} else {
-								client = connections_set.pop().unwrap();
-							}
-						}
 						tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
 							.execute(insertion_statement, &[
 								&scid,
@@ -261,7 +302,7 @@ impl<L: Deref> GossipPersister<L> where L::Target: Logger {
 		}
 		#[cfg(test)]
 		for task in tasks_spawned {
-			task.await;
+			task.await.unwrap();
 		}
 	}