bitfold_peer/
command_queue.rs

1use std::collections::VecDeque;
2
3use bitfold_protocol::command::ProtocolCommand;
4
5/// Command queue for batching protocol commands before transmission.
6/// Commands are aggregated into larger packets to improve bandwidth utilization.
7#[derive(Debug)]
8pub struct CommandQueue {
9    /// Pending commands to be processed
10    commands: VecDeque<ProtocolCommand>,
11    /// Maximum commands to queue before forcing a flush
12    max_queue_size: usize,
13}
14
15impl CommandQueue {
16    /// Creates a new command queue with the specified capacity.
17    pub fn new(capacity: usize) -> Self {
18        Self { commands: VecDeque::with_capacity(capacity), max_queue_size: capacity }
19    }
20
21    /// Enqueues a protocol command for later processing.
22    /// Returns true if the queue should be flushed (reached max size).
23    pub fn enqueue(&mut self, command: ProtocolCommand) -> bool {
24        self.commands.push_back(command);
25        self.commands.len() >= self.max_queue_size
26    }
27
28    /// Returns the number of queued commands.
29    pub fn len(&self) -> usize {
30        self.commands.len()
31    }
32
33    /// Returns true if the queue is empty.
34    pub fn is_empty(&self) -> bool {
35        self.commands.is_empty()
36    }
37
38    /// Drains all commands from the queue for processing.
39    pub fn drain(&mut self) -> impl Iterator<Item = ProtocolCommand> + '_ {
40        self.commands.drain(..)
41    }
42
43    /// Returns an iterator over the commands without draining.
44    pub fn iter(&self) -> impl Iterator<Item = &ProtocolCommand> {
45        self.commands.iter()
46    }
47
48    /// Clears all pending commands.
49    pub fn clear(&mut self) {
50        self.commands.clear();
51    }
52}
53
54impl Default for CommandQueue {
55    fn default() -> Self {
56        Self::new(256)
57    }
58}
59
60#[cfg(test)]
61mod tests {
62    use super::*;
63
64    #[test]
65    fn test_queue_basic_operations() {
66        let mut queue = CommandQueue::new(3);
67
68        assert!(queue.is_empty());
69        assert_eq!(queue.len(), 0);
70
71        queue.enqueue(ProtocolCommand::Ping { timestamp: 100 });
72        assert_eq!(queue.len(), 1);
73        assert!(!queue.is_empty());
74    }
75
76    #[test]
77    fn test_queue_max_size_trigger() {
78        let mut queue = CommandQueue::new(2);
79
80        assert!(!queue.enqueue(ProtocolCommand::Ping { timestamp: 100 }));
81        assert!(queue.enqueue(ProtocolCommand::Ping { timestamp: 200 })); // Should trigger flush
82
83        assert_eq!(queue.len(), 2);
84    }
85
86    #[test]
87    fn test_queue_drain() {
88        let mut queue = CommandQueue::new(10);
89
90        queue.enqueue(ProtocolCommand::Ping { timestamp: 100 });
91        queue.enqueue(ProtocolCommand::Disconnect { reason: 0 });
92
93        let commands: Vec<_> = queue.drain().collect();
94        assert_eq!(commands.len(), 2);
95        assert!(queue.is_empty());
96    }
97
98    #[test]
99    fn test_queue_command_aggregation() {
100        let mut queue = CommandQueue::new(10);
101
102        // Enqueue multiple protocol commands
103        queue.enqueue(ProtocolCommand::SendReliable {
104            channel_id: 0,
105            sequence: 1,
106            ordered: true,
107            data: vec![1, 2, 3].into(),
108        });
109        queue.enqueue(ProtocolCommand::Acknowledge {
110            sequence: 5,
111            received_mask: 0xFF,
112            sent_time: Some(1000),
113        });
114        queue.enqueue(ProtocolCommand::Ping { timestamp: 2000 });
115
116        assert_eq!(queue.len(), 3);
117
118        // Drain and verify all commands
119        let commands: Vec<_> = queue.drain().collect();
120        assert_eq!(commands.len(), 3);
121        assert!(queue.is_empty());
122    }
123
124    #[test]
125    fn test_queue_iter() {
126        let mut queue = CommandQueue::new(10);
127
128        queue.enqueue(ProtocolCommand::Ping { timestamp: 100 });
129        queue.enqueue(ProtocolCommand::Pong { timestamp: 200 });
130
131        let count = queue.iter().count();
132        assert_eq!(count, 2);
133        assert_eq!(queue.len(), 2); // Iterator shouldn't drain
134    }
135}
bitfold_peer/command_queue.rs

bitfold_peer/
command_queue.rs
