#![deny(broken_intra_doc_links)]
#![deny(private_intra_doc_links)]
#![deny(missing_docs)]
#![cfg_attr(docsrs, feature(doc_auto_cfg))]
use bitcoin::secp256k1::PublicKey;
use tokio::net::TcpStream;
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;
use lightning::ln::msgs::{ChannelMessageHandler, NetAddress, OnionMessageHandler, RoutingMessageHandler};
use lightning::util::logger::Logger;
use std::ops::Deref;
use std::task;
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use std::hash::Hash;
static ID_COUNTER: AtomicU64 = AtomicU64::new(0);
struct Connection {
writer: Option<io::WriteHalf<TcpStream>>,
write_avail: mpsc::Sender<()>,
read_waker: mpsc::Sender<()>,
read_paused: bool,
rl_requested_disconnect: bool,
id: u64,
}
impl Connection {
async fn poll_event_process<PM, CMH, RMH, OMH, L, UMH, NS>(
peer_manager: PM,
mut event_receiver: mpsc::Receiver<()>,
) where
PM: Deref<Target = peer_handler::PeerManager<SocketDescriptor, CMH, RMH, OMH, L, UMH, NS>> + '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() {
return;
}
peer_manager.process_events();
}
}
async fn schedule_read<PM, CMH, RMH, OMH, L, UMH, NS>(
peer_manager: PM,
us: Arc<Mutex<Self>>,
mut reader: io::ReadHalf<TcpStream>,
mut read_wake_receiver: mpsc::Receiver<()>,
mut write_avail_receiver: mpsc::Receiver<()>,
) where
PM: Deref<Target = peer_handler::PeerManager<SocketDescriptor, CMH, RMH, OMH, L, UMH, NS>> + '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,
{
let (event_waker, event_receiver) = mpsc::channel(1);
tokio::spawn(Self::poll_event_process(peer_manager.clone(), event_receiver));
let mut buf = [0; 4096];
let mut our_descriptor = SocketDescriptor::new(us.clone());
enum Disconnect {
CloseConnection,
PeerDisconnected
}
let disconnect_type = loop {
let read_paused = {
let us_lock = us.lock().unwrap();
if us_lock.rl_requested_disconnect {
break Disconnect::CloseConnection;
}
us_lock.read_paused
};
tokio::select! {
v = write_avail_receiver.recv() => {
assert!(v.is_some()); if peer_manager.write_buffer_space_avail(&mut our_descriptor).is_err() {
break Disconnect::CloseConnection;
}
},
_ = read_wake_receiver.recv() => {},
read = reader.read(&mut buf), if !read_paused => match read {
Ok(0) => break Disconnect::PeerDisconnected,
Ok(len) => {
let read_res = peer_manager.read_event(&mut our_descriptor, &buf[0..len]);
let mut us_lock = us.lock().unwrap();
match read_res {
Ok(pause_read) => {
if pause_read {
us_lock.read_paused = true;
}
},
Err(_) => break Disconnect::CloseConnection,
}
},
Err(_) => break Disconnect::PeerDisconnected,
},
}
let _ = event_waker.try_send(());
let _ = tokio::task::yield_now().await;
};
let writer_option = us.lock().unwrap().writer.take();
if let Some(mut writer) = writer_option {
let _ = writer.shutdown().await;
}
if let Disconnect::PeerDisconnected = disconnect_type {
peer_manager.socket_disconnected(&our_descriptor);
peer_manager.process_events();
}
}
fn new(stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
let (write_avail, write_receiver) = mpsc::channel(1);
let (read_waker, read_receiver) = mpsc::channel(1);
stream.set_nonblocking(true).unwrap();
let (reader, writer) = io::split(TcpStream::from_std(stream).unwrap());
(reader, write_receiver, read_receiver,
Arc::new(Mutex::new(Self {
writer: Some(writer), write_avail, read_waker, read_paused: false,
rl_requested_disconnect: false,
id: ID_COUNTER.fetch_add(1, Ordering::AcqRel)
})))
}
}
fn get_addr_from_stream(stream: &StdTcpStream) -> Option<NetAddress> {
match stream.peer_addr() {
Ok(SocketAddr::V4(sockaddr)) => Some(NetAddress::IPv4 {
addr: sockaddr.ip().octets(),
port: sockaddr.port(),
}),
Ok(SocketAddr::V6(sockaddr)) => Some(NetAddress::IPv6 {
addr: sockaddr.ip().octets(),
port: sockaddr.port(),
}),
Err(_) => None,
}
}
pub fn setup_inbound<PM, CMH, RMH, OMH, L, UMH, NS>(
peer_manager: PM,
stream: StdTcpStream,
) -> impl std::future::Future<Output=()> where
PM: Deref<Target = peer_handler::PeerManager<SocketDescriptor, CMH, RMH, OMH, L, UMH, NS>> + '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);
#[cfg(test)]
let last_us = Arc::clone(&us);
let handle_opt = if peer_manager.new_inbound_connection(SocketDescriptor::new(us.clone()), remote_addr).is_ok() {
Some(tokio::spawn(Connection::schedule_read(peer_manager, us, reader, read_receiver, write_receiver)))
} else {
None
};
async move {
if let Some(handle) = handle_opt {
if let Err(e) = handle.await {
assert!(e.is_cancelled());
} else {
#[cfg(test)]
debug_assert!(Arc::try_unwrap(last_us).is_ok());
}
}
}
}
pub fn setup_outbound<PM, CMH, RMH, OMH, L, UMH, NS>(
peer_manager: PM,
their_node_id: PublicKey,
stream: StdTcpStream,
) -> impl std::future::Future<Output=()> where
PM: Deref<Target = peer_handler::PeerManager<SocketDescriptor, CMH, RMH, OMH, L, UMH, NS>> + '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);
#[cfg(test)]
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()), remote_addr) {
Some(tokio::spawn(async move {
if let Ok(Ok(())) = tokio::time::timeout(Duration::from_millis(100), async {
loop {
match SocketDescriptor::new(us.clone()).send_data(&initial_send, true) {
v if v == initial_send.len() => break Ok(()),
0 => {
write_receiver.recv().await;
},
_ => {
eprintln!("Failed to write first full message to socket!");
peer_manager.socket_disconnected(&SocketDescriptor::new(Arc::clone(&us)));
break Err(());
}
}
}
}).await {
Connection::schedule_read(peer_manager, us, reader, read_receiver, write_receiver).await;
}
}))
} else {
None
};
async move {
if let Some(handle) = handle_opt {
if let Err(e) = handle.await {
assert!(e.is_cancelled());
} else {
#[cfg(test)]
debug_assert!(Arc::try_unwrap(last_us).is_ok());
}
}
}
}
pub async fn connect_outbound<PM, CMH, RMH, OMH, L, UMH, NS>(
peer_manager: PM,
their_node_id: PublicKey,
addr: SocketAddr,
) -> Option<impl std::future::Future<Output=()>> where
PM: Deref<Target = peer_handler::PeerManager<SocketDescriptor, CMH, RMH, OMH, L, UMH, NS>> + '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))
} else { None }
}
const SOCK_WAKER_VTABLE: task::RawWakerVTable =
task::RawWakerVTable::new(clone_socket_waker, wake_socket_waker, wake_socket_waker_by_ref, drop_socket_waker);
fn clone_socket_waker(orig_ptr: *const ()) -> task::RawWaker {
write_avail_to_waker(orig_ptr as *const mpsc::Sender<()>)
}
fn wake_socket_waker(orig_ptr: *const ()) {
let sender = unsafe { &mut *(orig_ptr as *mut mpsc::Sender<()>) };
let _ = sender.try_send(());
drop_socket_waker(orig_ptr);
}
fn wake_socket_waker_by_ref(orig_ptr: *const ()) {
let sender_ptr = orig_ptr as *const mpsc::Sender<()>;
let sender = unsafe { (*sender_ptr).clone() };
let _ = sender.try_send(());
}
fn drop_socket_waker(orig_ptr: *const ()) {
let _orig_box = unsafe { Box::from_raw(orig_ptr as *mut mpsc::Sender<()>) };
}
fn write_avail_to_waker(sender: *const mpsc::Sender<()>) -> task::RawWaker {
let new_box = Box::leak(Box::new(unsafe { (*sender).clone() }));
let new_ptr = new_box as *const mpsc::Sender<()>;
task::RawWaker::new(new_ptr as *const (), &SOCK_WAKER_VTABLE)
}
pub struct SocketDescriptor {
conn: Arc<Mutex<Connection>>,
id: u64,
}
impl SocketDescriptor {
fn new(conn: Arc<Mutex<Connection>>) -> Self {
let id = conn.lock().unwrap().id;
Self { conn, id }
}
}
impl peer_handler::SocketDescriptor for SocketDescriptor {
fn send_data(&mut self, data: &[u8], resume_read: bool) -> usize {
let mut us = self.conn.lock().unwrap();
if us.writer.is_none() {
return 0;
}
if resume_read && us.read_paused {
us.read_paused = false;
let _ = us.read_waker.try_send(());
}
if data.is_empty() { return 0; }
let waker = unsafe { task::Waker::from_raw(write_avail_to_waker(&us.write_avail)) };
let mut ctx = task::Context::from_waker(&waker);
let mut written_len = 0;
loop {
match std::pin::Pin::new(us.writer.as_mut().unwrap()).poll_write(&mut ctx, &data[written_len..]) {
task::Poll::Ready(Ok(res)) => {
assert_ne!(res, 0);
written_len += res;
if written_len == data.len() { return written_len; }
},
task::Poll::Ready(Err(e)) => {
assert_ne!(e.kind(), io::ErrorKind::WouldBlock);
return written_len;
},
task::Poll::Pending => {
us.read_paused = true;
let _ = us.read_waker.try_send(());
return written_len;
},
}
}
}
fn disconnect_socket(&mut self) {
let mut us = self.conn.lock().unwrap();
us.rl_requested_disconnect = true;
let _ = us.write_avail.try_send(());
}
}
impl Clone for SocketDescriptor {
fn clone(&self) -> Self {
Self {
conn: Arc::clone(&self.conn),
id: self.id,
}
}
}
impl Eq for SocketDescriptor {}
impl PartialEq for SocketDescriptor {
fn eq(&self, o: &Self) -> bool {
self.id == o.id
}
}
impl Hash for SocketDescriptor {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
self.id.hash(state);
}
}
#[cfg(test)]
mod tests {
use lightning::ln::features::*;
use lightning::ln::msgs::*;
use lightning::ln::peer_handler::{MessageHandler, PeerManager};
use lightning::ln::features::NodeFeatures;
use lightning::routing::gossip::NodeId;
use lightning::events::*;
use lightning::util::test_utils::TestNodeSigner;
use bitcoin::secp256k1::{Secp256k1, SecretKey, PublicKey};
use tokio::sync::mpsc;
use std::mem;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
pub struct TestLogger();
impl lightning::util::logger::Logger for TestLogger {
fn log(&self, record: &lightning::util::logger::Record) {
println!("{:<5} [{} : {}, {}] {}", record.level.to_string(), record.module_path, record.file, record.line, record.args);
}
}
struct MsgHandler{
expected_pubkey: PublicKey,
pubkey_connected: mpsc::Sender<()>,
pubkey_disconnected: mpsc::Sender<()>,
disconnected_flag: AtomicBool,
msg_events: Mutex<Vec<MessageSendEvent>>,
}
impl RoutingMessageHandler for MsgHandler {
fn handle_node_announcement(&self, _msg: &NodeAnnouncement) -> Result<bool, LightningError> { Ok(false) }
fn handle_channel_announcement(&self, _msg: &ChannelAnnouncement) -> Result<bool, LightningError> { Ok(false) }
fn handle_channel_update(&self, _msg: &ChannelUpdate) -> Result<bool, LightningError> { Ok(false) }
fn get_next_channel_announcement(&self, _starting_point: u64) -> Option<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> { None }
fn get_next_node_announcement(&self, _starting_point: Option<&NodeId>) -> Option<NodeAnnouncement> { None }
fn peer_connected(&self, _their_node_id: &PublicKey, _init_msg: &Init, _inbound: bool) -> Result<(), ()> { Ok(()) }
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(()) }
fn handle_query_short_channel_ids(&self, _their_node_id: &PublicKey, _msg: QueryShortChannelIds) -> Result<(), LightningError> { Ok(()) }
fn provided_node_features(&self) -> NodeFeatures { NodeFeatures::empty() }
fn provided_init_features(&self, _their_node_id: &PublicKey) -> InitFeatures { InitFeatures::empty() }
fn processing_queue_high(&self) -> bool { false }
}
impl ChannelMessageHandler for MsgHandler {
fn handle_open_channel(&self, _their_node_id: &PublicKey, _msg: &OpenChannel) {}
fn handle_accept_channel(&self, _their_node_id: &PublicKey, _msg: &AcceptChannel) {}
fn handle_funding_created(&self, _their_node_id: &PublicKey, _msg: &FundingCreated) {}
fn handle_funding_signed(&self, _their_node_id: &PublicKey, _msg: &FundingSigned) {}
fn handle_channel_ready(&self, _their_node_id: &PublicKey, _msg: &ChannelReady) {}
fn handle_shutdown(&self, _their_node_id: &PublicKey, _msg: &Shutdown) {}
fn handle_closing_signed(&self, _their_node_id: &PublicKey, _msg: &ClosingSigned) {}
fn handle_update_add_htlc(&self, _their_node_id: &PublicKey, _msg: &UpdateAddHTLC) {}
fn handle_update_fulfill_htlc(&self, _their_node_id: &PublicKey, _msg: &UpdateFulfillHTLC) {}
fn handle_update_fail_htlc(&self, _their_node_id: &PublicKey, _msg: &UpdateFailHTLC) {}
fn handle_update_fail_malformed_htlc(&self, _their_node_id: &PublicKey, _msg: &UpdateFailMalformedHTLC) {}
fn handle_commitment_signed(&self, _their_node_id: &PublicKey, _msg: &CommitmentSigned) {}
fn handle_revoke_and_ack(&self, _their_node_id: &PublicKey, _msg: &RevokeAndACK) {}
fn handle_update_fee(&self, _their_node_id: &PublicKey, _msg: &UpdateFee) {}
fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {}
fn handle_channel_update(&self, _their_node_id: &PublicKey, _msg: &ChannelUpdate) {}
fn peer_disconnected(&self, their_node_id: &PublicKey) {
if *their_node_id == self.expected_pubkey {
self.disconnected_flag.store(true, Ordering::SeqCst);
self.pubkey_disconnected.clone().try_send(()).unwrap();
}
}
fn peer_connected(&self, their_node_id: &PublicKey, _init_msg: &Init, _inbound: bool) -> Result<(), ()> {
if *their_node_id == self.expected_pubkey {
self.pubkey_connected.clone().try_send(()).unwrap();
}
Ok(())
}
fn handle_channel_reestablish(&self, _their_node_id: &PublicKey, _msg: &ChannelReestablish) {}
fn handle_error(&self, _their_node_id: &PublicKey, _msg: &ErrorMessage) {}
fn provided_node_features(&self) -> NodeFeatures { NodeFeatures::empty() }
fn provided_init_features(&self, _their_node_id: &PublicKey) -> InitFeatures { InitFeatures::empty() }
}
impl MessageSendEventsProvider for MsgHandler {
fn get_and_clear_pending_msg_events(&self) -> Vec<MessageSendEvent> {
let mut ret = Vec::new();
mem::swap(&mut *self.msg_events.lock().unwrap(), &mut ret);
ret
}
}
fn make_tcp_connection() -> (std::net::TcpStream, std::net::TcpStream) {
if let Ok(listener) = std::net::TcpListener::bind("127.0.0.1:9735") {
(std::net::TcpStream::connect("127.0.0.1:9735").unwrap(), listener.accept().unwrap().0)
} else if let Ok(listener) = std::net::TcpListener::bind("127.0.0.1:19735") {
(std::net::TcpStream::connect("127.0.0.1:19735").unwrap(), listener.accept().unwrap().0)
} else if let Ok(listener) = std::net::TcpListener::bind("127.0.0.1:9997") {
(std::net::TcpStream::connect("127.0.0.1:9997").unwrap(), listener.accept().unwrap().0)
} else if let Ok(listener) = std::net::TcpListener::bind("127.0.0.1:9998") {
(std::net::TcpStream::connect("127.0.0.1:9998").unwrap(), listener.accept().unwrap().0)
} else if let Ok(listener) = std::net::TcpListener::bind("127.0.0.1:9999") {
(std::net::TcpStream::connect("127.0.0.1:9999").unwrap(), listener.accept().unwrap().0)
} else if let Ok(listener) = std::net::TcpListener::bind("127.0.0.1:46926") {
(std::net::TcpStream::connect("127.0.0.1:46926").unwrap(), listener.accept().unwrap().0)
} else { panic!("Failed to bind to v4 localhost on common ports"); }
}
async fn do_basic_connection_test() {
let secp_ctx = Secp256k1::new();
let a_key = SecretKey::from_slice(&[1; 32]).unwrap();
let b_key = SecretKey::from_slice(&[1; 32]).unwrap();
let a_pub = PublicKey::from_secret_key(&secp_ctx, &a_key);
let b_pub = PublicKey::from_secret_key(&secp_ctx, &b_key);
let (a_connected_sender, mut a_connected) = mpsc::channel(1);
let (a_disconnected_sender, mut a_disconnected) = mpsc::channel(1);
let a_handler = Arc::new(MsgHandler {
expected_pubkey: b_pub,
pubkey_connected: a_connected_sender,
pubkey_disconnected: a_disconnected_sender,
disconnected_flag: AtomicBool::new(false),
msg_events: Mutex::new(Vec::new()),
});
let a_manager = Arc::new(PeerManager::new(MessageHandler {
chan_handler: Arc::clone(&a_handler),
route_handler: Arc::clone(&a_handler),
onion_message_handler: 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);
let b_handler = Arc::new(MsgHandler {
expected_pubkey: a_pub,
pubkey_connected: b_connected_sender,
pubkey_disconnected: b_disconnected_sender,
disconnected_flag: AtomicBool::new(false),
msg_events: Mutex::new(Vec::new()),
});
let b_manager = Arc::new(PeerManager::new(MessageHandler {
chan_handler: Arc::clone(&b_handler),
route_handler: Arc::clone(&b_handler),
onion_message_handler: 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))));
let (conn_a, conn_b) = make_tcp_connection();
let fut_a = super::setup_outbound(Arc::clone(&a_manager), b_pub, conn_a);
let fut_b = super::setup_inbound(b_manager, conn_b);
tokio::time::timeout(Duration::from_secs(10), a_connected.recv()).await.unwrap();
tokio::time::timeout(Duration::from_secs(1), b_connected.recv()).await.unwrap();
a_handler.msg_events.lock().unwrap().push(MessageSendEvent::HandleError {
node_id: b_pub, action: ErrorAction::DisconnectPeer { msg: None }
});
assert!(!a_handler.disconnected_flag.load(Ordering::SeqCst));
assert!(!b_handler.disconnected_flag.load(Ordering::SeqCst));
a_manager.process_events();
tokio::time::timeout(Duration::from_secs(10), a_disconnected.recv()).await.unwrap();
tokio::time::timeout(Duration::from_secs(1), b_disconnected.recv()).await.unwrap();
assert!(a_handler.disconnected_flag.load(Ordering::SeqCst));
assert!(b_handler.disconnected_flag.load(Ordering::SeqCst));
fut_a.await;
fut_b.await;
}
#[tokio::test(flavor = "multi_thread")]
async fn basic_threaded_connection_test() {
do_basic_connection_test().await;
}
#[tokio::test]
async fn basic_unthreaded_connection_test() {
do_basic_connection_test().await;
}
async fn race_disconnect_accept() {
let secp_ctx = Secp256k1::new();
let a_key = SecretKey::from_slice(&[1; 32]).unwrap();
let b_key = SecretKey::from_slice(&[2; 32]).unwrap();
let b_pub = PublicKey::from_secret_key(&secp_ctx, &b_key);
let a_manager = Arc::new(PeerManager::new(MessageHandler {
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{}),
}, 0, &[1; 32], Arc::new(TestLogger()), Arc::new(lightning::ln::peer_handler::IgnoringMessageHandler{}), Arc::new(TestNodeSigner::new(a_key))));
let conn_a = {
let (conn_a, _) = make_tcp_connection();
conn_a
};
let conn_b = {
let (_, conn_b) = make_tcp_connection();
conn_b
};
let manager_reference = Arc::clone(&a_manager);
tokio::spawn(async move {
super::setup_inbound(manager_reference, conn_a).await
});
tokio::spawn(async move {
super::setup_outbound(a_manager, b_pub, conn_b).await
});
}
#[tokio::test(flavor = "multi_thread")]
async fn threaded_race_disconnect_accept() {
race_disconnect_accept().await;
}
#[tokio::test]
async fn unthreaded_race_disconnect_accept() {
race_disconnect_accept().await;
}
}