+ }
+ }
+
+ #[cfg(test)]
+ pub(super) fn release_pending_msgs(&self) -> HashMap<PublicKey, VecDeque<msgs::OnionMessage>> {
+ let mut pending_msgs = self.pending_messages.lock().unwrap();
+ let mut msgs = HashMap::new();
+ // We don't want to disconnect the peers by removing them entirely from the original map, so we
+ // swap the pending message buffers individually.
+ for (peer_node_id, pending_messages) in &mut *pending_msgs {
+ msgs.insert(*peer_node_id, core::mem::take(pending_messages));
+ }
+ msgs
+ }
+}
+
+fn outbound_buffer_full(peer_node_id: &PublicKey, buffer: &HashMap<PublicKey, VecDeque<msgs::OnionMessage>>) -> bool {
+ const MAX_TOTAL_BUFFER_SIZE: usize = (1 << 20) * 128;
+ const MAX_PER_PEER_BUFFER_SIZE: usize = (1 << 10) * 256;
+ let mut total_buffered_bytes = 0;
+ let mut peer_buffered_bytes = 0;
+ for (pk, peer_buf) in buffer {
+ for om in peer_buf {
+ let om_len = om.serialized_length();
+ if pk == peer_node_id {
+ peer_buffered_bytes += om_len;
+ }
+ total_buffered_bytes += om_len;
+
+ if total_buffered_bytes >= MAX_TOTAL_BUFFER_SIZE ||
+ peer_buffered_bytes >= MAX_PER_PEER_BUFFER_SIZE
+ {
+ return true
+ }
+ }