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

use std::collections::VecDeque;

use crate::{
    messages::channels::receivers::reliable_message_receiver::{
        ReceiverArranger, ReliableMessageReceiver,
    },
    types::MessageIndex,
    MessageContainer,
};

// OrderedReliableReceiver
pub type OrderedReliableReceiver = ReliableMessageReceiver<OrderedArranger>;

impl OrderedReliableReceiver {
    pub fn new() -> Self {
        Self::with_arranger(OrderedArranger {
            oldest_received_message_index: 0,
            buffer: VecDeque::new(),
        })
    }
}

// OrderedArranger
pub struct OrderedArranger {
    buffer: VecDeque<(MessageIndex, Option<MessageContainer>)>,
    oldest_received_message_index: MessageIndex,
}

impl ReceiverArranger for OrderedArranger {
    fn process(
        &mut self,
        incoming_messages: &mut Vec<(MessageIndex, MessageContainer)>,
        message_index: MessageIndex,
        message: MessageContainer,
    ) {
        let mut current_index = 0;

        // Put message where it needs to go in buffer
        loop {
            if current_index < self.buffer.len() {
                if let Some((old_message_index, old_message)) = self.buffer.get_mut(current_index) {
                    if *old_message_index == message_index {
                        if old_message.is_none() {
                            *old_message = Some(message);
                            break;
                        }
                    }
                }
            } else {
                let next_message_index = self
                    .oldest_received_message_index
                    .wrapping_add(current_index as u16);

                if next_message_index == message_index {
                    self.buffer.push_back((next_message_index, Some(message)));
                    break;
                } else {
                    self.buffer.push_back((next_message_index, None));
                    // keep filling up buffer
                }
            }

            current_index += 1;
        }

        // Pop messages out in order
        loop {
            let Some((_, Some(_))) = self.buffer.front() else {
                // no more messages, return
                return;
            };
            let Some((index, Some(message))) = self.buffer.pop_front() else {
                panic!("shouldn't be possible due to above check");
            };

            incoming_messages.push((index, message));
            self.oldest_received_message_index = self.oldest_received_message_index.wrapping_add(1);
        }
    }
}