use ln::msgs;
use ln::msgs::ChannelMessageHandler;
use ln::channelmanager::{SimpleArcChannelManager, SimpleRefChannelManager};
+use util::ser::VecWriter;
use ln::peer_channel_encryptor::{PeerChannelEncryptor,NextNoiseStep};
use ln::wire;
use ln::wire::Encode;
use util::byte_utils;
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};
peer_counter_low: AtomicUsize,
peer_counter_high: AtomicUsize,
- initial_syncs_sent: AtomicUsize,
logger: Arc<Logger>,
}
-struct VecWriter(Vec<u8>);
-impl Writer for VecWriter {
- fn write_all(&mut self, buf: &[u8]) -> Result<(), ::std::io::Error> {
- self.0.extend_from_slice(buf);
- Ok(())
- }
- fn size_hint(&mut self, size: usize) {
- self.0.reserve_exact(size);
- }
-}
-
macro_rules! encode_msg {
($msg: expr) => {{
let mut buffer = VecWriter(Vec::new());
}}
}
-//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, CM: Deref> PeerManager<Descriptor, CM> where CM::Target: msgs::ChannelMessageHandler {
ephemeral_key_midstate,
peer_counter_low: AtomicUsize::new(0),
peer_counter_high: AtomicUsize::new(0),
- initial_syncs_sent: AtomicUsize::new(0),
logger,
}
}
}
}
- macro_rules! try_potential_decodeerror {
- ($thing: expr) => {
- match $thing {
- Ok(x) => x,
- Err(e) => {
- match e {
- msgs::DecodeError::UnknownVersion => return Err(PeerHandleError{ no_connection_possible: false }),
- msgs::DecodeError::UnknownRequiredFeature => {
- log_debug!(self, "Got a channel/node announcement with an known required feature flag, you may want to update!");
- continue;
- },
- msgs::DecodeError::InvalidValue => {
- log_debug!(self, "Got an invalid value while deserializing message");
- return Err(PeerHandleError{ no_connection_possible: false });
- },
- msgs::DecodeError::ShortRead => {
- log_debug!(self, "Deserialization failed due to shortness of message");
- return Err(PeerHandleError{ no_connection_possible: false });
- },
- msgs::DecodeError::ExtraAddressesPerType => {
- log_debug!(self, "Error decoding message, ignoring due to lnd spec incompatibility. See https://github.com/lightningnetwork/lnd/issues/1407");
- continue;
- },
- msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError{ no_connection_possible: false }),
- msgs::DecodeError::Io(_) => return Err(PeerHandleError{ no_connection_possible: false }),
- }
- }
- };
- }
- }
-
macro_rules! insert_node_id {
() => {
match peers.node_id_to_descriptor.entry(peer.their_node_id.unwrap()) {
peer.their_node_id = Some(their_node_id);
insert_node_id!();
let mut features = InitFeatures::supported();
- if self.initial_syncs_sent.load(Ordering::Acquire) < INITIAL_SYNCS_TO_SEND {
- self.initial_syncs_sent.fetch_add(1, Ordering::AcqRel);
+ if self.message_handler.route_handler.should_request_full_sync(&peer.their_node_id.unwrap()) {
features.set_initial_routing_sync();
}
peer.pending_read_is_header = true;
let mut reader = ::std::io::Cursor::new(&msg_data[..]);
- let message = try_potential_decodeerror!(wire::read(&mut reader));
+ let message_result = wire::read(&mut reader);
+ let message = match message_result {
+ Ok(x) => x,
+ Err(e) => {
+ match e {
+ msgs::DecodeError::UnknownVersion => return Err(PeerHandleError { no_connection_possible: false }),
+ msgs::DecodeError::UnknownRequiredFeature => {
+ log_debug!(self, "Got a channel/node announcement with an known required feature flag, you may want to update!");
+ continue;
+ }
+ msgs::DecodeError::InvalidValue => {
+ log_debug!(self, "Got an invalid value while deserializing message");
+ return Err(PeerHandleError { no_connection_possible: false });
+ }
+ msgs::DecodeError::ShortRead => {
+ log_debug!(self, "Deserialization failed due to shortness of message");
+ return Err(PeerHandleError { no_connection_possible: false });
+ }
+ msgs::DecodeError::ExtraAddressesPerType => {
+ log_debug!(self, "Error decoding message, ignoring due to lnd spec incompatibility. See https://github.com/lightningnetwork/lnd/issues/1407");
+ continue;
+ }
+ msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError { no_connection_possible: false }),
+ msgs::DecodeError::Io(_) => return Err(PeerHandleError { no_connection_possible: false }),
+ }
+ }
+ };
+
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 !peer.outbound {
let mut features = InitFeatures::supported();
- if self.initial_syncs_sent.load(Ordering::Acquire) < INITIAL_SYNCS_TO_SEND {
- self.initial_syncs_sent.fetch_add(1, Ordering::AcqRel);
+ if self.message_handler.route_handler.should_request_full_sync(&peer.their_node_id.unwrap()) {
features.set_initial_routing_sync();
}
},
// Unknown messages:
- wire::Message::Unknown(msg_type) => {
+ wire::Message::Unknown(msg_type) if msg_type.is_even() => {
// Fail the channel if message is an even, unknown type as per BOLT #1.
- if (msg_type & 1) == 0 {
- return Err(PeerHandleError{ no_connection_possible: true });
- }
+ return Err(PeerHandleError{ no_connection_possible: true });
},
+ wire::Message::Unknown(_) => {},
}
}
}