uflow 0.7.1

Provides ordered, mixed-reliability, and congestion-controlled data transfer over UDP
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132


use crate::frame;
use crate::MAX_FRAGMENT_SIZE;

use std::cell::RefCell;
use std::rc::Rc;
use std::rc::Weak;

#[derive(Debug)]
pub struct PendingPacket {
    data: Box<[u8]>,
    channel_id: u8,

    sequence_id: u32,
    window_parent_lead: u16,
    channel_parent_lead: u16,
    last_fragment_id: u16,

    ack_flags: Box<[u64]>,
}

impl PendingPacket {
    pub fn new(data: Box<[u8]>, channel_id: u8,
               sequence_id: u32, window_parent_lead: u16, channel_parent_lead: u16) -> Self {
        let num_fragments = (data.len() + MAX_FRAGMENT_SIZE - 1) / MAX_FRAGMENT_SIZE + (data.len() == 0) as usize;
        debug_assert!(num_fragments != 0);

        debug_assert!(num_fragments - 1 <= u16::MAX as usize);
        let last_fragment_id = (num_fragments - 1) as u16;

        Self {
            data,
            channel_id,

            sequence_id,
            window_parent_lead,
            channel_parent_lead,
            last_fragment_id,

            ack_flags: vec![0u64; (num_fragments + 63)/64].into_boxed_slice(),
        }
    }

    #[cfg(test)]
    pub fn sequence_id(&self) -> u32 {
        self.sequence_id
    }

    #[cfg(test)]
    pub fn channel_id(&self) -> u8 {
        self.channel_id
    }

    #[cfg(test)]
    pub fn window_parent_lead(&self) -> u16 {
        self.window_parent_lead
    }

    #[cfg(test)]
    pub fn channel_parent_lead(&self) -> u16 {
        self.channel_parent_lead
    }

    pub fn last_fragment_id(&self) -> u16 {
        self.last_fragment_id
    }

    pub fn fragment_acknowledged(&self, fragment_id: u16) -> bool {
        let flag_bit = 1 << (fragment_id % 64) as u64;
        let flags_index = (fragment_id / 64) as usize;
        self.ack_flags[flags_index] & flag_bit != 0
    }

    pub fn acknowledge_fragment(&mut self, fragment_id: u16) {
        let flag_bit = 1 << (fragment_id % 64) as u64;
        let flags_index = (fragment_id / 64) as usize;
        self.ack_flags[flags_index] |= flag_bit;
    }

    pub fn size(&self) -> usize {
        self.data.len()
    }

    pub fn datagram<'a>(&'a self, fragment_id: u16) -> frame::DatagramRef<'a> {
        debug_assert!(fragment_id <= self.last_fragment_id);

        let i = fragment_id as usize;
        let data = if fragment_id == self.last_fragment_id {
            &self.data[i * MAX_FRAGMENT_SIZE .. ]
        } else {
            &self.data[i * MAX_FRAGMENT_SIZE .. (i + 1)*MAX_FRAGMENT_SIZE]
        };

        frame::DatagramRef {
            sequence_id: self.sequence_id,
            channel_id: self.channel_id,
            window_parent_lead: self.window_parent_lead,
            channel_parent_lead: self.channel_parent_lead,
            fragment_id,
            fragment_id_last: self.last_fragment_id,
            data,
        }
    }
}

pub type PendingPacketRc = Rc<RefCell<PendingPacket>>;
pub type PendingPacketWeak = Weak<RefCell<PendingPacket>>;

#[derive(Debug)]
pub struct FragmentRef {
    pub packet: PendingPacketWeak,
    pub fragment_id: u16,
}

impl Clone for FragmentRef {
    fn clone(&self) -> Self {
        Self {
            packet: Weak::clone(&self.packet),
            fragment_id: self.fragment_id,
        }
    }
}

impl FragmentRef {
    pub fn new(packet_rc: &PendingPacketRc, fragment_id: u16) -> Self {
        Self {
            packet: Rc::downgrade(packet_rc),
            fragment_id,
        }
    }
}