/// indicating that read events on this descriptor should resume. A resume_read of false does
/// *not* imply that further read events should be paused.
fn send_data(&mut self, data: &[u8], resume_read: bool) -> usize;
- /// Disconnect the socket pointed to by this SocketDescriptor. Once this function returns, no
- /// more calls to write_buffer_space_avail, read_event or socket_disconnected may be made with
- /// this descriptor. No socket_disconnected call should be generated as a result of this call,
- /// though races may occur whereby disconnect_socket is called after a call to
- /// socket_disconnected but prior to socket_disconnected returning.
+ /// Disconnect the socket pointed to by this SocketDescriptor.
+ /// No [`PeerManager::socket_disconnected`] call need be generated as a result of this call.
fn disconnect_socket(&mut self);
}
pub fn write_buffer_space_avail(&self, descriptor: &mut Descriptor) -> Result<(), PeerHandleError> {
let mut peers = self.peers.lock().unwrap();
match peers.peers.get_mut(descriptor) {
- None => panic!("Descriptor for write_event is not already known to PeerManager"),
+ None => {
+ // This is most likely a simple race condition where the user found that the socket
+ // was writeable, then we told the user to `disconnect_socket()`, then they called
+ // this method. Return an error to make sure we get disconnected.
+ return Err(PeerHandleError { no_connection_possible: false });
+ },
Some(peer) => {
peer.awaiting_write_event = false;
self.do_attempt_write_data(descriptor, peer);
/// If Ok(true) is returned, further read_events should not be triggered until a send_data call
/// on this file descriptor has resume_read set (preventing DoS issues in the send buffer).
///
- /// Panics if the descriptor was not previously registered in a new_*_connection event.
pub fn read_event(&self, peer_descriptor: &mut Descriptor, data: &[u8]) -> Result<bool, PeerHandleError> {
match self.do_read_event(peer_descriptor, data) {
Ok(res) => Ok(res),
let mut msgs_to_forward = Vec::new();
let mut peer_node_id = None;
let pause_read = match peers.peers.get_mut(peer_descriptor) {
- None => panic!("Descriptor for read_event is not already known to PeerManager"),
+ None => {
+ // This is most likely a simple race condition where the user read some bytes
+ // from the socket, then we told the user to `disconnect_socket()`, then they
+ // called this method. Return an error to make sure we get disconnected.
+ return Err(PeerHandleError { no_connection_possible: false });
+ },
Some(peer) => {
assert!(peer.pending_read_buffer.len() > 0);
assert!(peer.pending_read_buffer.len() > peer.pending_read_buffer_pos);
self.message_handler.chan_handler.handle_announcement_signatures(&peer.their_node_id.unwrap(), &msg);
},
wire::Message::ChannelAnnouncement(msg) => {
- if match self.message_handler.route_handler.handle_channel_announcement(&msg) {
- Ok(v) => v,
- Err(e) => { return Err(e.into()); },
- } {
+ if self.message_handler.route_handler.handle_channel_announcement(&msg)
+ .map_err(|e| -> MessageHandlingError { e.into() })? {
should_forward = Some(wire::Message::ChannelAnnouncement(msg));
}
},
wire::Message::NodeAnnouncement(msg) => {
- if match self.message_handler.route_handler.handle_node_announcement(&msg) {
- Ok(v) => v,
- Err(e) => { return Err(e.into()); },
- } {
+ if self.message_handler.route_handler.handle_node_announcement(&msg)
+ .map_err(|e| -> MessageHandlingError { e.into() })? {
should_forward = Some(wire::Message::NodeAnnouncement(msg));
}
},
wire::Message::ChannelUpdate(msg) => {
self.message_handler.chan_handler.handle_channel_update(&peer.their_node_id.unwrap(), &msg);
- if match self.message_handler.route_handler.handle_channel_update(&msg) {
- Ok(v) => v,
- Err(e) => { return Err(e.into()); },
- } {
+ if self.message_handler.route_handler.handle_channel_update(&msg)
+ .map_err(|e| -> MessageHandlingError { e.into() })? {
should_forward = Some(wire::Message::ChannelUpdate(msg));
}
},
macro_rules! get_peer_for_forwarding {
($node_id: expr) => {
{
- let descriptor = match peers.node_id_to_descriptor.get($node_id) {
- Some(descriptor) => descriptor.clone(),
+ match peers.node_id_to_descriptor.get($node_id) {
+ Some(descriptor) => match peers.peers.get_mut(&descriptor) {
+ Some(peer) => {
+ if peer.their_features.is_none() {
+ continue;
+ }
+ peer
+ },
+ None => panic!("Inconsistent peers set state!"),
+ },
None => {
continue;
},
- };
- match peers.peers.get_mut(&descriptor) {
- Some(peer) => {
- if peer.their_features.is_none() {
- continue;
- }
- (descriptor, peer)
- },
- None => panic!("Inconsistent peers set state!"),
}
}
}
log_trace!(self.logger, "Handling SendAcceptChannel event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.temporary_channel_id));
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => {
log_trace!(self.logger, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.temporary_channel_id));
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => {
log_funding_channel_id!(msg.funding_txid, msg.funding_output_index));
// TODO: If the peer is gone we should generate a DiscardFunding event
// indicating to the wallet that they should just throw away this funding transaction
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
MessageSendEvent::SendFundingSigned { ref node_id, ref msg } => {
log_trace!(self.logger, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => {
log_trace!(self.logger, "Handling SendFundingLocked event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => {
log_trace!(self.logger, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
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 } } => {
update_fulfill_htlcs.len(),
update_fail_htlcs.len(),
log_bytes!(commitment_signed.channel_id));
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
for msg in update_add_htlcs {
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
}
log_trace!(self.logger, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => {
log_trace!(self.logger, "Handling SendClosingSigned event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
MessageSendEvent::SendShutdown { ref node_id, ref msg } => {
log_trace!(self.logger, "Handling Shutdown event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
log_trace!(self.logger, "Handling SendChannelReestablish event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
MessageSendEvent::BroadcastChannelAnnouncement { msg, update_msg } => {
log_trace!(self.logger, "Handling SendErrorMessage HandleError event in peer_handler for node {} with message {}",
log_pubkey!(node_id),
msg.data);
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
}
},
MessageSendEvent::SendChannelRangeQuery { ref node_id, ref msg } => {
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
},
MessageSendEvent::SendShortIdsQuery { ref node_id, ref msg } => {
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
}
MessageSendEvent::SendReplyChannelRange { ref node_id, ref msg } => {
msg.first_blocknum,
msg.number_of_blocks,
msg.sync_complete);
- let (_, peer) = get_peer_for_forwarding!(node_id);
+ let peer = get_peer_for_forwarding!(node_id);
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
}
}
/// Indicates that the given socket descriptor's connection is now closed.
///
- /// This must only be called if the socket has been disconnected by the peer or your own
- /// decision to disconnect it and must NOT be called in any case where other parts of this
- /// library (eg PeerHandleError, explicit disconnect_socket calls) instruct you to disconnect
- /// the peer.
- ///
- /// Panics if the descriptor was not previously registered in a successful new_*_connection event.
+ /// This need only be called if the socket has been disconnected by the peer or your own
+ /// decision to disconnect it and may be skipped in any case where other parts of this library
+ /// (eg PeerHandleError, explicit disconnect_socket calls) instruct you to disconnect the peer.
pub fn socket_disconnected(&self, descriptor: &Descriptor) {
self.disconnect_event_internal(descriptor, false);
}
let mut peers = self.peers.lock().unwrap();
let peer_option = peers.peers.remove(descriptor);
match peer_option {
- None => panic!("Descriptor for disconnect_event is not already known to PeerManager"),
+ None => {
+ // This is most likely a simple race condition where the user found that the socket
+ // was disconnected, then we told the user to `disconnect_socket()`, then they
+ // called this method. Either way we're disconnected, return.
+ },
Some(peer) => {
match peer.their_node_id {
Some(node_id) => {
projects / rust-lightning / blobdiff