- }
- for event in events_generated.drain(..) {
- match event {
- MessageSendEvent::SendAcceptChannel { ref node_id, ref msg } => {
- log_debug!(self.logger, "Handling SendAcceptChannel event in peer_handler for node {} for channel {}",
- log_pubkey!(node_id),
- log_bytes!(msg.temporary_channel_id));
- self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
- },
- MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => {
- log_debug!(self.logger, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
- log_pubkey!(node_id),
- log_bytes!(msg.temporary_channel_id));
- self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
- },
- MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => {
- log_debug!(self.logger, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
- log_pubkey!(node_id),
- log_bytes!(msg.temporary_channel_id),
- 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
- self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
- },
- MessageSendEvent::SendFundingSigned { ref node_id, ref msg } => {
- log_debug!(self.logger, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
- log_pubkey!(node_id),
- log_bytes!(msg.channel_id));
- self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
- },
- MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => {
- log_debug!(self.logger, "Handling SendFundingLocked event in peer_handler for node {} for channel {}",
- log_pubkey!(node_id),
- log_bytes!(msg.channel_id));
- self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
- },
- MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => {
- log_debug!(self.logger, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
- log_pubkey!(node_id),
- log_bytes!(msg.channel_id));
- self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), 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 } } => {
- log_debug!(self.logger, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
- log_pubkey!(node_id),
- update_add_htlcs.len(),
- update_fulfill_htlcs.len(),
- update_fail_htlcs.len(),
- log_bytes!(commitment_signed.channel_id));
- let mut peer = get_peer_for_forwarding!(node_id);
- for msg in update_add_htlcs {
- self.enqueue_message(&mut *peer, msg);
- }
- for msg in update_fulfill_htlcs {
- self.enqueue_message(&mut *peer, msg);
- }
- for msg in update_fail_htlcs {
- self.enqueue_message(&mut *peer, msg);
- }
- for msg in update_fail_malformed_htlcs {
- self.enqueue_message(&mut *peer, msg);
- }
- if let &Some(ref msg) = update_fee {
- self.enqueue_message(&mut *peer, msg);
+ for event in events_generated.drain(..) {
+ match event {
+ MessageSendEvent::SendAcceptChannel { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendAcceptChannel event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.temporary_channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendAcceptChannelV2 { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendAcceptChannelV2 event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.temporary_channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.temporary_channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendOpenChannelV2 { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendOpenChannelV2 event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.temporary_channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
+ log_pubkey!(node_id),
+ log_bytes!(msg.temporary_channel_id),
+ 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
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendFundingSigned { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendChannelReady { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendChannelReady event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendTxAddInput { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendTxAddInput event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendTxAddOutput { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendTxAddOutput event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendTxRemoveInput { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendTxRemoveInput event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendTxRemoveOutput { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendTxRemoveOutput event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendTxComplete { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendTxComplete event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendTxSignatures { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendTxSignatures event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendTxInitRbf { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendTxInitRbf event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendTxAckRbf { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendTxAckRbf event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendTxAbort { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendTxAbort event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), 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 } } => {
+ log_debug!(self.logger, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
+ log_pubkey!(node_id),
+ update_add_htlcs.len(),
+ update_fulfill_htlcs.len(),
+ update_fail_htlcs.len(),
+ log_bytes!(commitment_signed.channel_id));
+ let mut peer = get_peer_for_forwarding!(node_id);
+ for msg in update_add_htlcs {
+ self.enqueue_message(&mut *peer, msg);
+ }
+ for msg in update_fulfill_htlcs {
+ self.enqueue_message(&mut *peer, msg);
+ }
+ for msg in update_fail_htlcs {
+ self.enqueue_message(&mut *peer, msg);
+ }
+ for msg in update_fail_malformed_htlcs {
+ self.enqueue_message(&mut *peer, msg);
+ }
+ if let &Some(ref msg) = update_fee {
+ self.enqueue_message(&mut *peer, msg);
+ }
+ self.enqueue_message(&mut *peer, commitment_signed);
+ },
+ MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendClosingSigned event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendShutdown { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling Shutdown event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
+ log_debug!(self.logger, "Handling SendChannelReestablish event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendChannelAnnouncement { ref node_id, ref msg, ref update_msg } => {
+ log_debug!(self.logger, "Handling SendChannelAnnouncement event in peer_handler for node {} for short channel id {}",
+ log_pubkey!(node_id),
+ msg.contents.short_channel_id);
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), update_msg);
+ },
+ MessageSendEvent::BroadcastChannelAnnouncement { msg, update_msg } => {
+ log_debug!(self.logger, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
+ match self.message_handler.route_handler.handle_channel_announcement(&msg) {
+ Ok(_) | Err(LightningError { action: msgs::ErrorAction::IgnoreDuplicateGossip, .. }) =>
+ self.forward_broadcast_msg(peers, &wire::Message::ChannelAnnouncement(msg), None),
+ _ => {},
+ }
+ if let Some(msg) = update_msg {
+ match self.message_handler.route_handler.handle_channel_update(&msg) {
+ Ok(_) | Err(LightningError { action: msgs::ErrorAction::IgnoreDuplicateGossip, .. }) =>
+ self.forward_broadcast_msg(peers, &wire::Message::ChannelUpdate(msg), None),
+ _ => {},
+ }
+ }
+ },
+ MessageSendEvent::BroadcastChannelUpdate { msg } => {
+ log_debug!(self.logger, "Handling BroadcastChannelUpdate event in peer_handler for short channel id {}", msg.contents.short_channel_id);
+ match self.message_handler.route_handler.handle_channel_update(&msg) {
+ Ok(_) | Err(LightningError { action: msgs::ErrorAction::IgnoreDuplicateGossip, .. }) =>
+ self.forward_broadcast_msg(peers, &wire::Message::ChannelUpdate(msg), None),
+ _ => {},
+ }
+ },
+ MessageSendEvent::BroadcastNodeAnnouncement { msg } => {
+ log_debug!(self.logger, "Handling BroadcastNodeAnnouncement event in peer_handler for node {}", msg.contents.node_id);
+ match self.message_handler.route_handler.handle_node_announcement(&msg) {
+ Ok(_) | Err(LightningError { action: msgs::ErrorAction::IgnoreDuplicateGossip, .. }) =>
+ self.forward_broadcast_msg(peers, &wire::Message::NodeAnnouncement(msg), None),
+ _ => {},
+ }
+ },
+ MessageSendEvent::SendChannelUpdate { ref node_id, ref msg } => {
+ log_trace!(self.logger, "Handling SendChannelUpdate event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id), msg.contents.short_channel_id);
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::HandleError { node_id, action } => {
+ match action {
+ msgs::ErrorAction::DisconnectPeer { msg } => {
+ if let Some(msg) = msg.as_ref() {
+ log_trace!(self.logger, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}",
+ log_pubkey!(node_id), msg.data);
+ } else {
+ log_trace!(self.logger, "Handling DisconnectPeer HandleError event in peer_handler for node {}",
+ log_pubkey!(node_id));
+ }
+ // We do not have the peers write lock, so we just store that we're
+ // about to disconenct the peer and do it after we finish
+ // processing most messages.
+ let msg = msg.map(|msg| wire::Message::<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>::Error(msg));
+ peers_to_disconnect.insert(node_id, msg);
+ },
+ msgs::ErrorAction::DisconnectPeerWithWarning { msg } => {
+ log_trace!(self.logger, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}",
+ log_pubkey!(node_id), msg.data);
+ // We do not have the peers write lock, so we just store that we're
+ // about to disconenct the peer and do it after we finish
+ // processing most messages.
+ peers_to_disconnect.insert(node_id, Some(wire::Message::Warning(msg)));
+ },
+ msgs::ErrorAction::IgnoreAndLog(level) => {
+ log_given_level!(self.logger, level, "Received a HandleError event to be ignored for node {}", log_pubkey!(node_id));
+ },
+ msgs::ErrorAction::IgnoreDuplicateGossip => {},
+ msgs::ErrorAction::IgnoreError => {
+ log_debug!(self.logger, "Received a HandleError event to be ignored for node {}", log_pubkey!(node_id));
+ },
+ msgs::ErrorAction::SendErrorMessage { ref msg } => {
+ log_trace!(self.logger, "Handling SendErrorMessage HandleError event in peer_handler for node {} with message {}",
+ log_pubkey!(node_id),
+ msg.data);
+ self.enqueue_message(&mut *get_peer_for_forwarding!(&node_id), msg);
+ },
+ msgs::ErrorAction::SendWarningMessage { ref msg, ref log_level } => {
+ log_given_level!(self.logger, *log_level, "Handling SendWarningMessage HandleError event in peer_handler for node {} with message {}",
+ log_pubkey!(node_id),
+ msg.data);
+ self.enqueue_message(&mut *get_peer_for_forwarding!(&node_id), msg);
+ },
+ }
+ },
+ MessageSendEvent::SendChannelRangeQuery { ref node_id, ref msg } => {
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);
+ },
+ MessageSendEvent::SendShortIdsQuery { ref node_id, ref msg } => {
+ self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg);