X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning-net-tokio%2Fsrc%2Flib.rs;h=cee7c5c1b982882244cdb33a5d82d7abb0c8eadc;hb=b1550524cfe2689e3743ddfc0a78527d26de8613;hp=2582cc597f20d361906271811e8d523f5977516e;hpb=797a648b6caa105e4c6edd9dba337eebb15bc97c;p=rust-lightning

diff --git a/lightning-net-tokio/src/lib.rs b/lightning-net-tokio/src/lib.rs
index 2582cc59..cee7c5c1 100644
--- a/lightning-net-tokio/src/lib.rs
+++ b/lightning-net-tokio/src/lib.rs
@@ -81,10 +81,11 @@ use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use lightning::ln::peer_handler;
 use lightning::ln::peer_handler::SocketDescriptor as LnSocketTrait;
 use lightning::ln::peer_handler::CustomMessageHandler;
-use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler};
+use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler, NetAddress};
 use lightning::util::logger::Logger;
 
 use std::task;
+use std::net::IpAddr;
 use std::net::SocketAddr;
 use std::net::TcpStream as StdTcpStream;
 use std::sync::{Arc, Mutex};
@@ -120,11 +121,28 @@ struct Connection {
 	id: u64,
 }
 impl Connection {
+	async fn poll_event_process<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, mut event_receiver: mpsc::Receiver<()>) where
+			CMH: ChannelMessageHandler + 'static + Send + Sync,
+			RMH: RoutingMessageHandler + 'static + Send + Sync,
+			L: Logger + 'static + ?Sized + Send + Sync,
+			UMH: CustomMessageHandler + 'static + Send + Sync {
+		loop {
+			if event_receiver.recv().await.is_none() {
+				return;
+			}
+			peer_manager.process_events();
+		}
+	}
+
 	async fn schedule_read<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
-			CMH: ChannelMessageHandler + 'static,
-			RMH: RoutingMessageHandler + 'static,
-			L: Logger + 'static + ?Sized,
-			UMH: CustomMessageHandler + 'static {
+			CMH: ChannelMessageHandler + 'static + Send + Sync,
+			RMH: RoutingMessageHandler + 'static + Send + Sync,
+			L: Logger + 'static + ?Sized + Send + Sync,
+			UMH: CustomMessageHandler + 'static + Send + Sync {
+		// Create a waker to wake up poll_event_process, above
+		let (event_waker, event_receiver) = mpsc::channel(1);
+		tokio::spawn(Self::poll_event_process(Arc::clone(&peer_manager), event_receiver));
+
 		// 8KB is nice and big but also should never cause any issues with stack overflowing.
 		let mut buf = [0; 8192];
 
@@ -175,7 +193,7 @@ impl Connection {
 					Err(_) => break Disconnect::PeerDisconnected,
 				},
 			}
-			peer_manager.process_events();
+			let _ = event_waker.try_send(());
 		};
 		let writer_option = us.lock().unwrap().writer.take();
 		if let Some(mut writer) = writer_option {
@@ -222,11 +240,21 @@ pub fn setup_inbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManag
 		RMH: RoutingMessageHandler + 'static + Send + Sync,
 		L: Logger + 'static + ?Sized + Send + Sync,
 		UMH: CustomMessageHandler + 'static + Send + Sync {
+	let ip_addr = stream.peer_addr().unwrap();
 	let (reader, write_receiver, read_receiver, us) = Connection::new(stream);
 	#[cfg(debug_assertions)]
 	let last_us = Arc::clone(&us);
 
-	let handle_opt = if let Ok(_) = peer_manager.new_inbound_connection(SocketDescriptor::new(us.clone())) {
+	let handle_opt = if let Ok(_) = peer_manager.new_inbound_connection(SocketDescriptor::new(us.clone()), match ip_addr.ip() {
+		IpAddr::V4(ip) => Some(NetAddress::IPv4 {
+			addr: ip.octets(),
+			port: ip_addr.port(),
+		}),
+		IpAddr::V6(ip) => Some(NetAddress::IPv6 {
+			addr: ip.octets(),
+			port: ip_addr.port(),
+		}),
+	}) {
 		Some(tokio::spawn(Connection::schedule_read(peer_manager, us, reader, read_receiver, write_receiver)))
 	} else {
 		// Note that we will skip socket_disconnected here, in accordance with the PeerManager
@@ -263,11 +291,20 @@ pub fn setup_outbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerMana
 		RMH: RoutingMessageHandler + 'static + Send + Sync,
 		L: Logger + 'static + ?Sized + Send + Sync,
 		UMH: CustomMessageHandler + 'static + Send + Sync {
+	let ip_addr = stream.peer_addr().unwrap();
 	let (reader, mut write_receiver, read_receiver, us) = Connection::new(stream);
 	#[cfg(debug_assertions)]
 	let last_us = Arc::clone(&us);
-
-	let handle_opt = if let Ok(initial_send) = peer_manager.new_outbound_connection(their_node_id, SocketDescriptor::new(us.clone())) {
+	let handle_opt = if let Ok(initial_send) = peer_manager.new_outbound_connection(their_node_id, SocketDescriptor::new(us.clone()), match ip_addr.ip() {
+		IpAddr::V4(ip) => Some(NetAddress::IPv4 {
+			addr: ip.octets(),
+			port: ip_addr.port(),
+		}),
+		IpAddr::V6(ip) => Some(NetAddress::IPv6 {
+			addr: ip.octets(),
+			port: ip_addr.port(),
+		}),
+	}) {
 		Some(tokio::spawn(async move {
 			// We should essentially always have enough room in a TCP socket buffer to send the
 			// initial 10s of bytes. However, tokio running in single-threaded mode will always
@@ -496,7 +533,7 @@ mod tests {
 		fn handle_channel_update(&self, _msg: &ChannelUpdate) -> Result<bool, LightningError> { Ok(false) }
 		fn get_next_channel_announcements(&self, _starting_point: u64, _batch_amount: u8) -> Vec<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> { Vec::new() }
 		fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec<NodeAnnouncement> { Vec::new() }
-		fn sync_routing_table(&self, _their_node_id: &PublicKey, _init_msg: &Init) { }
+		fn peer_connected(&self, _their_node_id: &PublicKey, _init_msg: &Init) { }
 		fn handle_reply_channel_range(&self, _their_node_id: &PublicKey, _msg: ReplyChannelRange) -> Result<(), LightningError> { Ok(()) }
 		fn handle_reply_short_channel_ids_end(&self, _their_node_id: &PublicKey, _msg: ReplyShortChannelIdsEnd) -> Result<(), LightningError> { Ok(()) }
 		fn handle_query_channel_range(&self, _their_node_id: &PublicKey, _msg: QueryChannelRange) -> Result<(), LightningError> { Ok(()) }