X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning-net-tokio%2Fsrc%2Flib.rs;h=c4a4bedad8f14f91e3efae01d1488de7428f3ee4;hb=e64b5d9d2e6252954dce095989c1d55e7003299a;hp=38b09a2886c2b825acec3cf4f4f741626f184139;hpb=d66c70eed4c71119f9a44aa31b9f00d3677e7333;p=rust-lightning diff --git a/lightning-net-tokio/src/lib.rs b/lightning-net-tokio/src/lib.rs index 38b09a28..c4a4beda 100644 --- a/lightning-net-tokio/src/lib.rs +++ b/lightning-net-tokio/src/lib.rs @@ -32,6 +32,7 @@ //! type TxBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync; //! type FeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync; //! type Logger = dyn lightning::util::logger::Logger + Send + Sync; +//! type NodeSigner = dyn lightning::chain::keysinterface::NodeSigner + Send + Sync; //! type ChainAccess = dyn lightning::chain::Access + Send + Sync; //! type ChainFilter = dyn lightning::chain::Filter + Send + Sync; //! type DataPersister = dyn lightning::chain::chainmonitor::Persist + Send + Sync; @@ -80,6 +81,7 @@ use tokio::{io, time}; use tokio::sync::mpsc; use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use lightning::chain::keysinterface::NodeSigner; use lightning::ln::peer_handler; use lightning::ln::peer_handler::SocketDescriptor as LnSocketTrait; use lightning::ln::peer_handler::CustomMessageHandler; @@ -123,21 +125,23 @@ struct Connection { id: u64, } impl Connection { - async fn poll_event_process( + async fn poll_event_process( peer_manager: PM, mut event_receiver: mpsc::Receiver<()>, ) where - PM: Deref> + 'static + Send + Sync, + PM: Deref> + 'static + Send + Sync, CMH: Deref + 'static + Send + Sync, RMH: Deref + 'static + Send + Sync, OMH: Deref + 'static + Send + Sync, L: Deref + 'static + Send + Sync, UMH: Deref + 'static + Send + Sync, + NS: Deref + 'static + Send + Sync, CMH::Target: ChannelMessageHandler + Send + Sync, RMH::Target: RoutingMessageHandler + Send + Sync, OMH::Target: OnionMessageHandler + Send + Sync, L::Target: Logger + Send + Sync, UMH::Target: CustomMessageHandler + Send + Sync, + NS::Target: NodeSigner + Send + Sync, { loop { if event_receiver.recv().await.is_none() { @@ -147,24 +151,26 @@ impl Connection { } } - async fn schedule_read( + async fn schedule_read( peer_manager: PM, us: Arc>, mut reader: io::ReadHalf, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>, ) where - PM: Deref> + 'static + Send + Sync + Clone, + PM: Deref> + 'static + Send + Sync + Clone, CMH: Deref + 'static + Send + Sync, RMH: Deref + 'static + Send + Sync, OMH: Deref + 'static + Send + Sync, L: Deref + 'static + Send + Sync, UMH: Deref + 'static + Send + Sync, + NS: Deref + 'static + Send + Sync, CMH::Target: ChannelMessageHandler + 'static + Send + Sync, RMH::Target: RoutingMessageHandler + 'static + Send + Sync, OMH::Target: OnionMessageHandler + 'static + Send + Sync, L::Target: Logger + 'static + Send + Sync, UMH::Target: CustomMessageHandler + 'static + Send + Sync, + NS::Target: NodeSigner + 'static + Send + Sync, { // Create a waker to wake up poll_event_process, above let (event_waker, event_receiver) = mpsc::channel(1); @@ -283,21 +289,23 @@ fn get_addr_from_stream(stream: &StdTcpStream) -> Option { /// The returned future will complete when the peer is disconnected and associated handling /// futures are freed, though, because all processing futures are spawned with tokio::spawn, you do /// not need to poll the provided future in order to make progress. -pub fn setup_inbound( +pub fn setup_inbound( peer_manager: PM, stream: StdTcpStream, ) -> impl std::future::Future where - PM: Deref> + 'static + Send + Sync + Clone, + PM: Deref> + 'static + Send + Sync + Clone, CMH: Deref + 'static + Send + Sync, RMH: Deref + 'static + Send + Sync, OMH: Deref + 'static + Send + Sync, L: Deref + 'static + Send + Sync, UMH: Deref + 'static + Send + Sync, + NS: Deref + 'static + Send + Sync, CMH::Target: ChannelMessageHandler + Send + Sync, RMH::Target: RoutingMessageHandler + Send + Sync, OMH::Target: OnionMessageHandler + Send + Sync, L::Target: Logger + Send + Sync, UMH::Target: CustomMessageHandler + Send + Sync, + NS::Target: NodeSigner + Send + Sync, { let remote_addr = get_addr_from_stream(&stream); let (reader, write_receiver, read_receiver, us) = Connection::new(stream); @@ -336,22 +344,24 @@ pub fn setup_inbound( /// The returned future will complete when the peer is disconnected and associated handling /// futures are freed, though, because all processing futures are spawned with tokio::spawn, you do /// not need to poll the provided future in order to make progress. -pub fn setup_outbound( +pub fn setup_outbound( peer_manager: PM, their_node_id: PublicKey, stream: StdTcpStream, ) -> impl std::future::Future where - PM: Deref> + 'static + Send + Sync + Clone, + PM: Deref> + 'static + Send + Sync + Clone, CMH: Deref + 'static + Send + Sync, RMH: Deref + 'static + Send + Sync, OMH: Deref + 'static + Send + Sync, L: Deref + 'static + Send + Sync, UMH: Deref + 'static + Send + Sync, + NS: Deref + 'static + Send + Sync, CMH::Target: ChannelMessageHandler + Send + Sync, RMH::Target: RoutingMessageHandler + Send + Sync, OMH::Target: OnionMessageHandler + Send + Sync, L::Target: Logger + Send + Sync, UMH::Target: CustomMessageHandler + Send + Sync, + NS::Target: NodeSigner + Send + Sync, { let remote_addr = get_addr_from_stream(&stream); let (reader, mut write_receiver, read_receiver, us) = Connection::new(stream); @@ -419,22 +429,24 @@ pub fn setup_outbound( /// disconnected and associated handling futures are freed, though, because all processing in said /// futures are spawned with tokio::spawn, you do not need to poll the second future in order to /// make progress. -pub async fn connect_outbound( +pub async fn connect_outbound( peer_manager: PM, their_node_id: PublicKey, addr: SocketAddr, ) -> Option> where - PM: Deref> + 'static + Send + Sync + Clone, + PM: Deref> + 'static + Send + Sync + Clone, CMH: Deref + 'static + Send + Sync, RMH: Deref + 'static + Send + Sync, OMH: Deref + 'static + Send + Sync, L: Deref + 'static + Send + Sync, UMH: Deref + 'static + Send + Sync, + NS: Deref + 'static + Send + Sync, CMH::Target: ChannelMessageHandler + Send + Sync, RMH::Target: RoutingMessageHandler + Send + Sync, OMH::Target: OnionMessageHandler + Send + Sync, L::Target: Logger + Send + Sync, UMH::Target: CustomMessageHandler + Send + Sync, + NS::Target: NodeSigner + Send + Sync, { if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await { Some(setup_outbound(peer_manager, their_node_id, stream)) @@ -573,6 +585,7 @@ mod tests { use lightning::ln::peer_handler::{MessageHandler, PeerManager}; use lightning::ln::features::NodeFeatures; use lightning::util::events::*; + use lightning::util::test_utils::TestNodeSigner; use bitcoin::secp256k1::{Secp256k1, SecretKey, PublicKey}; use tokio::sync::mpsc; @@ -688,7 +701,7 @@ mod tests { chan_handler: Arc::clone(&a_handler), route_handler: Arc::clone(&a_handler), onion_message_handler: Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}), - }, a_key.clone(), 0, &[1; 32], Arc::new(TestLogger()), Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}))); + }, 0, &[1; 32], Arc::new(TestLogger()), Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}), Arc::new(TestNodeSigner::new(a_key)))); let (b_connected_sender, mut b_connected) = mpsc::channel(1); let (b_disconnected_sender, mut b_disconnected) = mpsc::channel(1); @@ -703,7 +716,7 @@ mod tests { chan_handler: Arc::clone(&b_handler), route_handler: Arc::clone(&b_handler), onion_message_handler: Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}), - }, b_key.clone(), 0, &[2; 32], Arc::new(TestLogger()), Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}))); + }, 0, &[2; 32], Arc::new(TestLogger()), Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}), Arc::new(TestNodeSigner::new(b_key)))); // We bind on localhost, hoping the environment is properly configured with a local // address. This may not always be the case in containers and the like, so if this test is @@ -756,7 +769,7 @@ mod tests { chan_handler: Arc::new(lightning::ln::peer_handler::ErroringMessageHandler::new()), onion_message_handler: Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}), route_handler: Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}), - }, a_key, 0, &[1; 32], Arc::new(TestLogger()), Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}))); + }, 0, &[1; 32], Arc::new(TestLogger()), Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}), Arc::new(TestNodeSigner::new(a_key)))); // Make two connections, one for an inbound and one for an outbound connection let conn_a = {