hopr_protocol_session/utils/
mod.rs

1pub mod skip_queue;
2
3#[cfg(test)]
4pub mod test;
5
6use std::{
7    cmp::Ordering,
8    sync::Arc,
9    time::{Duration, Instant},
10};
11
12use ringbuffer::{AllocRingBuffer, RingBuffer};
13
14use crate::{
15    errors::SessionError,
16    protocol::{FrameId, Segment, SeqIndicator, SeqNum},
17};
18
19#[derive(Clone)]
20pub(crate) struct RingBufferProducer<T>(Arc<parking_lot::FairMutex<AllocRingBuffer<T>>>);
21
22impl<T> std::fmt::Debug for RingBufferProducer<T> {
23    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24        f.debug_tuple("RingBufferProducer").finish()
25    }
26}
27
28impl<T> RingBufferProducer<T> {
29    pub fn push(&mut self, item: T) {
30        self.0.lock().enqueue(item);
31    }
32}
33
34#[derive(Debug)]
35pub(crate) struct RingBufferView<T>(Arc<parking_lot::FairMutex<AllocRingBuffer<T>>>);
36
37impl<T> Clone for RingBufferView<T> {
38    fn clone(&self) -> Self {
39        Self(self.0.clone())
40    }
41}
42
43impl<T: Clone> RingBufferView<T> {
44    pub fn find<F: FnMut(&T) -> bool>(&self, mut predicate: F) -> Vec<T> {
45        self.0.lock().iter().filter(|item| predicate(item)).cloned().collect()
46    }
47}
48
49pub(crate) fn searchable_ringbuffer<T: Send + 'static>(capacity: usize) -> (RingBufferProducer<T>, RingBufferView<T>) {
50    let rb = Arc::new(parking_lot::FairMutex::new(AllocRingBuffer::new(capacity)));
51    (RingBufferProducer(rb.clone()), RingBufferView(rb))
52}
53
54const MAX_BACKOFF: Duration = Duration::from_secs(300);
55
56pub(crate) fn next_deadline_with_backoff(n: usize, base: f64, duration: Duration) -> Instant {
57    let backoff = duration.mul_f64(base.powi(n.min((i32::MAX / 2) as usize) as i32 + 1));
58    Instant::now() + backoff.min(MAX_BACKOFF)
59}
60
61#[derive(Debug, Copy, Clone, Eq)]
62pub(crate) struct RetriedFrameId {
63    pub frame_id: FrameId,
64    pub retry_count: usize,
65    max_retries: usize,
66}
67
68impl RetriedFrameId {
69    pub fn no_retries(frame_id: FrameId) -> Self {
70        Self {
71            frame_id,
72            retry_count: 1,
73            max_retries: 1,
74        }
75    }
76
77    pub fn with_retries(frame_id: FrameId, max_retries: usize) -> Self {
78        Self {
79            frame_id,
80            retry_count: 1,
81            max_retries,
82        }
83    }
84
85    pub fn next(self) -> Option<Self> {
86        if self.retry_count < self.max_retries {
87            Some(Self {
88                frame_id: self.frame_id,
89                retry_count: self.retry_count + 1,
90                max_retries: self.max_retries,
91            })
92        } else {
93            None
94        }
95    }
96}
97
98impl PartialEq<Self> for RetriedFrameId {
99    fn eq(&self, other: &Self) -> bool {
100        self.frame_id.eq(&other.frame_id)
101    }
102}
103
104impl PartialOrd<Self> for RetriedFrameId {
105    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
106        Some(self.cmp(other))
107    }
108}
109
110impl Ord for RetriedFrameId {
111    fn cmp(&self, other: &Self) -> Ordering {
112        self.frame_id.cmp(&other.frame_id)
113    }
114}
115
116/// Helper function to segment `data` into segments of a given ` max_segment_size ` length.
117/// All segments are tagged with the same `frame_id` and output into the given ` segments ` buffer.
118pub fn segment_into<T: AsRef<[u8]>, E: Extend<Segment>>(
119    data: T,
120    max_segment_size: usize,
121    frame_id: FrameId,
122    segments: &mut E,
123) -> crate::errors::Result<()> {
124    if frame_id == 0 {
125        return Err(SessionError::InvalidFrameId);
126    }
127
128    if max_segment_size == 0 {
129        return Err(SessionError::IncorrectMessageLength);
130    }
131
132    let data = data.as_ref();
133
134    let num_chunks = data.len().div_ceil(max_segment_size);
135    if num_chunks > SeqNum::MAX as usize {
136        return Err(SessionError::DataTooLong);
137    }
138
139    let chunks = data.chunks(max_segment_size);
140
141    let seq_len = SeqIndicator::try_from(chunks.len() as SeqNum)?;
142    segments.extend(chunks.enumerate().map(|(idx, data)| Segment {
143        frame_id,
144        seq_flags: seq_len,
145        seq_idx: idx as u8,
146        data: data.into(),
147    }));
148
149    Ok(())
150}
151
152/// Convenience wrapper for [`segment_into`] that allocates its own output buffer and returns it.
153#[allow(unused)]
154pub fn segment<T: AsRef<[u8]>>(data: T, max_segment_size: usize, frame_id: u32) -> crate::errors::Result<Vec<Segment>> {
155    let mut out = Vec::with_capacity(data.as_ref().len().div_ceil(max_segment_size));
156    segment_into(data, max_segment_size, frame_id, &mut out)?;
157    Ok(out)
158}
159
160#[cfg(test)]
161mod tests {
162    use hex_literal::hex;
163
164    use super::*;
165
166    #[test]
167    fn segment_should_split_data_correctly() -> anyhow::Result<()> {
168        let data = hex!("deadbeefcafebabe");
169
170        let segments = segment(data, 3, 1)?;
171        assert_eq!(3, segments.len());
172
173        assert_eq!(hex!("deadbe"), segments[0].data.as_ref());
174        assert_eq!(0, segments[0].seq_idx);
175        assert_eq!(3, segments[0].seq_flags.seq_len());
176        assert_eq!(1, segments[0].frame_id);
177
178        assert_eq!(hex!("efcafe"), segments[1].data.as_ref());
179        assert_eq!(1, segments[1].seq_idx);
180        assert_eq!(3, segments[1].seq_flags.seq_len());
181        assert_eq!(1, segments[1].frame_id);
182
183        assert_eq!(hex!("babe"), segments[2].data.as_ref());
184        assert_eq!(2, segments[2].seq_idx);
185        assert_eq!(3, segments[2].seq_flags.seq_len());
186        assert_eq!(1, segments[2].frame_id);
187
188        Ok(())
189    }
190}
hopr_protocol_session/utils/mod.rs

hopr_protocol_session/utils/
mod.rs
