X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fpeer_handler.rs;h=c0035f4f42c09692970b5d472af6df0d97a4193b;hb=148c79719e9177709c7b546d2636e3fa28046fc9;hp=0e390bd32fc9d777dbd7dddbc45e071683c69514;hpb=2869e50d67ad5fc18b9434c771d7c227ead800ce;p=rust-lightning diff --git a/src/ln/peer_handler.rs b/src/ln/peer_handler.rs index 0e390bd3..c0035f4f 100644 --- a/src/ln/peer_handler.rs +++ b/src/ln/peer_handler.rs @@ -45,12 +45,13 @@ pub struct MessageHandler { /// careful to ensure you don't have races whereby you might register a new connection with an fd /// the same as a yet-to-be-disconnect_event()-ed. pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone { - /// Attempts to send some data from the given Vec starting at the given offset to the peer. + /// Attempts to send some data from the given slice to the peer. + /// /// Returns the amount of data which was sent, possibly 0 if the socket has since disconnected. /// Note that in the disconnected case, a disconnect_event must still fire and further write /// attempts may occur until that time. /// - /// If the returned size is smaller than data.len() - write_offset, a write_available event must + /// If the returned size is smaller than data.len(), a write_available event must /// trigger the next time more data can be written. Additionally, until the a send_data event /// completes fully, no further read_events should trigger on the same peer! /// @@ -58,7 +59,7 @@ pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone { /// events should be paused to prevent DoS in the send buffer), resume_read may be set /// 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: &Vec, write_offset: usize, resume_read: bool) -> usize; + 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_event, read_event or disconnect_event may be made with this descriptor. /// No disconnect_event should be generated as a result of this call, though obviously races @@ -387,7 +388,8 @@ impl PeerManager { }; let should_be_reading = peer.pending_outbound_buffer.len() < MSG_BUFF_SIZE; - let data_sent = descriptor.send_data(next_buff, peer.pending_outbound_buffer_first_msg_offset, should_be_reading); + let pending = &next_buff[peer.pending_outbound_buffer_first_msg_offset..]; + let data_sent = descriptor.send_data(pending, should_be_reading); peer.pending_outbound_buffer_first_msg_offset += data_sent; if peer.pending_outbound_buffer_first_msg_offset == next_buff.len() { true } else { false } } { @@ -479,26 +481,21 @@ impl PeerManager { match $thing { Ok(x) => x, Err(e) => { - if let Some(action) = e.action { - match action { - msgs::ErrorAction::DisconnectPeer { msg: _ } => { - //TODO: Try to push msg - log_trace!(self, "Got Err handling message, disconnecting peer because {}", e.err); - return Err(PeerHandleError{ no_connection_possible: false }); - }, - msgs::ErrorAction::IgnoreError => { - log_trace!(self, "Got Err handling message, ignoring because {}", e.err); - continue; - }, - msgs::ErrorAction::SendErrorMessage { msg } => { - log_trace!(self, "Got Err handling message, sending Error message because {}", e.err); - encode_and_send_msg!(msg, 17); - continue; - }, - } - } else { - log_debug!(self, "Got Err handling message, action not yet filled in: {}", e.err); - return Err(PeerHandleError{ no_connection_possible: false }); + match e.action { + msgs::ErrorAction::DisconnectPeer { msg: _ } => { + //TODO: Try to push msg + log_trace!(self, "Got Err handling message, disconnecting peer because {}", e.err); + return Err(PeerHandleError{ no_connection_possible: false }); + }, + msgs::ErrorAction::IgnoreError => { + log_trace!(self, "Got Err handling message, ignoring because {}", e.err); + continue; + }, + msgs::ErrorAction::SendErrorMessage { msg } => { + log_trace!(self, "Got Err handling message, sending Error message because {}", e.err); + encode_and_send_msg!(msg, 17); + continue; + }, } } }; @@ -1018,42 +1015,38 @@ impl PeerManager { self.message_handler.route_handler.handle_htlc_fail_channel_update(update); }, MessageSendEvent::HandleError { ref node_id, ref action } => { - if let Some(ref action) = *action { - match *action { - msgs::ErrorAction::DisconnectPeer { ref msg } => { - if let Some(mut descriptor) = peers.node_id_to_descriptor.remove(node_id) { - peers.peers_needing_send.remove(&descriptor); - if let Some(mut peer) = peers.peers.remove(&descriptor) { - if let Some(ref msg) = *msg { - 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))); - // 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); - } else { - log_trace!(self, "Handling DisconnectPeer HandleError event in peer_handler for node {} with no message", log_pubkey!(node_id)); - } + match *action { + msgs::ErrorAction::DisconnectPeer { ref msg } => { + if let Some(mut descriptor) = peers.node_id_to_descriptor.remove(node_id) { + peers.peers_needing_send.remove(&descriptor); + if let Some(mut peer) = peers.peers.remove(&descriptor) { + if let Some(ref msg) = *msg { + 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))); + // 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); + } else { + log_trace!(self, "Handling DisconnectPeer HandleError event in peer_handler for node {} with no message", log_pubkey!(node_id)); } - descriptor.disconnect_socket(); - self.message_handler.chan_handler.peer_disconnected(&node_id, false); } - }, - msgs::ErrorAction::IgnoreError => {}, - msgs::ErrorAction::SendErrorMessage { ref msg } => { - log_trace!(self, "Handling SendErrorMessage HandleError event in peer_handler for node {} with message {}", - log_pubkey!(node_id), - msg.data); - 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))); - self.do_attempt_write_data(&mut descriptor, peer); - }, - } - } else { - log_error!(self, "Got no-action HandleError Event in peer_handler for node {}, no such events should ever be generated!", log_pubkey!(node_id)); + descriptor.disconnect_socket(); + self.message_handler.chan_handler.peer_disconnected(&node_id, false); + } + }, + msgs::ErrorAction::IgnoreError => {}, + msgs::ErrorAction::SendErrorMessage { ref msg } => { + log_trace!(self, "Handling SendErrorMessage HandleError event in peer_handler for node {} with message {}", + log_pubkey!(node_id), + msg.data); + 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))); + self.do_attempt_write_data(&mut descriptor, peer); + }, } } } @@ -1118,9 +1111,8 @@ mod tests { } impl SocketDescriptor for FileDescriptor { - fn send_data(&mut self, data: &Vec, write_offset: usize, _resume_read: bool) -> usize { - assert!(write_offset < data.len()); - data.len() - write_offset + fn send_data(&mut self, data: &[u8], _resume_read: bool) -> usize { + data.len() } fn disconnect_socket(&mut self) {} @@ -1171,7 +1163,7 @@ mod tests { let chan_handler = test_utils::TestChannelMessageHandler::new(); chan_handler.pending_events.lock().unwrap().push(events::MessageSendEvent::HandleError { node_id: their_id, - action: Some(msgs::ErrorAction::DisconnectPeer { msg: None }), + 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);