use secp256k1::key::{SecretKey,PublicKey};
+use ln::features::InitFeatures;
use ln::msgs;
-use util::ser::{Writeable, Writer, Readable};
+use ln::msgs::ChannelMessageHandler;
+use ln::channelmanager::{SimpleArcChannelManager, SimpleRefChannelManager};
use ln::peer_channel_encryptor::{PeerChannelEncryptor,NextNoiseStep};
+use ln::wire;
+use ln::wire::Encode;
use util::byte_utils;
-use util::events::{MessageSendEvent};
+use util::events::{MessageSendEvent, MessageSendEventsProvider};
use util::logger::Logger;
+use util::ser::Writer;
use std::collections::{HashMap,hash_map,HashSet,LinkedList};
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::{cmp,error,hash,fmt};
+use std::ops::Deref;
use bitcoin_hashes::sha256::Hash as Sha256;
use bitcoin_hashes::sha256::HashEngine as Sha256Engine;
use bitcoin_hashes::{HashEngine, Hash};
/// Provides references to trait impls which handle different types of messages.
-pub struct MessageHandler {
+pub struct MessageHandler<CM: Deref> where CM::Target: msgs::ChannelMessageHandler {
/// A message handler which handles messages specific to channels. Usually this is just a
/// ChannelManager object.
- pub chan_handler: Arc<msgs::ChannelMessageHandler>,
+ pub chan_handler: CM,
/// A message handler which handles messages updating our knowledge of the network channel
/// graph. Usually this is just a Router object.
pub route_handler: Arc<msgs::RoutingMessageHandler>,
channel_encryptor: PeerChannelEncryptor,
outbound: bool,
their_node_id: Option<PublicKey>,
- their_features: Option<msgs::InitFeatures>,
+ their_features: Option<InitFeatures>,
pending_outbound_buffer: LinkedList<Vec<u8>>,
pending_outbound_buffer_first_msg_offset: usize,
/// Only add to this set when noise completes:
node_id_to_descriptor: HashMap<PublicKey, Descriptor>,
}
-struct MutPeerHolder<'a, Descriptor: SocketDescriptor + 'a> {
- peers: &'a mut HashMap<Descriptor, Peer>,
- peers_needing_send: &'a mut HashSet<Descriptor>,
- node_id_to_descriptor: &'a mut HashMap<PublicKey, Descriptor>,
-}
-impl<Descriptor: SocketDescriptor> PeerHolder<Descriptor> {
- fn borrow_parts(&mut self) -> MutPeerHolder<Descriptor> {
- MutPeerHolder {
- peers: &mut self.peers,
- peers_needing_send: &mut self.peers_needing_send,
- node_id_to_descriptor: &mut self.node_id_to_descriptor,
- }
- }
-}
#[cfg(not(any(target_pointer_width = "32", target_pointer_width = "64")))]
fn _check_usize_is_32_or_64() {
unsafe { mem::transmute::<*const usize, [u8; 4]>(panic!()); }
}
+/// SimpleArcPeerManager is useful when you need a PeerManager with a static lifetime, e.g.
+/// when you're using lightning-net-tokio (since tokio::spawn requires parameters with static
+/// lifetimes). Other times you can afford a reference, which is more efficient, in which case
+/// SimpleRefPeerManager is the more appropriate type. Defining these type aliases prevents
+/// issues such as overly long function definitions.
+pub type SimpleArcPeerManager<SD, M> = Arc<PeerManager<SD, SimpleArcChannelManager<M>>>;
+
+/// SimpleRefPeerManager is a type alias for a PeerManager reference, and is the reference
+/// counterpart to the SimpleArcPeerManager type alias. Use this type by default when you don't
+/// need a PeerManager with a static lifetime. You'll need a static lifetime in cases such as
+/// usage of lightning-net-tokio (since tokio::spawn requires parameters with static lifetimes).
+/// But if this is not necessary, using a reference is more efficient. Defining these type aliases
+/// helps with issues such as long function definitions.
+pub type SimpleRefPeerManager<'a, SD, M> = PeerManager<SD, SimpleRefChannelManager<'a, M>>;
+
/// A PeerManager manages a set of peers, described by their SocketDescriptor and marshalls socket
/// events into messages which it passes on to its MessageHandlers.
-pub struct PeerManager<Descriptor: SocketDescriptor> {
- message_handler: MessageHandler,
+///
+/// Rather than using a plain PeerManager, it is preferable to use either a SimpleArcPeerManager
+/// a SimpleRefPeerManager, for conciseness. See their documentation for more details, but
+/// essentially you should default to using a SimpleRefPeerManager, and use a
+/// SimpleArcPeerManager when you require a PeerManager with a static lifetime, such as when
+/// you're using lightning-net-tokio.
+pub struct PeerManager<Descriptor: SocketDescriptor, CM: Deref> where CM::Target: msgs::ChannelMessageHandler {
+ message_handler: MessageHandler<CM>,
peers: Mutex<PeerHolder<Descriptor>>,
our_node_secret: SecretKey,
ephemeral_key_midstate: Sha256Engine,
peer_counter_low: AtomicUsize,
peer_counter_high: AtomicUsize,
- initial_syncs_sent: AtomicUsize,
logger: Arc<Logger>,
}
}
macro_rules! encode_msg {
- ($msg: expr, $msg_code: expr) => {{
- let mut msg = VecWriter(Vec::new());
- ($msg_code as u16).write(&mut msg).unwrap();
- $msg.write(&mut msg).unwrap();
- msg.0
+ ($msg: expr) => {{
+ let mut buffer = VecWriter(Vec::new());
+ wire::write($msg, &mut buffer).unwrap();
+ buffer.0
}}
}
-//TODO: Really should do something smarter for this
-const INITIAL_SYNCS_TO_SEND: usize = 5;
-
/// Manages and reacts to connection events. You probably want to use file descriptors as PeerIds.
/// PeerIds may repeat, but only after disconnect_event() has been called.
-impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
+impl<Descriptor: SocketDescriptor, CM: Deref> PeerManager<Descriptor, CM> where CM::Target: msgs::ChannelMessageHandler {
/// Constructs a new PeerManager with the given message handlers and node_id secret key
/// ephemeral_random_data is used to derive per-connection ephemeral keys and must be
/// cryptographically secure random bytes.
- pub fn new(message_handler: MessageHandler, our_node_secret: SecretKey, ephemeral_random_data: &[u8; 32], logger: Arc<Logger>) -> PeerManager<Descriptor> {
+ pub fn new(message_handler: MessageHandler<CM>, our_node_secret: SecretKey, ephemeral_random_data: &[u8; 32], logger: Arc<Logger>) -> PeerManager<Descriptor, CM> {
let mut ephemeral_key_midstate = Sha256::engine();
ephemeral_key_midstate.input(ephemeral_random_data);
ephemeral_key_midstate,
peer_counter_low: AtomicUsize::new(0),
peer_counter_high: AtomicUsize::new(0),
- initial_syncs_sent: AtomicUsize::new(0),
logger,
}
}
fn do_attempt_write_data(&self, descriptor: &mut Descriptor, peer: &mut Peer) {
macro_rules! encode_and_send_msg {
- ($msg: expr, $msg_code: expr) => {
+ ($msg: expr) => {
{
- log_trace!(self, "Encoding and sending sync update message of type {} to {}", $msg_code, log_pubkey!(peer.their_node_id.unwrap()));
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg, $msg_code)[..]));
+ log_trace!(self, "Encoding and sending sync update message of type {} to {}", $msg.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg)[..]));
}
}
}
let steps = ((MSG_BUFF_SIZE - peer.pending_outbound_buffer.len() + 2) / 3) as u8;
let all_messages = self.message_handler.route_handler.get_next_channel_announcements(0, steps);
for &(ref announce, ref update_a, ref update_b) in all_messages.iter() {
- encode_and_send_msg!(announce, 256);
- encode_and_send_msg!(update_a, 258);
- encode_and_send_msg!(update_b, 258);
+ encode_and_send_msg!(announce);
+ encode_and_send_msg!(update_a);
+ encode_and_send_msg!(update_b);
peer.sync_status = InitSyncTracker::ChannelsSyncing(announce.contents.short_channel_id + 1);
}
if all_messages.is_empty() || all_messages.len() != steps as usize {
let steps = (MSG_BUFF_SIZE - peer.pending_outbound_buffer.len()) as u8;
let all_messages = self.message_handler.route_handler.get_next_node_announcements(None, steps);
for msg in all_messages.iter() {
- encode_and_send_msg!(msg, 256);
+ encode_and_send_msg!(msg);
peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
}
if all_messages.is_empty() || all_messages.len() != steps as usize {
let steps = (MSG_BUFF_SIZE - peer.pending_outbound_buffer.len()) as u8;
let all_messages = self.message_handler.route_handler.get_next_node_announcements(Some(&key), steps);
for msg in all_messages.iter() {
- encode_and_send_msg!(msg, 256);
+ encode_and_send_msg!(msg);
peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
}
if all_messages.is_empty() || all_messages.len() != steps as usize {
fn do_read_event(&self, peer_descriptor: &mut Descriptor, data: Vec<u8>) -> Result<bool, PeerHandleError> {
let pause_read = {
let mut peers_lock = self.peers.lock().unwrap();
- let peers = peers_lock.borrow_parts();
+ let peers = &mut *peers_lock;
let pause_read = match peers.peers.get_mut(peer_descriptor) {
None => panic!("Descriptor for read_event is not already known to PeerManager"),
Some(peer) => {
peer.pending_read_buffer_pos = 0;
macro_rules! encode_and_send_msg {
- ($msg: expr, $msg_code: expr) => {
+ ($msg: expr) => {
{
- log_trace!(self, "Encoding and sending message of type {} to {}", $msg_code, log_pubkey!(peer.their_node_id.unwrap()));
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg, $msg_code)[..]));
+ log_trace!(self, "Encoding and sending message of type {} to {}", $msg.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(&$msg)[..]));
peers.peers_needing_send.insert(peer_descriptor.clone());
}
}
},
msgs::ErrorAction::SendErrorMessage { msg } => {
log_trace!(self, "Got Err handling message, sending Error message because {}", e.err);
- encode_and_send_msg!(msg, 17);
+ encode_and_send_msg!(msg);
continue;
},
}
peer.their_node_id = Some(their_node_id);
insert_node_id!();
- let mut features = msgs::InitFeatures::supported();
- if self.initial_syncs_sent.load(Ordering::Acquire) < INITIAL_SYNCS_TO_SEND {
- self.initial_syncs_sent.fetch_add(1, Ordering::AcqRel);
+ let mut features = InitFeatures::supported();
+ if self.message_handler.route_handler.should_request_full_sync(&peer.their_node_id.unwrap()) {
features.set_initial_routing_sync();
}
- encode_and_send_msg!(msgs::Init {
- features,
- }, 16);
+
+ let resp = msgs::Init { features };
+ encode_and_send_msg!(resp);
},
NextNoiseStep::ActThree => {
let their_node_id = try_potential_handleerror!(peer.channel_encryptor.process_act_three(&peer.pending_read_buffer[..]));
peer.pending_read_buffer = [0; 18].to_vec();
peer.pending_read_is_header = true;
- let msg_type = byte_utils::slice_to_be16(&msg_data[0..2]);
- log_trace!(self, "Received message of type {} from {}", msg_type, log_pubkey!(peer.their_node_id.unwrap()));
- if msg_type != 16 && peer.their_features.is_none() {
- // Need an init message as first message
+ let mut reader = ::std::io::Cursor::new(&msg_data[..]);
+ let message = try_potential_decodeerror!(wire::read(&mut reader));
+ log_trace!(self, "Received message of type {} from {}", message.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
+
+ // Need an Init as first message
+ if let wire::Message::Init(_) = message {
+ } else if peer.their_features.is_none() {
log_trace!(self, "Peer {} sent non-Init first message", log_pubkey!(peer.their_node_id.unwrap()));
return Err(PeerHandleError{ no_connection_possible: false });
}
- let mut reader = ::std::io::Cursor::new(&msg_data[2..]);
- match msg_type {
- // Connection control:
- 16 => {
- let msg = try_potential_decodeerror!(msgs::Init::read(&mut reader));
+
+ match message {
+ // Setup and Control messages:
+ wire::Message::Init(msg) => {
if msg.features.requires_unknown_bits() {
log_info!(self, "Peer global features required unknown version bits");
return Err(PeerHandleError{ no_connection_possible: true });
peer.sync_status = InitSyncTracker::ChannelsSyncing(0);
peers.peers_needing_send.insert(peer_descriptor.clone());
}
- peer.their_features = Some(msg.features);
if !peer.outbound {
- let mut features = msgs::InitFeatures::supported();
- if self.initial_syncs_sent.load(Ordering::Acquire) < INITIAL_SYNCS_TO_SEND {
- self.initial_syncs_sent.fetch_add(1, Ordering::AcqRel);
+ let mut features = InitFeatures::supported();
+ if self.message_handler.route_handler.should_request_full_sync(&peer.their_node_id.unwrap()) {
features.set_initial_routing_sync();
}
- encode_and_send_msg!(msgs::Init {
- features,
- }, 16);
+ let resp = msgs::Init { features };
+ encode_and_send_msg!(resp);
}
- self.message_handler.chan_handler.peer_connected(&peer.their_node_id.unwrap());
+ self.message_handler.chan_handler.peer_connected(&peer.their_node_id.unwrap(), &msg);
+ peer.their_features = Some(msg.features);
},
- 17 => {
- let msg = try_potential_decodeerror!(msgs::ErrorMessage::read(&mut reader));
+ wire::Message::Error(msg) => {
let mut data_is_printable = true;
for b in msg.data.bytes() {
if b < 32 || b > 126 {
}
},
- 18 => {
- let msg = try_potential_decodeerror!(msgs::Ping::read(&mut reader));
+ wire::Message::Ping(msg) => {
if msg.ponglen < 65532 {
let resp = msgs::Pong { byteslen: msg.ponglen };
- encode_and_send_msg!(resp, 19);
+ encode_and_send_msg!(resp);
}
},
- 19 => {
+ wire::Message::Pong(_msg) => {
peer.awaiting_pong = false;
- try_potential_decodeerror!(msgs::Pong::read(&mut reader));
},
- // Channel control:
- 32 => {
- let msg = try_potential_decodeerror!(msgs::OpenChannel::read(&mut reader));
+
+ // Channel messages:
+ wire::Message::OpenChannel(msg) => {
self.message_handler.chan_handler.handle_open_channel(&peer.their_node_id.unwrap(), peer.their_features.clone().unwrap(), &msg);
},
- 33 => {
- let msg = try_potential_decodeerror!(msgs::AcceptChannel::read(&mut reader));
+ wire::Message::AcceptChannel(msg) => {
self.message_handler.chan_handler.handle_accept_channel(&peer.their_node_id.unwrap(), peer.their_features.clone().unwrap(), &msg);
},
- 34 => {
- let msg = try_potential_decodeerror!(msgs::FundingCreated::read(&mut reader));
+ wire::Message::FundingCreated(msg) => {
self.message_handler.chan_handler.handle_funding_created(&peer.their_node_id.unwrap(), &msg);
},
- 35 => {
- let msg = try_potential_decodeerror!(msgs::FundingSigned::read(&mut reader));
+ wire::Message::FundingSigned(msg) => {
self.message_handler.chan_handler.handle_funding_signed(&peer.their_node_id.unwrap(), &msg);
},
- 36 => {
- let msg = try_potential_decodeerror!(msgs::FundingLocked::read(&mut reader));
+ wire::Message::FundingLocked(msg) => {
self.message_handler.chan_handler.handle_funding_locked(&peer.their_node_id.unwrap(), &msg);
},
- 38 => {
- let msg = try_potential_decodeerror!(msgs::Shutdown::read(&mut reader));
+ wire::Message::Shutdown(msg) => {
self.message_handler.chan_handler.handle_shutdown(&peer.their_node_id.unwrap(), &msg);
},
- 39 => {
- let msg = try_potential_decodeerror!(msgs::ClosingSigned::read(&mut reader));
+ wire::Message::ClosingSigned(msg) => {
self.message_handler.chan_handler.handle_closing_signed(&peer.their_node_id.unwrap(), &msg);
},
- 128 => {
- let msg = try_potential_decodeerror!(msgs::UpdateAddHTLC::read(&mut reader));
+ // Commitment messages:
+ wire::Message::UpdateAddHTLC(msg) => {
self.message_handler.chan_handler.handle_update_add_htlc(&peer.their_node_id.unwrap(), &msg);
},
- 130 => {
- let msg = try_potential_decodeerror!(msgs::UpdateFulfillHTLC::read(&mut reader));
+ wire::Message::UpdateFulfillHTLC(msg) => {
self.message_handler.chan_handler.handle_update_fulfill_htlc(&peer.their_node_id.unwrap(), &msg);
},
- 131 => {
- let msg = try_potential_decodeerror!(msgs::UpdateFailHTLC::read(&mut reader));
+ wire::Message::UpdateFailHTLC(msg) => {
self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg);
},
- 135 => {
- let msg = try_potential_decodeerror!(msgs::UpdateFailMalformedHTLC::read(&mut reader));
+ wire::Message::UpdateFailMalformedHTLC(msg) => {
self.message_handler.chan_handler.handle_update_fail_malformed_htlc(&peer.their_node_id.unwrap(), &msg);
},
- 132 => {
- let msg = try_potential_decodeerror!(msgs::CommitmentSigned::read(&mut reader));
+ wire::Message::CommitmentSigned(msg) => {
self.message_handler.chan_handler.handle_commitment_signed(&peer.their_node_id.unwrap(), &msg);
},
- 133 => {
- let msg = try_potential_decodeerror!(msgs::RevokeAndACK::read(&mut reader));
+ wire::Message::RevokeAndACK(msg) => {
self.message_handler.chan_handler.handle_revoke_and_ack(&peer.their_node_id.unwrap(), &msg);
},
- 134 => {
- let msg = try_potential_decodeerror!(msgs::UpdateFee::read(&mut reader));
+ wire::Message::UpdateFee(msg) => {
self.message_handler.chan_handler.handle_update_fee(&peer.their_node_id.unwrap(), &msg);
},
- 136 => {
- let msg = try_potential_decodeerror!(msgs::ChannelReestablish::read(&mut reader));
+ wire::Message::ChannelReestablish(msg) => {
self.message_handler.chan_handler.handle_channel_reestablish(&peer.their_node_id.unwrap(), &msg);
},
- // Routing control:
- 259 => {
- let msg = try_potential_decodeerror!(msgs::AnnouncementSignatures::read(&mut reader));
+ // Routing messages:
+ wire::Message::AnnouncementSignatures(msg) => {
self.message_handler.chan_handler.handle_announcement_signatures(&peer.their_node_id.unwrap(), &msg);
},
- 256 => {
- let msg = try_potential_decodeerror!(msgs::ChannelAnnouncement::read(&mut reader));
+ wire::Message::ChannelAnnouncement(msg) => {
let should_forward = try_potential_handleerror!(self.message_handler.route_handler.handle_channel_announcement(&msg));
if should_forward {
// TODO: forward msg along to all our other peers!
}
},
- 257 => {
- let msg = try_potential_decodeerror!(msgs::NodeAnnouncement::read(&mut reader));
+ wire::Message::NodeAnnouncement(msg) => {
let should_forward = try_potential_handleerror!(self.message_handler.route_handler.handle_node_announcement(&msg));
if should_forward {
// TODO: forward msg along to all our other peers!
}
},
- 258 => {
- let msg = try_potential_decodeerror!(msgs::ChannelUpdate::read(&mut reader));
+ wire::Message::ChannelUpdate(msg) => {
let should_forward = try_potential_handleerror!(self.message_handler.route_handler.handle_channel_update(&msg));
if should_forward {
// TODO: forward msg along to all our other peers!
}
},
- _ => {
- if (msg_type & 1) == 0 {
- return Err(PeerHandleError{ no_connection_possible: true });
- }
+
+ // Unknown messages:
+ wire::Message::Unknown(msg_type) if msg_type.is_even() => {
+ // Fail the channel if message is an even, unknown type as per BOLT #1.
+ return Err(PeerHandleError{ no_connection_possible: true });
},
+ wire::Message::Unknown(_) => {},
}
}
}
let mut events_generated = self.message_handler.chan_handler.get_and_clear_pending_msg_events();
let mut peers_lock = self.peers.lock().unwrap();
- let peers = peers_lock.borrow_parts();
+ let peers = &mut *peers_lock;
for event in events_generated.drain(..) {
macro_rules! get_peer_for_forwarding {
($node_id: expr, $handle_no_such_peer: block) => {
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
//TODO: Drop the pending channel? (or just let it timeout, but that sucks)
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 33)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => {
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
//TODO: Drop the pending channel? (or just let it timeout, but that sucks)
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 32)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => {
//TODO: generate a DiscardFunding event indicating to the wallet that
//they should just throw away this funding transaction
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 34)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::SendFundingSigned { ref node_id, ref msg } => {
//TODO: generate a DiscardFunding event indicating to the wallet that
//they should just throw away this funding transaction
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 35)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => {
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
//TODO: Do whatever we're gonna do for handling dropped messages
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 36)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => {
//TODO: generate a DiscardFunding event indicating to the wallet that
//they should just throw away this funding transaction
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 259)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref update_fee, ref commitment_signed } } => {
//TODO: Do whatever we're gonna do for handling dropped messages
});
for msg in update_add_htlcs {
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 128)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
}
for msg in update_fulfill_htlcs {
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 130)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
}
for msg in update_fail_htlcs {
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 131)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
}
for msg in update_fail_malformed_htlcs {
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 135)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
}
if let &Some(ref msg) = update_fee {
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 134)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
}
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_signed, 132)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_signed)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
//TODO: Do whatever we're gonna do for handling dropped messages
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 133)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => {
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
//TODO: Do whatever we're gonna do for handling dropped messages
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 39)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::SendShutdown { ref node_id, ref msg } => {
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
//TODO: Do whatever we're gonna do for handling dropped messages
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 38)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
//TODO: Do whatever we're gonna do for handling dropped messages
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 136)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
log_trace!(self, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
if self.message_handler.route_handler.handle_channel_announcement(msg).is_ok() && self.message_handler.route_handler.handle_channel_update(update_msg).is_ok() {
- let encoded_msg = encode_msg!(msg, 256);
- let encoded_update_msg = encode_msg!(update_msg, 258);
+ let encoded_msg = encode_msg!(msg);
+ let encoded_update_msg = encode_msg!(update_msg);
for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
MessageSendEvent::BroadcastChannelUpdate { ref msg } => {
log_trace!(self, "Handling BroadcastChannelUpdate event in peer_handler for short channel id {}", msg.contents.short_channel_id);
if self.message_handler.route_handler.handle_channel_update(msg).is_ok() {
- let encoded_msg = encode_msg!(msg, 258);
+ let encoded_msg = encode_msg!(msg);
for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
log_trace!(self, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}",
log_pubkey!(node_id),
msg.data);
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
// This isn't guaranteed to work, but if there is enough free
// room in the send buffer, put the error message there...
self.do_attempt_write_data(&mut descriptor, &mut peer);
let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
//TODO: Do whatever we're gonna do for handling dropped messages
});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
self.do_attempt_write_data(&mut descriptor, peer);
},
}
pub fn timer_tick_occured(&self) {
let mut peers_lock = self.peers.lock().unwrap();
{
- let peers = peers_lock.borrow_parts();
- let peers_needing_send = peers.peers_needing_send;
- let node_id_to_descriptor = peers.node_id_to_descriptor;
- let peers = peers.peers;
+ let peers = &mut *peers_lock;
+ let peers_needing_send = &mut peers.peers_needing_send;
+ let node_id_to_descriptor = &mut peers.node_id_to_descriptor;
+ let peers = &mut peers.peers;
peers.retain(|descriptor, peer| {
if peer.awaiting_pong == true {
ponglen: 0,
byteslen: 64,
};
- peer.pending_outbound_buffer.push_back(encode_msg!(ping, 18));
+ peer.pending_outbound_buffer.push_back(encode_msg!(&ping));
let mut descriptor_clone = descriptor.clone();
self.do_attempt_write_data(&mut descriptor_clone, peer);
fn disconnect_socket(&mut self) {}
}
- fn create_network(peer_count: usize) -> Vec<PeerManager<FileDescriptor>> {
+ fn create_chan_handlers(peer_count: usize) -> Vec<test_utils::TestChannelMessageHandler> {
+ let mut chan_handlers = Vec::new();
+ for _ in 0..peer_count {
+ let chan_handler = test_utils::TestChannelMessageHandler::new();
+ chan_handlers.push(chan_handler);
+ }
+
+ chan_handlers
+ }
+
+ fn create_network<'a>(peer_count: usize, chan_handlers: &'a Vec<test_utils::TestChannelMessageHandler>) -> Vec<PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler>> {
let mut peers = Vec::new();
let mut rng = thread_rng();
let logger : Arc<Logger> = Arc::new(test_utils::TestLogger::new());
let mut ephemeral_bytes = [0; 32];
rng.fill_bytes(&mut ephemeral_bytes);
- for _ in 0..peer_count {
- let chan_handler = test_utils::TestChannelMessageHandler::new();
+ for i in 0..peer_count {
let router = test_utils::TestRoutingMessageHandler::new();
let node_id = {
let mut key_slice = [0;32];
rng.fill_bytes(&mut key_slice);
SecretKey::from_slice(&key_slice).unwrap()
};
- let msg_handler = MessageHandler { chan_handler: Arc::new(chan_handler), route_handler: Arc::new(router) };
+ let msg_handler = MessageHandler { chan_handler: &chan_handlers[i], route_handler: Arc::new(router) };
let peer = PeerManager::new(msg_handler, node_id, &ephemeral_bytes, Arc::clone(&logger));
peers.push(peer);
}
peers
}
- fn establish_connection(peer_a: &PeerManager<FileDescriptor>, peer_b: &PeerManager<FileDescriptor>) {
+ fn establish_connection<'a>(peer_a: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler>, peer_b: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler>) {
let secp_ctx = Secp256k1::new();
let their_id = PublicKey::from_secret_key(&secp_ctx, &peer_b.our_node_secret);
let fd = FileDescriptor { fd: 1};
fn test_disconnect_peer() {
// Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and
// push a DisconnectPeer event to remove the node flagged by id
- let mut peers = create_network(2);
+ let chan_handlers = create_chan_handlers(2);
+ let chan_handler = test_utils::TestChannelMessageHandler::new();
+ let mut peers = create_network(2, &chan_handlers);
establish_connection(&peers[0], &peers[1]);
assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
let secp_ctx = Secp256k1::new();
let their_id = PublicKey::from_secret_key(&secp_ctx, &peers[1].our_node_secret);
- let chan_handler = test_utils::TestChannelMessageHandler::new();
chan_handler.pending_events.lock().unwrap().push(events::MessageSendEvent::HandleError {
node_id: their_id,
action: msgs::ErrorAction::DisconnectPeer { msg: None },
});
assert_eq!(chan_handler.pending_events.lock().unwrap().len(), 1);
- peers[0].message_handler.chan_handler = Arc::new(chan_handler);
+ peers[0].message_handler.chan_handler = &chan_handler;
peers[0].process_events();
assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 0);
#[test]
fn test_timer_tick_occured(){
// Create peers, a vector of two peer managers, perform initial set up and check that peers[0] has one Peer.
- let peers = create_network(2);
+ let chan_handlers = create_chan_handlers(2);
+ let peers = create_network(2, &chan_handlers);
establish_connection(&peers[0], &peers[1]);
assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
projects / rust-lightning / blobdiff