X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fpeer_handler.rs;h=a6992eae4ff1fcf62dab20875407e2741dcebec5;hb=f7dd69240afcd3a0087f7796824a9e2697ed85cf;hp=3a0d6b66d3578135dbe050cb936d74893b74215c;hpb=ac3bf5b71d80d0dbdbacfbac1fc9528d09391237;p=rust-lightning diff --git a/src/ln/peer_handler.rs b/src/ln/peer_handler.rs index 3a0d6b66..a6992eae 100644 --- a/src/ln/peer_handler.rs +++ b/src/ln/peer_handler.rs @@ -219,7 +219,6 @@ impl PeerManager { } fn do_read_event(&self, peer_descriptor: &mut Descriptor, data: Vec) -> Result { - let mut upstream_events = Vec::new(); let pause_read = { let mut peers = self.peers.lock().unwrap(); let (should_insert_node_id, pause_read) = match peers.peers.get_mut(peer_descriptor) { @@ -357,11 +356,20 @@ impl PeerManager { 38 => { let msg = try_potential_decodeerror!(msgs::Shutdown::decode(&msg_data[2..])); - try_potential_handleerror!(self.message_handler.chan_handler.handle_shutdown(&peer.their_node_id.unwrap(), &msg)); + let resp_options = try_potential_handleerror!(self.message_handler.chan_handler.handle_shutdown(&peer.their_node_id.unwrap(), &msg)); + if let Some(resp) = resp_options.0 { + encode_and_send_msg!(resp, 38); + } + if let Some(resp) = resp_options.1 { + encode_and_send_msg!(resp, 39); + } }, 39 => { let msg = try_potential_decodeerror!(msgs::ClosingSigned::decode(&msg_data[2..])); - try_potential_handleerror!(self.message_handler.chan_handler.handle_closing_signed(&peer.their_node_id.unwrap(), &msg)); + let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_closing_signed(&peer.their_node_id.unwrap(), &msg)); + if let Some(resp) = resp_option { + encode_and_send_msg!(resp, 39); + } }, 128 => { @@ -370,42 +378,15 @@ impl PeerManager { }, 130 => { let msg = try_potential_decodeerror!(msgs::UpdateFulfillHTLC::decode(&msg_data[2..])); - let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fulfill_htlc(&peer.their_node_id.unwrap(), &msg)); - match resp_option { - Some(resps) => { - for resp in resps.0 { - encode_and_send_msg!(resp, 128); - } - encode_and_send_msg!(resps.1, 132); - }, - None => {}, - } + try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fulfill_htlc(&peer.their_node_id.unwrap(), &msg)); }, 131 => { let msg = try_potential_decodeerror!(msgs::UpdateFailHTLC::decode(&msg_data[2..])); - let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg)); - match resp_option { - Some(resps) => { - for resp in resps.0 { - encode_and_send_msg!(resp, 128); - } - encode_and_send_msg!(resps.1, 132); - }, - None => {}, - } + try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg)); }, 135 => { let msg = try_potential_decodeerror!(msgs::UpdateFailMalformedHTLC::decode(&msg_data[2..])); - let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_malformed_htlc(&peer.their_node_id.unwrap(), &msg)); - match resp_option { - Some(resps) => { - for resp in resps.0 { - encode_and_send_msg!(resp, 128); - } - encode_and_send_msg!(resps.1, 132); - }, - None => {}, - } + try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_malformed_htlc(&peer.their_node_id.unwrap(), &msg)); }, 132 => { @@ -415,9 +396,17 @@ impl PeerManager { }, 133 => { let msg = try_potential_decodeerror!(msgs::RevokeAndACK::decode(&msg_data[2..])); - try_potential_handleerror!(self.message_handler.chan_handler.handle_revoke_and_ack(&peer.their_node_id.unwrap(), &msg)); + let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_revoke_and_ack(&peer.their_node_id.unwrap(), &msg)); + match resp_option { + Some(resps) => { + for resp in resps.0 { + encode_and_send_msg!(resp, 128); + } + encode_and_send_msg!(resps.1, 132); + }, + None => {}, + } }, - 134 => { let msg = try_potential_decodeerror!(msgs::UpdateFee::decode(&msg_data[2..])); try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fee(&peer.their_node_id.unwrap(), &msg)); @@ -473,11 +462,25 @@ impl PeerManager { None => {} }; + pause_read + }; + + self.process_events(); + + Ok(pause_read) + } + + /// Checks for any events generated by our handlers and processes them. May be needed after eg + /// calls to ChannelManager::process_pending_htlc_forward. + pub fn process_events(&self) { + let mut upstream_events = Vec::new(); + { // TODO: There are some DoS attacks here where you can flood someone's outbound send // buffer by doing things like announcing channels on another node. We should be willing to // drop optional-ish messages when send buffers get full! let mut events_generated = self.message_handler.chan_handler.get_and_clear_pending_events(); + let mut peers = self.peers.lock().unwrap(); for event in events_generated.drain(..) { macro_rules! get_peer_for_forwarding { ($node_id: expr, $handle_no_such_peer: block) => { @@ -502,6 +505,8 @@ impl PeerManager { Event::FundingGenerationReady {..} => { /* Hand upstream */ }, Event::FundingBroadcastSafe {..} => { /* Hand upstream */ }, Event::PaymentReceived {..} => { /* Hand upstream */ }, + Event::PaymentSent {..} => { /* Hand upstream */ }, + Event::PaymentFailed {..} => { /* Hand upstream */ }, Event::PendingHTLCsForwardable {..} => { //TODO: Handle upstream in some confused form so that upstream just knows @@ -547,6 +552,14 @@ impl PeerManager { Self::do_attempt_write_data(&mut descriptor, peer); continue; }, + Event::SendFailHTLC { 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, 131))); + Self::do_attempt_write_data(&mut descriptor, peer); + continue; + }, Event::BroadcastChannelAnnouncement { ref msg, ref update_msg } => { let encoded_msg = encode_msg!(msg, 256); let encoded_update_msg = encode_msg!(update_msg, 258); @@ -573,16 +586,12 @@ impl PeerManager { upstream_events.push(event); } - - pause_read - }; + } let mut pending_events = self.pending_events.lock().unwrap(); for event in upstream_events.drain(..) { pending_events.push(event); } - - Ok(pause_read) } /// Indicates that the given socket descriptor's connection is now closed.