1use crate::sequence::{self, ReliableWindow};
2
3use std::{
4 collections::{HashMap, VecDeque},
5 net::SocketAddr,
6 sync::{
7 Arc,
8 atomic::{AtomicUsize, Ordering},
9 },
10};
11
12use crate::{datatype::*, error::*, fragment::FragmentQ, raknet_log_debug, utils::*};
13
14pub(crate) struct OutgoingFrames {
15 frames: Vec<FrameSetPacket>,
16 pub(crate) coalesced: bool,
17}
18
19impl OutgoingFrames {
20 #[cfg(test)]
21 pub(crate) fn new() -> Self {
22 Self::with_capacity(0)
23 }
24
25 fn with_capacity(capacity: usize) -> Self {
26 Self {
27 frames: Vec::with_capacity(capacity),
28 coalesced: false,
29 }
30 }
31}
32
33impl std::ops::Deref for OutgoingFrames {
34 type Target = Vec<FrameSetPacket>;
35 fn deref(&self) -> &Self::Target {
36 &self.frames
37 }
38}
39
40impl std::ops::DerefMut for OutgoingFrames {
41 fn deref_mut(&mut self) -> &mut Self::Target {
42 &mut self.frames
43 }
44}
45
46impl IntoIterator for OutgoingFrames {
47 type Item = FrameSetPacket;
48 type IntoIter = std::vec::IntoIter<FrameSetPacket>;
49 fn into_iter(self) -> Self::IntoIter {
50 self.frames.into_iter()
51 }
52}
53
54impl<'a> IntoIterator for &'a OutgoingFrames {
55 type Item = &'a FrameSetPacket;
56 type IntoIter = std::slice::Iter<'a, FrameSetPacket>;
57 fn into_iter(self) -> Self::IntoIter {
58 self.frames.iter()
59 }
60}
61
62#[derive(Clone, Copy, Debug, PartialEq, Eq)]
64#[repr(u8)]
65pub enum Reliability {
66 Unreliable = 0x00,
68 UnreliableSequenced = 0x01,
70 Reliable = 0x02,
72 ReliableOrdered = 0x03,
74 ReliableSequenced = 0x04,
76}
77
78impl Reliability {
79 pub fn to_u8(&self) -> u8 {
80 *self as u8
81 }
82
83 pub fn from(flags: u8) -> Result<Self> {
84 Self::try_from(flags)
85 }
86}
87
88impl TryFrom<u8> for Reliability {
89 type Error = RaknetError;
90
91 fn try_from(flags: u8) -> Result<Self> {
92 match flags {
93 0x00 => Ok(Self::Unreliable),
94 0x01 => Ok(Self::UnreliableSequenced),
95 0x02 => Ok(Self::Reliable),
96 0x03 => Ok(Self::ReliableOrdered),
97 0x04 => Ok(Self::ReliableSequenced),
98 _ => Err(RaknetError::IncorrectReliability),
99 }
100 }
101}
102
103const NEEDS_B_AND_AS_FLAG: u8 = 0x4;
104const CONTINUOUS_SEND_FLAG: u8 = 0x8;
105
106#[derive(Clone, Debug)]
109pub(crate) enum FramePayload {
110 Owned(Vec<u8>),
111 Bytes(bytes::Bytes),
112 Shared {
113 owner: Arc<[u8]>,
114 start: u32,
115 length: u16,
116 },
117}
118
119impl Default for FramePayload {
120 fn default() -> Self {
121 Self::Owned(Vec::new())
122 }
123}
124
125impl FramePayload {
126 fn copy_shared(data: &[u8]) -> Self {
127 Self::Shared {
128 owner: Arc::from(data),
129 start: 0,
130 length: data.len() as u16,
131 }
132 }
133}
134
135impl std::ops::Deref for FramePayload {
136 type Target = [u8];
137 fn deref(&self) -> &[u8] {
138 match self {
139 Self::Owned(data) => data,
140 Self::Bytes(data) => data,
141 Self::Shared {
142 owner,
143 start,
144 length,
145 } => &owner[*start as usize..*start as usize + usize::from(*length)],
146 }
147 }
148}
149
150impl AsRef<[u8]> for FramePayload {
151 fn as_ref(&self) -> &[u8] {
152 self
153 }
154}
155
156impl From<Vec<u8>> for FramePayload {
157 fn from(data: Vec<u8>) -> Self {
158 Self::Owned(data)
159 }
160}
161
162impl From<FramePayload> for Vec<u8> {
163 fn from(data: FramePayload) -> Self {
164 match data {
165 FramePayload::Owned(data) => data,
166 shared => shared.to_vec(),
167 }
168 }
169}
170
171impl PartialEq for FramePayload {
172 fn eq(&self, other: &Self) -> bool {
173 self.as_ref() == other.as_ref()
174 }
175}
176
177struct SharedPayload {
180 remaining: AtomicUsize,
181 bytes: usize,
182}
183
184#[derive(Clone)]
185pub struct FrameSetPacket {
186 pub sequence_number: u32,
187 pub flags: u8,
188 pub length_in_bytes: u16,
189 pub reliable_frame_index: u32,
190 pub sequenced_frame_index: u32,
191 pub ordered_frame_index: u32,
192 pub order_channel: u8,
193 pub compound_size: u32,
194 pub compound_id: u16,
195 pub fragment_index: u32,
196 pub(crate) data: FramePayload,
197 shared_payload: Option<Arc<SharedPayload>>,
198}
199
200impl FrameSetPacket {
201 pub fn new(r: Reliability, data: Vec<u8>) -> Self {
202 Self::from_payload(r, data.into())
203 }
204
205 fn from_payload(r: Reliability, data: FramePayload) -> Self {
206 let flag = r.to_u8() << 5;
207
208 Self {
209 sequence_number: 0,
210 flags: flag,
211 length_in_bytes: data.len() as u16,
212 reliable_frame_index: 0,
213 sequenced_frame_index: 0,
214 ordered_frame_index: 0,
215 order_channel: 0,
216 compound_size: 0,
217 compound_id: 0,
218 fragment_index: 0,
219 data,
220 shared_payload: None,
221 }
222 }
223
224 #[cfg(test)]
225 pub fn deserialize(buf: &[u8]) -> Result<(Self, bool)> {
226 let mut reader = RaknetReader::new(buf);
227 let input_len = buf.len() as u64;
228
229 let mut ret = Self {
230 sequence_number: 0,
231 flags: 0,
232 length_in_bytes: 0,
233 reliable_frame_index: 0,
234 sequenced_frame_index: 0,
235 ordered_frame_index: 0,
236 order_channel: 0,
237 compound_size: 0,
238 compound_id: 0,
239 fragment_index: 0,
240 data: FramePayload::default(),
241 shared_payload: None,
242 };
243
244 let packet_id = reader.read_u8()?;
245 if !(0x80..=0x8d).contains(&packet_id) {
246 return Err(RaknetError::PacketHeaderError);
247 }
248 ret.sequence_number = reader.read_u24(Endian::Little)?;
249 ret.flags = reader.read_u8()?;
250
251 let length_in_bits = reader.read_u16(Endian::Big)?;
252 if length_in_bits % 8 != 0 {
253 return Err(RaknetError::PacketParseError);
254 }
255 ret.length_in_bytes = length_in_bits / 8;
256
257 if ret.is_reliable()? {
258 ret.reliable_frame_index = reader.read_u24(Endian::Little)?;
259 }
260
261 if ret.is_sequenced()? {
262 ret.sequenced_frame_index = reader.read_u24(Endian::Little)?;
263 }
264 if ret.is_ordered()? {
265 ret.ordered_frame_index = reader.read_u24(Endian::Little)?;
266 ret.order_channel = reader.read_u8()?;
267 }
268
269 if ret.is_fragment() {
270 ret.compound_size = reader.read_u32(Endian::Big)?;
271 ret.compound_id = reader.read_u16(Endian::Big)?;
272 ret.fragment_index = reader.read_u32(Endian::Big)?;
273 }
274
275 let mut data = vec![0; usize::from(ret.length_in_bytes)];
276 reader.read(&mut data)?;
277 ret.data = data.into();
278
279 if ret.is_fragment() && (ret.compound_size == 0 || ret.fragment_index >= ret.compound_size)
280 {
281 return Err(RaknetError::PacketParseError);
282 }
283
284 Ok((ret, reader.pos() == input_len))
285 }
286
287 pub fn serialize(&self) -> Result<Vec<u8>> {
288 let mut buffer = Vec::new();
289 self.serialize_into(&mut buffer)?;
290 Ok(buffer)
291 }
292
293 pub(crate) fn serialize_into(&self, buffer: &mut Vec<u8>) -> Result<()> {
294 self.validate()?;
295 let size = self._size()?;
296 buffer.clear();
297 buffer.reserve(size);
298 let mut writer = RaknetWriter::from_buffer(std::mem::take(buffer));
299 let result = self
300 .write_header(&mut writer)
301 .and_then(|()| writer.write(self.data.as_ref()));
302 *buffer = writer.get_raw_payload();
303 result
304 }
305
306 pub(crate) fn serialize_group_into(frames: &[Self], output: &mut Vec<u8>) -> Result<()> {
307 let first = frames.first().ok_or(RaknetError::PacketParseError)?;
308 output.clear();
309 for (index, frame) in frames.iter().enumerate() {
310 if frame.sequence_number != first.sequence_number {
311 return Err(RaknetError::PacketParseError);
312 }
313 let mut header = [0; 32];
314 let length = frame.encode_header(&mut header)?;
315 output.extend_from_slice(&header[if index == 0 { 0 } else { 4 }..length]);
316 output.extend_from_slice(frame.data.as_ref());
317 }
318 Ok(())
319 }
320
321 pub(crate) fn validate(&self) -> Result<()> {
322 if self.data.len() != usize::from(self.length_in_bytes)
323 || self.data.len() > usize::from(u16::MAX / 8)
324 || (self.is_fragment()
325 && (self.compound_size == 0 || self.fragment_index >= self.compound_size))
326 {
327 return Err(RaknetError::PacketParseError);
328 }
329
330 self._size()?;
331 Ok(())
332 }
333
334 pub(crate) fn encode_header(&self, header: &mut [u8; 32]) -> Result<usize> {
335 self.validate()?;
336 let mut writer = RaknetWriter::from_buffer(header.as_mut_slice());
337 self.write_header(&mut writer)?;
338 Ok(32 - writer.remaining_mut())
339 }
340
341 fn write_header<B: bytes::BufMut>(&self, writer: &mut RaknetWriter<B>) -> Result<()> {
342 let mut id = 0x80 | NEEDS_B_AND_AS_FLAG;
343
344 if (self.flags & 16) != 0 && self.fragment_index != 0 {
346 id |= CONTINUOUS_SEND_FLAG;
347 }
348
349 writer.write_u8(id)?;
350 writer.write_u24(self.sequence_number, Endian::Little)?;
351
352 writer.write_u8(self.flags)?;
354 writer.write_u16(self.length_in_bytes * 8, Endian::Big)?;
355
356 if self.is_reliable()? {
357 writer.write_u24(self.reliable_frame_index, Endian::Little)?;
358 }
359
360 if self.is_sequenced()? {
361 writer.write_u24(self.sequenced_frame_index, Endian::Little)?;
362 }
363 if self.is_ordered()? {
364 writer.write_u24(self.ordered_frame_index, Endian::Little)?;
365 writer.write_u8(self.order_channel)?;
366 }
367
368 if (self.flags & 16) != 0 {
370 writer.write_u32(self.compound_size, Endian::Big)?;
371 writer.write_u16(self.compound_id, Endian::Big)?;
372 writer.write_u32(self.fragment_index, Endian::Big)?;
373 }
374 Ok(())
375 }
376
377 pub fn is_fragment(&self) -> bool {
378 (self.flags & 16) != 0
379 }
380
381 pub fn is_reliable(&self) -> Result<bool> {
382 let r = Reliability::from((self.flags & 224) >> 5)?;
383 Ok(matches!(
384 r,
385 Reliability::Reliable | Reliability::ReliableOrdered | Reliability::ReliableSequenced
386 ))
387 }
388
389 pub fn is_ordered(&self) -> Result<bool> {
390 let r = Reliability::from((self.flags & 224) >> 5)?;
391 Ok(matches!(
392 r,
393 Reliability::UnreliableSequenced
394 | Reliability::ReliableOrdered
395 | Reliability::ReliableSequenced
396 ))
397 }
398
399 pub fn is_sequenced(&self) -> Result<bool> {
400 let r = Reliability::from((self.flags & 224) >> 5)?;
401 Ok(matches!(
402 r,
403 Reliability::UnreliableSequenced | Reliability::ReliableSequenced
404 ))
405 }
406 pub fn reliability(&self) -> Result<Reliability> {
407 Reliability::from((self.flags & 224) >> 5)
408 }
409
410 pub fn _size(&self) -> Result<usize> {
411 let mut ret = 0;
412 ret += 1;
414 ret += 3;
416 ret += 1;
418 ret += 2;
420
421 if self.is_reliable()? {
422 ret += 3;
424 }
425 if self.is_sequenced()? {
426 ret += 3;
428 }
429 if self.is_ordered()? {
430 ret += 4;
432 }
433 if (self.flags & 16) != 0 {
434 ret += 10;
436 }
437 ret += self.data.len();
439 Ok(ret)
440 }
441}
442
443pub struct FrameVec {
444 #[cfg(test)]
445 pub frames: Vec<FrameSetPacket>,
446}
447
448impl FrameVec {
449 #[cfg(test)]
450 pub fn new(buf: &[u8]) -> Result<Self> {
451 let mut frames = Vec::new();
452 Self::decode_into(buf, &mut frames)?;
453 Ok(Self { frames })
454 }
455
456 pub(crate) fn decode_into(buf: &[u8], frames: &mut Vec<FrameSetPacket>) -> Result<()> {
458 Self::decode(buf, frames, false)
459 }
460
461 pub(crate) fn decode_owned_into(
462 mut buf: Vec<u8>,
463 frames: &mut Vec<FrameSetPacket>,
464 ) -> Result<()> {
465 Self::decode(&buf, frames, true)?;
466 if let [frame] = frames.as_mut_slice() {
467 let length = usize::from(frame.length_in_bytes);
470 let offset = buf.len() - length;
471 buf.copy_within(offset.., 0);
472 buf.truncate(length);
473 frame.data = buf.into();
474 }
475 Ok(())
476 }
477
478 fn decode(buf: &[u8], frames: &mut Vec<FrameSetPacket>, owned: bool) -> Result<()> {
479 frames.clear();
480 let mut reader = RaknetReader::new(buf);
481 let id = reader.read_u8()?;
482 if !(0x80..=0x8d).contains(&id) {
483 return Err(RaknetError::PacketHeaderError);
484 }
485 let sequence_number = reader.read_u24(Endian::Little)?;
486 if reader.pos() == buf.len() as u64 {
487 return Err(RaknetError::PacketParseError);
488 }
489 while reader.pos() < buf.len() as u64 {
490 let mut frame = FrameSetPacket {
491 sequence_number,
492 flags: 0,
493 length_in_bytes: 0,
494 reliable_frame_index: 0,
495 sequenced_frame_index: 0,
496 ordered_frame_index: 0,
497 order_channel: 0,
498 compound_size: 0,
499 compound_id: 0,
500 fragment_index: 0,
501 data: FramePayload::default(),
502 shared_payload: None,
503 };
504
505 frame.flags = reader.read_u8()?;
506 let length_in_bits = reader.read_u16(Endian::Big)?;
507 if length_in_bits % 8 != 0 {
508 return Err(RaknetError::PacketParseError);
509 }
510 frame.length_in_bytes = length_in_bits / 8;
511
512 if frame.is_reliable()? {
513 frame.reliable_frame_index = reader.read_u24(Endian::Little)?;
514 }
515 if frame.is_sequenced()? {
516 frame.sequenced_frame_index = reader.read_u24(Endian::Little)?;
517 }
518 if frame.is_ordered()? {
519 frame.ordered_frame_index = reader.read_u24(Endian::Little)?;
520 frame.order_channel = reader.read_u8()?;
521 }
522 if frame.is_fragment() {
523 frame.compound_size = reader.read_u32(Endian::Big)?;
524 frame.compound_id = reader.read_u16(Endian::Big)?;
525 frame.fragment_index = reader.read_u32(Endian::Big)?;
526 if frame.compound_size == 0 || frame.fragment_index >= frame.compound_size {
527 return Err(RaknetError::PacketParseError);
528 }
529 }
530
531 let data = reader.read_slice(usize::from(frame.length_in_bytes))?;
532 if !owned || !frames.is_empty() || reader.pos() != buf.len() as u64 {
535 frame.data = data.to_vec().into();
536 }
537 frames.push(frame);
538 }
539
540 Ok(())
541 }
542}
543
544#[derive(Default)]
545pub struct ACKSet {
546 ack: Vec<(u32, u32)>,
547 nack: Vec<(u32, u32)>,
548 next_expected: u32,
549}
550
551impl ACKSet {
552 pub fn insert(&mut self, s: u32) {
553 let mut split = None;
555 self.nack.retain_mut(|(start, end)| {
556 if s < *start || s > *end {
557 return true;
558 }
559 if *start == *end {
560 return false;
561 }
562 if s == *start {
563 *start += 1;
564 } else if s == *end {
565 *end -= 1;
566 } else {
567 split = Some((s + 1, *end));
568 *end = s - 1;
569 }
570 true
571 });
572 if let Some(range) = split {
573 self.nack.push(range);
574 }
575 if s == self.next_expected || sequence::newer(s, self.next_expected) {
576 if s != self.next_expected {
577 if s > self.next_expected {
578 self.nack.push((self.next_expected, s - 1));
579 } else {
580 self.nack.push((self.next_expected, sequence::MASK));
581 if s != 0 {
582 self.nack.push((0, s - 1));
583 }
584 }
585 }
586 self.next_expected = sequence::next(s);
587 }
588
589 for i in 0..self.ack.len() {
590 let a = self.ack[i];
591 if (a.0..=a.1).contains(&s) {
592 return;
593 }
594 if a.0 != 0 && s == a.0 - 1 {
595 self.ack[i].0 = s;
596 return;
597 }
598 if s == a.1.saturating_add(1) {
599 self.ack[i].1 = s;
600 return;
601 }
602 }
603 self.ack.push((s, s));
604 }
605
606 #[cfg(test)]
607 pub fn get_ack(&mut self) -> Vec<(u32, u32)> {
608 std::mem::take(&mut self.ack)
609 }
610
611 fn take_ack_into(&mut self, output: &mut Vec<(u32, u32)>) {
612 output.clear();
613 std::mem::swap(output, &mut self.ack);
614 }
615
616 fn take_nack_into(&mut self, output: &mut Vec<(u32, u32)>) {
617 output.clear();
618 std::mem::swap(output, &mut self.nack);
619 }
620
621 #[cfg(test)]
622 pub fn get_nack(&mut self) -> Vec<(u32, u32)> {
623 std::mem::take(&mut self.nack)
624 }
625}
626
627#[derive(Default)]
629struct ChannelIndexes {
630 primary: u32,
631 others: HashMap<u8, u32>,
632}
633
634impl ChannelIndexes {
635 fn get(&self, channel: u8) -> u32 {
636 if channel == 0 {
637 self.primary
638 } else {
639 self.others.get(&channel).copied().unwrap_or(0)
640 }
641 }
642
643 fn get_mut(&mut self, channel: u8) -> &mut u32 {
644 if channel == 0 {
645 &mut self.primary
646 } else {
647 self.others.entry(channel).or_default()
648 }
649 }
650
651 #[cfg(test)]
652 fn insert(&mut self, channel: u8, index: u32) {
653 *self.get_mut(channel) = index;
654 }
655}
656
657#[derive(Default)]
658pub struct RecvQ {
659 sequenced_frame_indexes: ChannelIndexes,
660 last_ordered_indexes: ChannelIndexes,
661 sequence_number_ackset: ACKSet,
662 packets: HashMap<u32, FrameSetPacket>,
663 ordered_packets: HashMap<(u8, u32), FrameSetPacket>,
664 fragment_queue: FragmentQ,
665 reliable_window: ReliableWindow,
666 ordered_bytes: usize,
667 ready_ordered: Vec<FrameSetPacket>,
668 ready_bytes: usize,
669}
670
671impl RecvQ {
672 pub fn new() -> Self {
673 Self::default()
674 }
675
676 pub fn insert(&mut self, frame: FrameSetPacket) -> Result<()> {
677 let reliability = frame.reliability()?;
678 self.sequence_number_ackset.insert(frame.sequence_number);
679 if frame.is_reliable()? && !self.reliable_window.accept(frame.reliable_frame_index)? {
680 return Ok(());
681 }
682
683 match reliability {
684 Reliability::Unreliable => {
685 self.packets.entry(frame.sequence_number).or_insert(frame);
686 }
687 Reliability::UnreliableSequenced => {
688 let channel = frame.order_channel;
689 let sequence_index = frame.sequenced_frame_index;
690 let sequence_number = frame.sequence_number;
691 let last_index = self.sequenced_frame_indexes.get_mut(channel);
692 if sequence_index == *last_index || sequence::newer(sequence_index, *last_index) {
693 self.packets.entry(sequence_number).or_insert(frame);
694 *last_index = sequence::next(sequence_index);
695 }
696 }
697 Reliability::Reliable => {
698 self.packets.insert(frame.sequence_number, frame);
699 }
700 Reliability::ReliableOrdered => {
701 let expected_index = self.last_ordered_indexes.get(frame.order_channel);
702 if sequence::newer(expected_index, frame.ordered_frame_index) {
703 return Ok(());
704 }
705 if sequence::distance(frame.ordered_frame_index, expected_index)
706 >= sequence::RECEIVE_WINDOW
707 {
708 return Err(RaknetError::PacketParseError);
709 }
710 let complete = if frame.is_fragment() {
711 self.fragment_queue.insert_and_take_completed(frame)?
712 } else {
713 Some(frame)
714 };
715 if let Some(frame) = complete {
716 let key = (frame.order_channel, frame.ordered_frame_index);
717 if self.ordered_packets.contains_key(&key) {
718 return Ok(());
719 }
720 if self.ordered_packets.len() + self.ready_ordered.len()
721 >= sequence::RECEIVE_WINDOW as usize
722 || self.ordered_bytes + self.ready_bytes + frame.data.len()
723 > 64 * 1024 * 1024
724 {
725 return Err(RaknetError::PacketParseError);
726 }
727 if frame.ordered_frame_index == expected_index {
728 let channel = frame.order_channel;
729 self.ready_bytes += frame.data.len();
730 self.ready_ordered.push(frame);
731 let expected = self.last_ordered_indexes.get_mut(channel);
732 *expected = sequence::next(*expected);
733 while !self.ordered_packets.is_empty() {
735 let Some(frame) = self.ordered_packets.remove(&(channel, *expected))
736 else {
737 break;
738 };
739 self.ordered_bytes -= frame.data.len();
740 self.ready_bytes += frame.data.len();
741 self.ready_ordered.push(frame);
742 *expected = sequence::next(*expected);
743 }
744 } else {
745 self.ordered_bytes += frame.data.len();
746 self.ordered_packets.insert(key, frame);
747 }
748 }
749 }
750
751 Reliability::ReliableSequenced => {
752 let channel = frame.order_channel;
753 let sequence_index = frame.sequenced_frame_index;
754 let sequence_number = frame.sequence_number;
755 let last_index = self.sequenced_frame_indexes.get_mut(channel);
756 if sequence_index == *last_index || sequence::newer(sequence_index, *last_index) {
757 self.packets.entry(sequence_number).or_insert(frame);
758 *last_index = sequence::next(sequence_index);
759 }
760 }
761 }
762 Ok(())
763 }
764
765 #[cfg(test)]
766 pub fn get_ack(&mut self) -> Vec<(u32, u32)> {
767 self.sequence_number_ackset.get_ack()
768 }
769
770 #[cfg(test)]
771 pub fn get_nack(&mut self) -> Vec<(u32, u32)> {
772 self.sequence_number_ackset.get_nack()
773 }
774
775 pub(crate) fn take_ack_into(&mut self, output: &mut Vec<(u32, u32)>) {
776 self.sequence_number_ackset.take_ack_into(output);
777 }
778
779 pub(crate) fn take_nack_into(&mut self, output: &mut Vec<(u32, u32)>) {
780 self.sequence_number_ackset.take_nack_into(output);
781 }
782
783 #[cfg(test)]
784 pub fn flush(&mut self, _peer_addr: &SocketAddr) -> Vec<FrameSetPacket> {
785 let mut frames = Vec::new();
786 self.flush_into(&mut frames);
787 frames
788 }
789
790 pub(crate) fn flush_into(&mut self, frames: &mut Vec<FrameSetPacket>) {
791 frames.append(&mut self.ready_ordered);
792 self.ready_bytes = 0;
793 if self.packets.is_empty() {
794 return;
795 }
796 let mut packets: Vec<_> = self.packets.drain().map(|(_, frame)| frame).collect();
797 packets.sort_unstable_by_key(|frame| frame.sequence_number);
798 frames.extend(packets);
799 }
800 pub(crate) fn needs_maintenance(&self) -> bool {
801 !self.sequence_number_ackset.nack.is_empty() || self.fragment_queue.size() != 0
802 }
803
804 pub fn fragments_expired(&self) -> bool {
805 self.fragment_queue.expired()
806 }
807
808 pub fn get_ordered_packet(&self) -> usize {
809 self.ordered_packets.len() + self.ready_ordered.len()
810 }
811
812 pub fn get_fragment_queue_size(&self) -> usize {
813 self.fragment_queue.size()
814 }
815
816 pub fn get_ordered_keys(&self) -> Vec<(u8, u32)> {
817 self.ordered_packets.keys().copied().collect()
818 }
819
820 pub fn get_size(&self) -> usize {
821 self.packets.len()
822 }
823}
824
825#[cfg(feature = "send-policy")]
826struct SendScheduler {
827 capacity_notifier: std::sync::Weak<tokio::sync::Notify>,
828 options: crate::SendOptions,
829 unreliable: VecDeque<FrameSetPacket>,
830 unreliable_bytes: usize,
831 use_legacy: bool,
832 persistent: bool,
833 next_lane: usize,
834}
835
836pub struct SendQ {
837 mtu: u16,
838 ack_sequence_number: u32,
839 sequence_number: u32,
840 reliable_frame_index: u32,
841 sequenced_frame_indexes: ChannelIndexes,
842 ordered_frame_indexes: ChannelIndexes,
843 compound_id: u16,
844 packets: VecDeque<FrameSetPacket>,
846 queued_unreliable: usize,
847 buffered_bytes: usize,
848 rto: i64,
849 srtt: i64,
850 sent_packet: Vec<(FrameSetPacket, bool, i64, u32, Vec<u32>)>,
851 flight_order_dirty: bool,
852 next_retry: i64,
853 retries_pending: bool,
854 coalesce: bool,
855 coalescing_allowed: bool,
856 grouped_acks: bool,
857 #[cfg(feature = "send-policy")]
858 scheduler: Option<Box<SendScheduler>>,
859}
860
861impl SendQ {
862 pub const DEFAULT_TIMEOUT_MILLS: i64 = 50;
863 const MAX_IN_FLIGHT_PACKETS: usize = 64;
864 const MAX_COALESCED_FRAMES: usize = 8;
865 const MAX_BUFFERED_BYTES: usize = 64 * 1024 * 1024;
866 const SEND_HIGH_WATER: usize = 256 * 1024;
867 const FRAME_BUDGET: usize = 128;
868
869 const RTO_UBOUND: i64 = 12000;
870 const RTO_LBOUND: i64 = 50;
871
872 pub fn new(mtu: u16) -> Self {
873 Self {
874 mtu,
875 ack_sequence_number: sequence::MASK,
876 sequence_number: 0,
877 packets: VecDeque::new(),
878 #[cfg(feature = "send-policy")]
879 scheduler: None,
880 queued_unreliable: 0,
881 buffered_bytes: 0,
882 sent_packet: vec![],
883 flight_order_dirty: false,
884 next_retry: i64::MAX,
885 retries_pending: false,
886 coalesce: false,
887 coalescing_allowed: true,
888 grouped_acks: false,
889 reliable_frame_index: 0,
890 sequenced_frame_indexes: ChannelIndexes::default(),
891 ordered_frame_indexes: ChannelIndexes::default(),
892 compound_id: 0,
893
894 rto: SendQ::DEFAULT_TIMEOUT_MILLS,
895 srtt: SendQ::DEFAULT_TIMEOUT_MILLS,
896 }
897 }
898
899 fn required_bytes(&self, reliability: Reliability, len: usize) -> Result<usize> {
900 let payload = usize::from(self.mtu)
901 .checked_sub(60)
902 .filter(|n| *n > 0)
903 .ok_or(RaknetError::PacketSizeExceedMTU)?;
904 let frames = if reliability == Reliability::ReliableOrdered && len > payload {
905 let frames = len.div_ceil(payload);
906 if frames > crate::fragment::MAX_FRAGMENTS {
907 return Err(RaknetError::PacketSizeExceedMTU);
908 }
909 frames
910 } else {
911 1
912 };
913 let bytes = frames
914 .checked_mul(Self::FRAME_BUDGET)
915 .and_then(|n| n.checked_add(len))
916 .filter(|n| *n <= Self::MAX_BUFFERED_BYTES)
917 .ok_or(RaknetError::PacketSizeExceedMTU)?;
918 Ok(bytes)
919 }
920
921 pub fn has_capacity(&self, reliability: Reliability, len: usize) -> Result<bool> {
922 let required = self.required_bytes(reliability, len)?;
923 #[cfg(feature = "send-policy")]
925 {
926 self.has_reserved_capacity(reliability, required)
927 }
928 #[cfg(not(feature = "send-policy"))]
929 {
930 Ok(self.buffered_bytes.saturating_add(required) <= Self::SEND_HIGH_WATER.max(required))
931 }
932 }
933
934 pub(crate) fn enable_coalescing(&mut self) {
935 self.coalesce = self.coalescing_allowed;
936 }
937
938 pub(crate) fn allows_coalescing(&self) -> bool {
939 self.coalescing_allowed
940 }
941
942 pub(crate) fn has_batch_capacity(
943 &self,
944 reliability: Reliability,
945 mut lengths: impl Iterator<Item = usize>,
946 ) -> Result<bool> {
947 let required = lengths.try_fold(0usize, |total, length| {
948 total
949 .checked_add(self.required_bytes(reliability, length)?)
950 .filter(|&bytes| bytes <= Self::MAX_BUFFERED_BYTES)
951 .ok_or(RaknetError::PacketSizeExceedMTU)
952 })?;
953 #[cfg(feature = "send-policy")]
954 {
955 self.has_batch_required_capacity(reliability, required)
956 }
957 #[cfg(not(feature = "send-policy"))]
958 {
959 Ok(self.buffered_bytes.saturating_add(required) <= Self::SEND_HIGH_WATER.max(required))
960 }
961 }
962
963 pub fn insert(&mut self, reliability: Reliability, buf: &[u8]) -> Result<()> {
964 self.insert_with_order_channel(reliability, buf, 0)
965 }
966
967 pub fn insert_with_order_channel(
968 &mut self,
969 reliability: Reliability,
970 buf: &[u8],
971 order_channel: u8,
972 ) -> Result<()> {
973 self.insert_payload(reliability, buf, order_channel, None)
974 }
975
976 pub(crate) fn insert_bytes(
977 &mut self,
978 reliability: Reliability,
979 data: &bytes::Bytes,
980 order_channel: u8,
981 ) -> Result<()> {
982 self.insert_payload(reliability, data, order_channel, Some(data))
983 }
984
985 fn insert_payload(
986 &mut self,
987 reliability: Reliability,
988 buf: &[u8],
989 order_channel: u8,
990 owned: Option<&bytes::Bytes>,
991 ) -> Result<()> {
992 let reserved = self.required_bytes(reliability, buf.len())?;
993 if reserved > Self::MAX_BUFFERED_BYTES - self.buffered_bytes {
994 return Err(RaknetError::PacketSizeExceedMTU);
995 }
996 let max_payload = self
997 .mtu
998 .checked_sub(60)
999 .map(usize::from)
1000 .filter(|size| *size > 0)
1001 .ok_or(RaknetError::PacketSizeExceedMTU)?;
1002
1003 match reliability {
1004 Reliability::Unreliable => {
1005 if buf.len() > max_payload {
1006 return Err(RaknetError::PacketSizeExceedMTU);
1007 }
1008
1009 let frame = FrameSetPacket::from_payload(
1010 reliability,
1011 owned.map_or_else(
1012 || FramePayload::Owned(buf.to_vec()),
1013 |data| FramePayload::Bytes(data.clone()),
1014 ),
1015 );
1016 #[cfg(feature = "send-policy")]
1017 self.push_unreliable(frame);
1018 #[cfg(not(feature = "send-policy"))]
1019 self.packets.push_back(frame);
1020 }
1021 Reliability::UnreliableSequenced => {
1022 if buf.len() > max_payload {
1023 return Err(RaknetError::PacketSizeExceedMTU);
1024 }
1025
1026 let sequenced_frame_index = {
1027 let index = self.sequenced_frame_indexes.get_mut(order_channel);
1028 let current = *index;
1029 *index = sequence::next(*index);
1030 current
1031 };
1032 let ordered_frame_index = self.ordered_frame_indexes.get(order_channel);
1033 let mut frame = FrameSetPacket::from_payload(
1034 reliability,
1035 owned.map_or_else(
1036 || FramePayload::Owned(buf.to_vec()),
1037 |data| FramePayload::Bytes(data.clone()),
1038 ),
1039 );
1040 frame.order_channel = order_channel;
1041 frame.ordered_frame_index = ordered_frame_index;
1042 frame.sequenced_frame_index = sequenced_frame_index;
1043 #[cfg(feature = "send-policy")]
1044 self.push_unreliable(frame);
1045 #[cfg(not(feature = "send-policy"))]
1046 self.packets.push_back(frame);
1047 }
1048 Reliability::Reliable => {
1049 if buf.len() > max_payload {
1050 return Err(RaknetError::PacketSizeExceedMTU);
1051 }
1052
1053 let mut frame = FrameSetPacket::from_payload(
1054 reliability,
1055 owned.map_or_else(
1056 || FramePayload::copy_shared(buf),
1057 |data| FramePayload::Bytes(data.clone()),
1058 ),
1059 );
1060 frame.reliable_frame_index = self.reliable_frame_index;
1061 self.packets.push_back(frame);
1062 self.reliable_frame_index = sequence::next(self.reliable_frame_index);
1063 }
1064 Reliability::ReliableOrdered => {
1065 if buf.len() <= max_payload {
1066 let mut frame = FrameSetPacket::from_payload(
1067 reliability,
1068 owned.map_or_else(
1069 || FramePayload::copy_shared(buf),
1070 |data| FramePayload::Bytes(data.clone()),
1071 ),
1072 );
1073 frame.order_channel = order_channel;
1074 frame.reliable_frame_index = self.reliable_frame_index;
1075 frame.ordered_frame_index = *self.ordered_frame_indexes.get_mut(order_channel);
1076 self.packets.push_back(frame);
1077 self.reliable_frame_index = sequence::next(self.reliable_frame_index);
1078 } else {
1079 let compound_size = buf.len().div_ceil(max_payload);
1080 let compound_size = u32::try_from(compound_size)
1081 .map_err(|_| RaknetError::PacketSizeExceedMTU)?;
1082 let ordered_frame_index = *self.ordered_frame_indexes.get_mut(order_channel);
1083
1084 let data: Option<Arc<[u8]>> = owned.is_none().then(|| Arc::from(buf));
1085 let shared_payload = Arc::new(SharedPayload {
1086 remaining: AtomicUsize::new(compound_size as usize),
1087 bytes: buf.len(),
1088 });
1089 for (fragment_index, chunk) in buf.chunks(max_payload).enumerate() {
1090 let start = fragment_index * max_payload;
1091 let mut frame = FrameSetPacket::from_payload(
1092 reliability,
1093 match owned {
1094 Some(data) => {
1095 FramePayload::Bytes(data.slice(start..start + chunk.len()))
1096 }
1097 None => FramePayload::Shared {
1098 owner: data.as_ref().unwrap().clone(),
1099 start: start as u32,
1100 length: chunk.len() as u16,
1101 },
1102 },
1103 );
1104 frame.shared_payload = Some(shared_payload.clone());
1105 frame.flags |= 16;
1106 frame.compound_size = compound_size;
1107 frame.compound_id = self.compound_id;
1108 frame.fragment_index = fragment_index as u32;
1109 frame.order_channel = order_channel;
1110 frame.reliable_frame_index = self.reliable_frame_index;
1111 frame.ordered_frame_index = ordered_frame_index;
1112 self.packets.push_back(frame);
1113 self.reliable_frame_index = sequence::next(self.reliable_frame_index);
1114 }
1115 self.compound_id = self.compound_id.wrapping_add(1);
1116 }
1117 let index = self.ordered_frame_indexes.get_mut(order_channel);
1118 *index = sequence::next(*index);
1119 }
1120 Reliability::ReliableSequenced => {
1121 if buf.len() > max_payload {
1122 return Err(RaknetError::PacketSizeExceedMTU);
1123 }
1124
1125 let sequenced_frame_index = {
1126 let index = self.sequenced_frame_indexes.get_mut(order_channel);
1127 let current = *index;
1128 *index = sequence::next(*index);
1129 current
1130 };
1131 let ordered_frame_index = self.ordered_frame_indexes.get(order_channel);
1132 let mut frame = FrameSetPacket::from_payload(
1133 reliability,
1134 owned.map_or_else(
1135 || FramePayload::copy_shared(buf),
1136 |data| FramePayload::Bytes(data.clone()),
1137 ),
1138 );
1139 frame.order_channel = order_channel;
1140 frame.reliable_frame_index = self.reliable_frame_index;
1141 frame.sequenced_frame_index = sequenced_frame_index;
1142 frame.ordered_frame_index = ordered_frame_index;
1143 self.packets.push_back(frame);
1144 self.reliable_frame_index = sequence::next(self.reliable_frame_index);
1145 }
1146 };
1147 self.buffered_bytes += reserved;
1148 #[cfg(not(feature = "send-policy"))]
1149 if matches!(
1150 reliability,
1151 Reliability::Unreliable | Reliability::UnreliableSequenced
1152 ) {
1153 self.queued_unreliable += 1;
1154 }
1155 Ok(())
1156 }
1157
1158 fn update_rto(&mut self, rtt: i64) {
1159 let previous = self.rto;
1160 (self.srtt, self.rto) = Self::latency_sample(self.srtt, rtt);
1161 if self.rto != previous {
1162 self.next_retry = 0;
1163 }
1164 }
1165
1166 fn latency_sample(srtt: i64, rtt: i64) -> (i64, i64) {
1167 let srtt = (srtt * 4 + rtt) / 5;
1168 (
1169 srtt,
1170 (srtt * 3 / 2).clamp(Self::RTO_LBOUND, Self::RTO_UBOUND),
1171 )
1172 }
1173
1174 pub(crate) fn mtu(&self) -> u16 {
1175 self.mtu
1176 }
1177
1178 pub fn get_rto(&self) -> i64 {
1179 self.rto
1180 }
1181
1182 pub fn nack(&mut self, sequence: u32, tick: i64) {
1183 for i in 0..self.sent_packet.len() {
1184 let item = &mut self.sent_packet[i];
1185 if item.1 && item.0.sequence_number == sequence {
1186 self.coalescing_allowed = false;
1189 self.coalesce = false;
1190 raknet_log_debug!(
1191 "packet {}-{}-{} nack {} times",
1192 item.0.sequence_number,
1193 item.0.reliable_frame_index,
1194 item.0.ordered_frame_index,
1195 item.3 + 1
1196 );
1197 let previous = item.0.sequence_number;
1198 self.flight_order_dirty = true;
1199 item.0.sequence_number = self.sequence_number;
1200 self.sequence_number = sequence::next(self.sequence_number);
1201 item.1 = false;
1202 self.retries_pending = true;
1203 item.2 = tick;
1204
1205 if item.4.len() == 64 {
1206 item.4.remove(0);
1207 }
1208 item.4.push(previous);
1209 }
1210 }
1211 }
1212
1213 pub fn ack(&mut self, sequence: u32, tick: i64) {
1214 if self.grouped_acks && self.sent_packet.len() > 1 {
1215 self.ack_ranges(&[(sequence, sequence)], tick);
1216 return;
1217 }
1218 let Some(index) = self
1220 .sent_packet
1221 .iter()
1222 .position(|item| item.0.sequence_number == sequence || item.4.contains(&sequence))
1223 else {
1224 return;
1225 };
1226 let item = self.sent_packet.remove(index);
1227 if self.grouped_acks && self.sent_packet.is_empty() {
1228 self.grouped_acks = false;
1229 }
1230 let released_payload = match &item.0.shared_payload {
1231 Some(payload) if payload.remaining.fetch_sub(1, Ordering::Relaxed) == 1 => {
1232 payload.bytes
1233 }
1234 Some(_) => 0,
1235 None => item.0.data.len(),
1236 };
1237 self.buffered_bytes -= released_payload + Self::FRAME_BUDGET;
1238 if sequence::newer(sequence, self.ack_sequence_number) {
1239 let span = sequence::distance(sequence, self.ack_sequence_number);
1240 let missing: Vec<_> = self
1241 .sent_packet
1242 .iter()
1243 .filter_map(|item| {
1244 let distance =
1245 sequence::distance(item.0.sequence_number, self.ack_sequence_number);
1246 (distance > 0 && distance < span).then_some(item.0.sequence_number)
1247 })
1248 .collect();
1249 for id in missing {
1250 self.nack(id, tick);
1251 }
1252 self.ack_sequence_number = sequence;
1253 }
1254 if item.3 == 0 {
1256 self.update_rto(tick.saturating_sub(item.2).max(0));
1257 }
1258 }
1259
1260 #[inline]
1261 pub fn ack_ranges(&mut self, ranges: &[(u32, u32)], tick: i64) {
1262 if self.grouped_acks {
1263 self.ack_ranges_inner::<true>(ranges, tick);
1264 } else {
1265 self.ack_ranges_inner::<false>(ranges, tick);
1266 }
1267 }
1268
1269 fn ack_ranges_inner<const GROUPED: bool>(&mut self, ranges: &[(u32, u32)], tick: i64) {
1270 if let [(start, end)] = ranges {
1271 if start == end && (!GROUPED || self.sent_packet.len() <= 1) {
1272 self.ack(*start, tick);
1273 return;
1274 }
1275 }
1276 let previous = self.ack_sequence_number;
1279 let mut newest = previous;
1280 let mut latency = (self.srtt, self.rto);
1281 let mut released_bytes = 0;
1282 let mut previous_sample = None;
1283 self.sent_packet.retain_mut(|item| {
1284 let acknowledged_id = std::iter::once(item.0.sequence_number)
1285 .chain(item.4.iter().copied())
1286 .find(|id| {
1287 ranges
1288 .iter()
1289 .any(|&(start, end)| start <= *id && *id <= end)
1290 });
1291 let Some(id) = acknowledged_id else {
1292 return true;
1293 };
1294 let payload_bytes = match &item.0.shared_payload {
1295 Some(payload) if payload.remaining.fetch_sub(1, Ordering::Relaxed) == 1 => {
1296 payload.bytes
1297 }
1298 Some(_) => 0,
1299 None => item.0.data.len(),
1300 };
1301 released_bytes += payload_bytes + Self::FRAME_BUDGET;
1302 if sequence::newer(id, newest) {
1303 newest = id;
1304 }
1305 if item.3 == 0 && (!GROUPED || previous_sample != Some(id)) {
1308 if GROUPED {
1309 previous_sample = Some(id);
1310 }
1311 latency = Self::latency_sample(latency.0, tick.saturating_sub(item.2).max(0));
1312 }
1313 false
1314 });
1315 self.buffered_bytes -= released_bytes;
1316 if GROUPED && self.sent_packet.is_empty() {
1317 self.grouped_acks = false;
1318 }
1319
1320 if self.rto != latency.1 {
1321 self.next_retry = 0;
1322 }
1323 (self.srtt, self.rto) = latency;
1324 if newest != previous {
1325 let span = sequence::distance(newest, previous);
1326 let missing: Vec<_> = self
1327 .sent_packet
1328 .iter()
1329 .filter_map(|item| {
1330 let distance = sequence::distance(item.0.sequence_number, previous);
1331 (distance > 0 && distance < span).then_some(item.0.sequence_number)
1332 })
1333 .collect();
1334 for id in missing {
1335 self.nack(id, tick);
1336 }
1337 self.ack_sequence_number = newest;
1338 }
1339 }
1340
1341 pub fn nack_ranges(&mut self, ranges: &[(u32, u32)], tick: i64) {
1342 if let [(start, end)] = ranges {
1343 if start == end {
1344 self.nack(*start, tick);
1345 return;
1346 }
1347 }
1348 let missing: Vec<_> = self
1349 .sent_packet
1350 .iter()
1351 .filter_map(|item| {
1352 let id = item.0.sequence_number;
1353 ranges
1354 .iter()
1355 .any(|&(start, end)| start <= id && id <= end)
1356 .then_some(id)
1357 })
1358 .collect();
1359 for id in missing {
1360 self.nack(id, tick);
1361 }
1362 }
1363
1364 fn retry_timeout(rto: i64, retries: u32) -> i64 {
1365 let mut timeout = rto;
1366 for _ in 0..retries.min(32) {
1367 timeout = ((timeout as f64 * 1.5) as i64).min(Self::RTO_UBOUND);
1368 }
1369 timeout
1370 }
1371
1372 fn tick(&mut self, tick: i64) {
1373 if tick < self.next_retry {
1374 return;
1375 }
1376 self.next_retry = i64::MAX;
1377 for i in 0..self.sent_packet.len() {
1378 let p = &mut self.sent_packet[i];
1379
1380 let cur_rto = Self::retry_timeout(self.rto, p.3);
1381
1382 if p.1 && tick - p.2 >= cur_rto {
1383 self.coalescing_allowed = false;
1384 self.coalesce = false;
1385 let previous = p.0.sequence_number;
1386 self.flight_order_dirty = true;
1387 p.0.sequence_number = self.sequence_number;
1388 self.sequence_number = sequence::next(self.sequence_number);
1389 p.1 = false;
1390 self.retries_pending = true;
1391 if p.4.len() == 64 {
1392 p.4.remove(0);
1393 }
1394 p.4.push(previous);
1395 } else if p.1 {
1396 self.next_retry = self.next_retry.min(p.2.saturating_add(cur_rto));
1397 }
1398 }
1399 }
1400
1401 pub fn flush(&mut self, tick: i64, peer_addr: &SocketAddr) -> OutgoingFrames {
1402 #[cfg(feature = "send-policy")]
1403 if self
1404 .scheduler
1405 .as_ref()
1406 .is_some_and(|s| !s.use_legacy || !s.unreliable.is_empty())
1407 {
1408 return self.flush_scheduled(tick);
1409 }
1410 self.tick(tick);
1411
1412 let capacity = self.packets.len().min(
1415 Self::MAX_IN_FLIGHT_PACKETS.saturating_sub(self.sent_packet.len())
1416 + self.queued_unreliable,
1417 );
1418 let mut ret = OutgoingFrames::with_capacity(capacity);
1419 if self.flight_order_dirty {
1422 self.sent_packet
1423 .sort_unstable_by_key(|packet| packet.0.sequence_number);
1424 self.flight_order_dirty = false;
1425 }
1426
1427 if self.retries_pending {
1428 for packet in &mut self.sent_packet {
1429 if !packet.1 {
1430 raknet_log_debug!(
1431 "{} , packet {}-{}-{} resend {} times",
1432 peer_addr,
1433 packet.0.sequence_number,
1434 packet.0.reliable_frame_index,
1435 packet.0.ordered_frame_index,
1436 packet.3 + 1
1437 );
1438 ret.push(packet.0.clone());
1439 packet.1 = true;
1440 packet.2 = tick;
1441 packet.3 = packet.3.saturating_add(1);
1442 self.next_retry = self
1443 .next_retry
1444 .min(tick.saturating_add(Self::retry_timeout(self.rto, packet.3)));
1445 }
1446 }
1447 self.retries_pending = false;
1448 }
1449
1450 let mut group_sequence = None;
1451 let mut group_bytes = 0;
1452 let mut group_frames = 0;
1453 let datagram_budget = usize::from(self.mtu).saturating_sub(28);
1454 let queued_count = self.packets.len();
1455 for _ in 0..queued_count {
1456 if self.sent_packet.len() >= Self::MAX_IN_FLIGHT_PACKETS && self.queued_unreliable == 0
1457 {
1458 break;
1459 }
1460 let Some(mut packet) = self.packets.pop_front() else {
1461 break;
1462 };
1463 let reliable = packet.is_reliable().unwrap_or(false);
1464 if reliable && self.sent_packet.len() >= Self::MAX_IN_FLIGHT_PACKETS {
1465 self.packets.push_back(packet);
1466 continue;
1467 }
1468
1469 if self.coalesce {
1470 let packable = matches!(packet.reliability(), Ok(Reliability::ReliableOrdered))
1471 && !packet.is_fragment();
1472 let frame_bytes = packet._size().expect("queued frames were validated") - 4;
1473 if packable
1474 && group_sequence.is_some()
1475 && group_frames < Self::MAX_COALESCED_FRAMES
1476 && group_bytes + frame_bytes <= datagram_budget
1477 {
1478 packet.sequence_number = group_sequence.unwrap();
1479 group_bytes += frame_bytes;
1480 group_frames += 1;
1481 self.grouped_acks = true;
1482 ret.coalesced = true;
1483 } else {
1484 packet.sequence_number = self.sequence_number;
1485 self.sequence_number = sequence::next(self.sequence_number);
1486 group_sequence = packable.then_some(packet.sequence_number);
1487 group_bytes = 4 + frame_bytes;
1488 group_frames = 1;
1491 }
1492 } else {
1493 packet.sequence_number = self.sequence_number;
1494 self.sequence_number = sequence::next(self.sequence_number);
1495 }
1496
1497 if reliable {
1498 ret.push(packet.clone());
1499 if self
1500 .sent_packet
1501 .last()
1502 .is_some_and(|previous| previous.0.sequence_number > packet.sequence_number)
1503 {
1504 self.flight_order_dirty = true;
1505 }
1506 self.sent_packet.push((packet, true, tick, 0, Vec::new()));
1507 self.next_retry = self.next_retry.min(tick.saturating_add(self.rto));
1508 } else {
1509 self.queued_unreliable -= 1;
1510 self.buffered_bytes -= packet.data.len() + Self::FRAME_BUDGET;
1511 ret.push(packet);
1512 }
1513 }
1514
1515 ret
1516 }
1517
1518 pub fn is_empty(&self) -> bool {
1519 #[cfg(feature = "send-policy")]
1520 {
1521 self.packets.is_empty()
1522 && self.sent_packet.is_empty()
1523 && self
1524 .scheduler
1525 .as_ref()
1526 .is_none_or(|s| s.unreliable.is_empty())
1527 }
1528 #[cfg(not(feature = "send-policy"))]
1529 {
1530 self.packets.is_empty() && self.sent_packet.is_empty()
1531 }
1532 }
1533
1534 pub fn get_reliable_queue_size(&self) -> usize {
1535 self.packets.len()
1536 }
1537
1538 pub fn get_sent_queue_size(&self) -> usize {
1539 self.sent_packet.len()
1540 }
1541}
1542
1543#[test]
1544fn malformed_datagrams_are_rejected() {
1545 assert!(FrameVec::new(&[0x80, 0, 0]).is_err());
1546 assert!(FrameVec::new(&[0x10, 0, 0, 0]).is_err());
1547}
1548
1549#[test]
1550fn send_queue_rejects_an_invalid_mtu() {
1551 let mut send_queue = SendQ::new(59);
1552 assert!(send_queue.insert(Reliability::Reliable, &[0xfe]).is_err());
1553}
1554
1555#[cfg(feature = "send-policy")]
1556impl SendQ {
1557 pub(crate) fn batch_can_coalesce(
1561 &self,
1562 reliability: Reliability,
1563 mut lengths: impl Iterator<Item = usize>,
1564 ) -> Result<bool> {
1565 let allowed = self.allows_coalescing() && reliability == Reliability::ReliableOrdered;
1566 if reliability == Reliability::ReliableOrdered && !allowed {
1567 self.has_batch_capacity(reliability, lengths)?;
1568 return Ok(false);
1569 }
1570 let max_payload = usize::from(self.mtu).saturating_sub(60);
1571 let datagram_budget = usize::from(self.mtu).saturating_sub(28);
1572 let mut previous = None;
1573 let mut packable = false;
1574 let required = lengths.try_fold(0usize, |total, length| {
1575 if reliability != Reliability::ReliableOrdered && length > max_payload {
1576 return Err(RaknetError::PacketSizeExceedMTU);
1577 }
1578 if allowed && !packable && length <= max_payload {
1579 packable = previous.is_some_and(|last: usize| {
1580 last <= max_payload
1581 && last.saturating_add(length).saturating_add(24) <= datagram_budget
1582 });
1583 }
1584 previous = Some(length);
1585 total
1586 .checked_add(self.required_bytes(reliability, length)?)
1587 .filter(|&bytes| bytes <= Self::MAX_BUFFERED_BYTES)
1588 .ok_or(RaknetError::PacketSizeExceedMTU)
1589 })?;
1590 self.has_batch_required_capacity(reliability, required)?;
1591 Ok(allowed && packable)
1592 }
1593
1594 fn has_batch_required_capacity(
1595 &self,
1596 reliability: Reliability,
1597 required: usize,
1598 ) -> Result<bool> {
1599 if matches!(
1602 reliability,
1603 Reliability::Unreliable | Reliability::UnreliableSequenced
1604 ) && self
1605 .send_options()
1606 .is_some_and(|options| required > options.unreliable_queue_bytes)
1607 {
1608 return Ok(false);
1609 }
1610 self.has_reserved_capacity(reliability, required)
1611 }
1612
1613 fn ensure_scheduler(&mut self) -> &mut SendScheduler {
1614 if self.scheduler.is_none() {
1615 let mut unreliable = VecDeque::with_capacity(self.queued_unreliable);
1616 let mut unreliable_bytes = 0;
1617 for _ in 0..self.packets.len() {
1618 let frame = self.packets.pop_front().unwrap();
1619 if frame.is_reliable().unwrap_or(false) {
1620 self.packets.push_back(frame);
1621 } else {
1622 unreliable_bytes += frame.data.len() + Self::FRAME_BUDGET;
1623 unreliable.push_back(frame);
1624 }
1625 }
1626 self.queued_unreliable = 0;
1627 self.scheduler = Some(Box::new(SendScheduler {
1628 capacity_notifier: std::sync::Weak::new(),
1629 options: crate::SendOptions::default(),
1630 unreliable,
1631 unreliable_bytes,
1632 use_legacy: usize::from(self.mtu) * Self::MAX_IN_FLIGHT_PACKETS
1633 <= crate::SendOptions::default().flush_bytes,
1634 persistent: true,
1635 next_lane: 2,
1636 }));
1637 }
1638 self.scheduler.as_mut().unwrap()
1639 }
1640
1641 fn push_unreliable(&mut self, mut frame: FrameSetPacket) {
1642 if let Some(scheduler) = &mut self.scheduler {
1643 scheduler.unreliable_bytes += frame.data.len() + Self::FRAME_BUDGET;
1644 scheduler.unreliable.push_back(frame);
1645 } else {
1646 frame.sequence_number = self.sequence_number;
1647 self.sequence_number = sequence::next(self.sequence_number);
1648 self.queued_unreliable += 1;
1649 self.packets.push_back(frame);
1650 }
1651 }
1652
1653 pub(crate) fn register_capacity_notifier(&mut self, notifier: &Arc<tokio::sync::Notify>) {
1655 let Some(scheduler) = &mut self.scheduler else {
1656 return;
1657 };
1658 if !std::ptr::eq(scheduler.capacity_notifier.as_ptr(), Arc::as_ptr(notifier)) {
1659 scheduler.capacity_notifier = Arc::downgrade(notifier);
1660 }
1661 }
1662
1663 pub(crate) fn set_send_options(&mut self, options: crate::SendOptions) -> Result<()> {
1664 let mtu = usize::from(self.mtu);
1665 if options.in_flight_bytes < mtu
1666 || options.flush_bytes < mtu
1667 || options.unreliable_queue_bytes < mtu
1668 {
1669 return Err(RaknetError::PacketSizeExceedMTU);
1670 }
1671 let scheduler = self.ensure_scheduler();
1672 scheduler.options = options;
1673 scheduler.persistent = true;
1674 let defaults = crate::SendOptions::default();
1675 scheduler.use_legacy = options.in_flight_frames == defaults.in_flight_frames
1676 && options.in_flight_bytes == defaults.in_flight_bytes
1677 && options.flush_bytes == defaults.flush_bytes
1678 && mtu * Self::MAX_IN_FLIGHT_PACKETS <= options.flush_bytes;
1679 Ok(())
1680 }
1681
1682 pub(crate) fn send_options(&self) -> Option<crate::SendOptions> {
1683 self.scheduler.as_ref().map(|s| s.options)
1684 }
1685
1686 #[inline]
1687 fn has_reserved_capacity(&self, reliability: Reliability, required: usize) -> Result<bool> {
1688 if self.scheduler.is_none() {
1689 return Ok(
1690 self.buffered_bytes.saturating_add(required) <= Self::SEND_HIGH_WATER.max(required)
1691 );
1692 }
1693 let unreliable = matches!(
1694 reliability,
1695 Reliability::Unreliable | Reliability::UnreliableSequenced
1696 );
1697 if !unreliable {
1698 let budget = match &self.scheduler {
1702 None if required <= Self::MAX_BUFFERED_BYTES - 64 * 1024 => {
1703 Some(Self::SEND_HIGH_WATER.max(required))
1704 }
1705 None => None,
1706 Some(s)
1707 if s.unreliable_bytes == 0 && required <= s.options.reliable_queue_bytes =>
1708 {
1709 Some(s.options.reliable_queue_bytes)
1710 }
1711 Some(_) => None,
1712 };
1713 if let Some(budget) = budget {
1714 return Ok(self.buffered_bytes.saturating_add(required) <= budget);
1715 }
1716 }
1717 let options = self.send_options().unwrap();
1718 let used_unreliable = self.scheduler.as_ref().map_or(0, |s| s.unreliable_bytes);
1719 let (used, limit) = if unreliable {
1720 if required > options.unreliable_queue_bytes {
1721 return Err(RaknetError::PacketSizeExceedMTU);
1722 }
1723 (used_unreliable, options.unreliable_queue_bytes)
1724 } else {
1725 let hard_limit = Self::MAX_BUFFERED_BYTES - options.unreliable_queue_bytes;
1726 if required > hard_limit {
1727 return Err(RaknetError::PacketSizeExceedMTU);
1728 }
1729 (
1730 self.buffered_bytes - used_unreliable,
1731 options.reliable_queue_bytes.max(required).min(hard_limit),
1732 )
1733 };
1734 Ok(used.saturating_add(required) <= limit
1735 && self.buffered_bytes.saturating_add(required) <= Self::MAX_BUFFERED_BYTES)
1736 }
1737
1738 fn flush_scheduled(&mut self, tick: i64) -> OutgoingFrames {
1739 self.tick(tick);
1740 if self.flight_order_dirty {
1741 self.sent_packet
1742 .sort_unstable_by_key(|packet| packet.0.sequence_number);
1743 self.flight_order_dirty = false;
1744 }
1745 let mut scheduler = self.scheduler.take().unwrap();
1748 let options = scheduler.options;
1749 let mut flight_bytes: usize = self.sent_packet.iter().map(|p| p.0._size().unwrap()).sum();
1752 let capacity = self.packets.len().min(
1753 options
1754 .in_flight_frames
1755 .saturating_sub(self.sent_packet.len()),
1756 ) + scheduler.unreliable.len();
1757 let mut output = OutgoingFrames::with_capacity(capacity.min(256));
1758 let quantum = usize::from(self.mtu) * 4;
1759 let mut bytes = 0;
1760 let mut unreliable_drained = false;
1761 let mut retry_cursor = 0;
1762 let mut group_sequence = None;
1763 let mut group_bytes = 0;
1764 let mut group_frames = 0;
1765 let datagram_budget = usize::from(self.mtu).saturating_sub(28);
1766 'rounds: loop {
1767 let mut progressed = false;
1768 let start = scheduler.next_lane;
1769 for offset in 0..3 {
1770 let lane = (start + offset) % 3;
1771 let mut lane_bytes = 0;
1772 while lane_bytes < quantum {
1773 let charge = match lane {
1774 0 => scheduler
1775 .unreliable
1776 .front()
1777 .map(|frame| frame._size().unwrap()),
1778 1 => {
1779 if !self.retries_pending {
1780 break;
1781 }
1782 while retry_cursor < self.sent_packet.len()
1783 && self.sent_packet[retry_cursor].1
1784 {
1785 retry_cursor += 1;
1786 }
1787 self.sent_packet
1788 .get(retry_cursor)
1789 .map(|item| item.0._size().unwrap())
1790 }
1791 _ => self.packets.front().and_then(|frame| {
1792 let size = frame._size().unwrap();
1793 (self.sent_packet.len() < options.in_flight_frames
1794 && flight_bytes + size <= options.in_flight_bytes)
1795 .then_some(size)
1796 }),
1797 };
1798 let Some(charge) = charge else {
1799 break;
1800 };
1801 if bytes + charge > options.flush_bytes {
1804 break;
1805 }
1806 match lane {
1807 0 => {
1808 let mut packet = scheduler.unreliable.pop_front().unwrap();
1809 packet.sequence_number = self.sequence_number;
1810 self.sequence_number = sequence::next(self.sequence_number);
1811 let reserved = packet.data.len() + Self::FRAME_BUDGET;
1812 self.buffered_bytes -= reserved;
1813 scheduler.unreliable_bytes -= reserved;
1814 unreliable_drained = true;
1815 output.push(packet);
1816 group_sequence = None;
1817 }
1818 1 => {
1819 let item = &mut self.sent_packet[retry_cursor];
1820 output.push(item.0.clone());
1821 item.1 = true;
1822 item.2 = tick;
1823 item.3 = item.3.saturating_add(1);
1824 self.next_retry = self
1825 .next_retry
1826 .min(tick.saturating_add(Self::retry_timeout(self.rto, item.3)));
1827 retry_cursor += 1;
1828 group_sequence = None;
1829 }
1830 _ => {
1831 let mut packet = self.packets.pop_front().unwrap();
1832 let packable = self.coalesce
1833 && matches!(packet.reliability(), Ok(Reliability::ReliableOrdered))
1834 && !packet.is_fragment();
1835 let frame_bytes = charge - 4;
1836 if packable
1837 && group_sequence.is_some()
1838 && group_frames < Self::MAX_COALESCED_FRAMES
1839 && group_bytes + frame_bytes <= datagram_budget
1840 {
1841 packet.sequence_number = group_sequence.unwrap();
1842 group_bytes += frame_bytes;
1843 group_frames += 1;
1844 self.grouped_acks = true;
1845 output.coalesced = true;
1846 } else {
1847 packet.sequence_number = self.sequence_number;
1848 self.sequence_number = sequence::next(self.sequence_number);
1849 group_sequence = packable.then_some(packet.sequence_number);
1850 group_bytes = charge;
1851 group_frames = 1;
1852 }
1853 output.push(packet.clone());
1854 if self
1855 .sent_packet
1856 .last()
1857 .is_some_and(|p| p.0.sequence_number > packet.sequence_number)
1858 {
1859 self.flight_order_dirty = true;
1860 }
1861 self.sent_packet.push((packet, true, tick, 0, Vec::new()));
1862 flight_bytes += charge;
1863 self.next_retry = self.next_retry.min(tick.saturating_add(self.rto));
1864 }
1865 }
1866 progressed = true;
1867 bytes += charge;
1868 lane_bytes += charge;
1869 scheduler.next_lane = (lane + 1) % 3;
1870 if bytes == options.flush_bytes {
1871 break 'rounds;
1872 }
1873 }
1874 }
1875 if !progressed {
1876 break;
1877 }
1878 }
1879 self.retries_pending = self.sent_packet.iter().any(|item| !item.1);
1880 if unreliable_drained {
1881 if let Some(capacity) = scheduler.capacity_notifier.upgrade() {
1882 capacity.notify_waiters();
1883 }
1884 }
1885 if scheduler.persistent || !scheduler.use_legacy || !scheduler.unreliable.is_empty() {
1886 self.scheduler = Some(scheduler);
1887 }
1888 output
1889 }
1890}
1891
1892#[tokio::test]
1893async fn test_recvq_orders_channels_independently() {
1894 let mut recvq = RecvQ::new();
1895
1896 let mut later_on_channel_one = FrameSetPacket::new(Reliability::ReliableOrdered, vec![1]);
1897 later_on_channel_one.sequence_number = 0;
1898 later_on_channel_one.order_channel = 1;
1899 later_on_channel_one.ordered_frame_index = 1;
1900 recvq.insert(later_on_channel_one).unwrap();
1901
1902 let mut first_on_channel_two = FrameSetPacket::new(Reliability::ReliableOrdered, vec![2]);
1903 first_on_channel_two.sequence_number = 1;
1904 first_on_channel_two.reliable_frame_index = 1;
1905 first_on_channel_two.order_channel = 2;
1906 first_on_channel_two.ordered_frame_index = 0;
1907 recvq.insert(first_on_channel_two).unwrap();
1908
1909 let ready = recvq.flush(&"127.0.0.1:0".parse().unwrap());
1910 assert_eq!(ready.len(), 1);
1911 assert_eq!(ready[0].order_channel, 2);
1912
1913 let mut first_on_channel_one = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0]);
1914 first_on_channel_one.sequence_number = 2;
1915 first_on_channel_one.reliable_frame_index = 2;
1916 first_on_channel_one.order_channel = 1;
1917 first_on_channel_one.ordered_frame_index = 0;
1918 recvq.insert(first_on_channel_one).unwrap();
1919
1920 let ready = recvq.flush(&"127.0.0.1:0".parse().unwrap());
1921 assert_eq!(ready.len(), 2);
1922 assert_eq!(ready[0].ordered_frame_index, 0);
1923 assert_eq!(ready[1].ordered_frame_index, 1);
1924 assert!(ready.iter().all(|frame| frame.order_channel == 1));
1925}
1926
1927#[tokio::test]
1928async fn test_sendq_maintains_ordered_indexes_per_channel() {
1929 let mut sendq = SendQ::new(1500);
1930 sendq
1931 .insert_with_order_channel(Reliability::ReliableOrdered, &[0xfe, 1], 3)
1932 .unwrap();
1933 sendq
1934 .insert_with_order_channel(Reliability::ReliableOrdered, &[0xfe, 2], 9)
1935 .unwrap();
1936
1937 let sent = sendq.flush(0, &"127.0.0.1:0".parse().unwrap());
1938 assert_eq!(sent.len(), 2);
1939 assert_eq!(sent[0].order_channel, 3);
1940 assert_eq!(sent[0].ordered_frame_index, 0);
1941 assert_eq!(sent[1].order_channel, 9);
1942 assert_eq!(sent[1].ordered_frame_index, 0);
1943}
1944
1945#[tokio::test]
1946async fn test_ackset() {
1947 let mut ackset = ACKSet::default();
1948
1949 ackset.insert(0);
1950 ackset.insert(1);
1951 ackset.insert(2);
1952 ackset.insert(4);
1953
1954 let acks = ackset.get_ack();
1955
1956 assert!(acks == vec![(0, 2), (4, 4)]);
1957
1958 let mut ackset = ACKSet::default();
1959
1960 ackset.insert(0);
1961 ackset.insert(1);
1962 ackset.insert(2);
1963 ackset.insert(6);
1964
1965 let acks = ackset.get_ack();
1966
1967 assert!(acks == vec![(0, 2), (6, 6)]);
1968
1969 let acks = ackset.get_ack();
1970
1971 assert!(acks == vec![]);
1972
1973 ackset.insert(0);
1974 ackset.insert(2);
1975
1976 let acks = ackset.get_ack();
1977
1978 assert!(acks == vec![(0, 0), (2, 2)]);
1979}
1980
1981#[tokio::test]
1982async fn test_frame_serialize_deserialize() {
1983 let p: Vec<u8> = [
1985 132, 0, 0, 0, 64, 0, 144, 0, 0, 0, 9, 146, 33, 7, 47, 57, 18, 128, 111, 0, 0, 0, 0, 20,
1986 200, 47, 41, 0,
1987 ]
1988 .to_vec();
1989
1990 let a = FrameSetPacket::deserialize(&p).unwrap();
1991 assert!(a.0.serialize().unwrap() == p);
1992}
1993
1994#[tokio::test]
1995async fn test_recvq() {
1996 let mut r = RecvQ::new();
1997 let mut p = FrameSetPacket::new(Reliability::Reliable, vec![]);
1998 p.sequence_number = 0;
1999 p.ordered_frame_index = 0;
2000 r.insert(p).unwrap();
2001
2002 let mut p = FrameSetPacket::new(Reliability::Reliable, vec![]);
2003 p.sequence_number = 1;
2004 p.reliable_frame_index = 1;
2005 p.ordered_frame_index = 1;
2006 r.insert(p).unwrap();
2007
2008 let ret = r.flush(&"0.0.0.0:0".parse().unwrap());
2009 assert!(ret.len() == 2);
2010}
2011
2012#[tokio::test]
2013async fn test_recvq_fragment() {
2014 let mut r = RecvQ::new();
2015 let mut p = FrameSetPacket::new(Reliability::ReliableOrdered, vec![1]);
2016 p.flags |= 16;
2017 p.sequence_number = 0;
2018 p.ordered_frame_index = 0;
2019 p.compound_id = 1;
2020 p.compound_size = 3;
2021 p.fragment_index = 0;
2022 p.order_channel = 7;
2023 r.insert(p).unwrap();
2024
2025 let mut p = FrameSetPacket::new(Reliability::ReliableOrdered, vec![2]);
2026 p.flags |= 16;
2027 p.sequence_number = 1;
2028 p.reliable_frame_index = 1;
2029 p.ordered_frame_index = 0;
2030 p.compound_id = 1;
2031 p.compound_size = 3;
2032 p.fragment_index = 1;
2033 p.order_channel = 7;
2034 r.insert(p).unwrap();
2035
2036 let mut p = FrameSetPacket::new(Reliability::ReliableOrdered, vec![3]);
2037 p.flags |= 16;
2038 p.sequence_number = 2;
2039 p.reliable_frame_index = 2;
2040 p.ordered_frame_index = 0;
2041 p.compound_id = 1;
2042 p.compound_size = 3;
2043 p.fragment_index = 2;
2044 p.order_channel = 7;
2045 r.insert(p).unwrap();
2046
2047 let ret = r.flush(&"0.0.0.0:0".parse().unwrap());
2048 assert!(ret.len() == 1);
2049 assert_eq!(ret[0].data.as_ref(), &[1, 2, 3]);
2050 assert_eq!(ret[0].order_channel, 7);
2051}
2052
2053#[tokio::test]
2054async fn test_sendq() {
2055 let mut s = SendQ::new(1500);
2056 let p = FrameSetPacket::new(Reliability::Reliable, vec![]);
2057 s.insert(Reliability::Reliable, &p.serialize().unwrap())
2058 .unwrap();
2059
2060 let p = FrameSetPacket::new(Reliability::Reliable, vec![]);
2061 s.insert(Reliability::Reliable, &p.serialize().unwrap())
2062 .unwrap();
2063
2064 let sockaddr: SocketAddr = "127.0.0.1:8000".parse().unwrap();
2065 let ret = s.flush(0, &sockaddr);
2066 assert!(ret.len() == 2);
2067
2068 s.ack(0, 0);
2069 s.ack(1, 0);
2070
2071 let ret = s.flush(300, &sockaddr);
2072 assert!(ret.is_empty());
2073}
2074
2075#[test]
2076fn send_queue_pipelines_reliable_packets_with_a_bounded_window() {
2077 let mut sendq = SendQ::new(1500);
2078 let peer = "127.0.0.1:8000".parse().unwrap();
2079
2080 for index in 0..SendQ::MAX_IN_FLIGHT_PACKETS + 1 {
2081 sendq
2082 .insert(Reliability::ReliableOrdered, &[0xfe, index as u8])
2083 .unwrap();
2084 let sent = sendq.flush(0, &peer);
2085 if index < SendQ::MAX_IN_FLIGHT_PACKETS {
2086 assert_eq!(sent.len(), 1);
2087 assert_eq!(sent[0].ordered_frame_index, index as u32);
2088 } else {
2089 assert!(sent.is_empty());
2090 }
2091 }
2092
2093 assert_eq!(sendq.get_sent_queue_size(), SendQ::MAX_IN_FLIGHT_PACKETS);
2094 assert_eq!(sendq.get_reliable_queue_size(), 1);
2095
2096 sendq.ack(0, 1);
2097 let sent = sendq.flush(1, &peer);
2098 assert_eq!(sent.len(), 1);
2099 assert_eq!(
2100 sent[0].ordered_frame_index,
2101 SendQ::MAX_IN_FLIGHT_PACKETS as u32
2102 );
2103 assert_eq!(sendq.get_sent_queue_size(), SendQ::MAX_IN_FLIGHT_PACKETS);
2104 assert_eq!(sendq.get_reliable_queue_size(), 0);
2105}
2106
2107#[test]
2108fn send_queue_retransmits_immediately_after_nack() {
2109 let mut sendq = SendQ::new(1500);
2110 let peer = "127.0.0.1:8000".parse().unwrap();
2111 sendq.insert(Reliability::Reliable, &[0xfe]).unwrap();
2112
2113 let first = sendq.flush(0, &peer);
2114 assert_eq!(first.len(), 1);
2115 assert_eq!(first[0].sequence_number, 0);
2116
2117 sendq.nack(0, 1);
2118 let retransmitted = sendq.flush(1, &peer);
2119 assert_eq!(retransmitted.len(), 1);
2120 assert_eq!(retransmitted[0].sequence_number, 1);
2121}
2122
2123#[tokio::test]
2124async fn test_client_packet1() {
2125 let a = [
2126 140, 3, 0, 0, 112, 44, 192, 2, 0, 0, 1, 0, 0, 0, 0, 0, 0, 19, 0, 0, 0, 0, 0, 0, 254, 236,
2127 189, 203, 114, 234, 74, 183, 239, 185, 79, 175, 170, 30, 99, 159, 110, 125, 59, 36, 1, 222,
2128 166, 122, 19, 35, 97, 48, 18, 19, 161, 11, 82, 69, 197, 23, 128, 248, 38, 32, 9, 107, 218,
2129 152, 91, 69, 61, 79, 53, 234, 141, 78, 243, 52, 234, 5, 170, 119, 90, 149, 41, 165, 48, 41,
2130 11, 12, 2, 95, 214, 154, 255, 198, 47, 86, 120, 77, 15, 75, 202, 203, 24, 57, 46, 153, 249,
2131 223, 255, 199, 127, 253, 47, 255, 246, 111, 255, 229, 255, 253, 111, 255, 227, 191, 254,
2132 63, 255, 243, 191, 253, 219, 255, 249, 239, 163, 201, 96, 58, 255, 247, 255, 237, 127, 255,
2133 247, 241, 166, 53, 25, 54, 70, 211, 206, 180, 165, 152, 91, 181, 172, 221, 53, 159, 155,
2134 243, 95, 162, 215, 107, 222, 52, 125, 177, 105, 247, 61, 165, 107, 5, 205, 158, 84, 29,
2135 244, 250, 209, 111, 85, 214, 106, 166, 21, 52, 122, 91, 93, 209, 77, 165, 222, 51, 117,
2136 215, 21, 20, 197, 109, 148, 55, 166, 220, 122, 234, 216, 101, 105, 108, 43, 157, 241, 180,
2137 218, 49, 26, 250, 189, 165, 184, 243, 129, 242, 171, 50, 170, 7, 247, 109, 241, 113, 173,
2138 221, 235, 205, 225, 60, 18, 218, 226, 196, 234, 134, 186, 57, 240, 43, 83, 43, 112, 167,
2139 150, 57, 89, 13, 75, 90, 205, 146, 244, 123, 183, 46, 111, 199, 182, 176, 214, 182, 147,
2140 141, 214, 16, 187, 227, 134, 38, 247, 234, 150, 219, 107, 84, 231, 78, 201, 17, 186, 115,
2141 189, 174, 139, 154, 51, 82, 212, 210, 192, 178, 30, 13, 49, 216, 186, 179, 218, 203, 192,
2142 143, 238, 61, 177, 37, 186, 211, 234, 147, 41, 105, 91, 87, 169, 61, 142, 239, 22, 207,
2143 170, 28, 205, 199, 117, 253, 121, 32, 107, 117, 213, 119, 166, 255, 234, 62, 254, 7, 249,
2144 238, 153, 219, 111, 9, 3, 219, 141, 28, 73, 17, 92, 83, 17, 189, 198, 100, 57, 10, 3, 97,
2145 76, 190, 221, 187, 39, 127, 167, 183, 154, 186, 253, 201, 170, 57, 35, 239, 61, 235, 150,
2146 58, 117, 103, 173, 25, 242, 186, 125, 215, 138, 220, 134, 245, 226, 53, 200, 239, 90, 53,
2147 209, 9, 215, 145, 35, 44, 130, 241, 153, 109, 214, 145, 45, 117, 32, 5, 229, 177, 185, 158,
2148 121, 210, 122, 48, 154, 7, 150, 105, 107, 162, 106, 233, 146, 41, 87, 23, 61, 163, 117,
2149 167, 149, 92, 167, 83, 215, 94, 220, 70, 165, 111, 6, 214, 196, 110, 8, 37, 237, 222, 107,
2150 184, 161, 76, 191, 243, 217, 19, 149, 142, 213, 112, 55, 134, 226, 54, 29, 163, 213, 29,
2151 218, 214, 203, 72, 246, 90, 154, 31, 61, 246, 76, 209, 234, 133, 74, 223, 158, 183, 126,
2152 15, 77, 241, 119, 199, 168, 117, 134, 194, 162, 163, 7, 90, 167, 59, 183, 218, 110, 67, 40,
2153 143, 130, 32, 178, 239, 181, 208, 233, 63, 110, 123, 91, 85, 26, 223, 221, 174, 45, 163,
2154 41, 245, 238, 107, 247, 170, 18, 149, 123, 155, 170, 173, 90, 206, 148, 124, 243, 139, 19,
2155 58, 211, 206, 76, 150, 180, 250, 72, 234, 212, 127, 73, 154, 161, 84, 239, 126, 253, 199,
2156 63, 158, 123, 119, 119, 203, 126, 120, 247, 220, 113, 55, 245, 97, 189, 212, 172, 255, 136,
2157 30, 159, 164, 122, 255, 95, 245, 141, 220, 159, 41, 149, 169, 218, 173, 174, 5, 249, 247,
2158 83, 205, 188, 185, 81, 220, 113, 251, 229, 101, 213, 249, 245, 123, 212, 248, 71, 239, 105,
2159 48, 151, 68, 103, 32, 175, 215, 63, 220, 113, 216, 169, 252, 243, 247, 211, 118, 230, 255,
2160 179, 210, 173, 173, 123, 230, 203, 63, 170, 163, 127, 12, 52, 39, 26, 247, 158, 182, 147,
2161 113, 52, 154, 4, 75, 219, 122, 168, 223, 120, 255, 152, 223, 138, 173, 135, 122, 79, 43,
2162 245, 27, 255, 254, 191, 210, 81, 92, 214, 250, 38, 29, 197, 90, 79, 9, 74, 164, 149, 221,
2163 158, 188, 88, 218, 126, 112, 51, 152, 213, 234, 93, 50, 12, 117, 255, 89, 208, 77, 171,
2164 214, 21, 2, 217, 182, 189, 154, 110, 76, 20, 163, 177, 136, 198, 247, 129, 234, 148, 188,
2165 103, 210, 74, 21, 203, 82, 102, 158, 161, 152, 94, 80, 251, 57, 52, 253, 77, 151, 244, 139,
2166 93, 255, 181, 29, 134, 250, 139, 37, 182, 44, 199, 154, 44, 180, 173, 94, 210, 67, 119,
2167 162, 217, 238, 168, 45, 173, 75, 166, 89, 49, 221, 121, 75, 181, 252, 201, 68, 13, 106, 11,
2168 215, 174, 172, 116, 193, 151, 122, 225, 164, 101, 88, 214, 131, 99, 121, 131, 81, 24, 45,
2169 12, 251, 177, 162, 202, 149, 101, 79, 168, 52, 76, 161, 114, 231, 153, 11, 127, 104, 76,
2170 54, 182, 29, 56, 35, 201, 157, 140, 103, 90, 91, 221, 186, 229, 126, 182, 7, 196, 213, 210,
2171 154, 41, 118, 115, 186, 154, 58, 246, 122, 78, 70, 227, 84, 183, 212, 109, 167, 222, 37,
2172 35, 153, 14, 228, 156, 142, 145, 201, 36, 38, 19, 59, 212, 151, 67, 179, 178, 28, 134, 90,
2173 64, 196, 74, 234, 204, 41, 107, 51, 191, 172, 209, 137, 62, 107, 10, 228, 191, 2, 249, 125,
2174 129, 14, 232, 81, 73, 141, 155, 111, 40, 69, 68, 110, 68, 59, 60, 24, 223, 255, 72, 255,
2175 46, 249, 125, 89, 84, 233, 223, 37, 127, 107, 52, 215, 35, 55, 12, 102, 78, 95, 15, 186,
2176 125, 75, 24, 52, 170, 155, 65, 95, 175, 52, 103, 145, 48, 154, 91, 1, 253, 123, 78, 191,
2177 187, 255, 78, 165, 68, 54, 240, 73, 243, 8, 244, 119, 205, 123, 107, 58, 108, 4, 179, 158,
2178 100, 85, 232, 39, 25, 230, 196, 245, 4, 171, 102, 155, 147, 246, 80, 140, 90, 227, 240,
2179 113, 213, 21, 20, 221, 54, 221, 182, 38, 91, 74, 215, 212, 90, 186, 18, 204, 187, 166, 181,
2180 26, 26, 90, 93, 187, 91, 180, 212, 128, 40, 129, 198, 194, 232, 88, 171, 229, 80, 28, 109,
2181 92, 203, 43, 121, 115, 125, 221, 19, 162, 178, 102, 138, 174, 109, 77, 12, 215, 168, 217,
2182 186, 36, 151, 71, 164, 139, 61, 223, 26, 12, 132, 245, 114, 68, 148, 196, 64, 138, 106,
2183 166, 160, 77, 6, 210, 66, 232, 137, 122, 107, 228, 59, 226, 184, 111, 169, 35, 203, 85, 61,
2184 165, 86, 25, 247, 221, 126, 215, 94, 76, 186, 118, 180, 30, 207, 38, 91, 77, 209, 219, 61,
2185 203, 213, 186, 225, 164, 231, 206, 213, 167, 14, 153, 252, 142, 45, 170, 227, 198, 164, 57,
2186 188, 143, 30, 109, 197, 93, 147, 103, 205, 173, 251, 214, 202, 178, 221, 27, 39, 80, 250,
2187 134, 160, 210, 46, 50, 255, 53, 31, 119, 199, 195, 137, 182, 49, 141, 199, 40, 108, 191,
2188 244, 212, 222, 63, 198, 213, 255, 188, 239, 206, 102, 155, 223, 102, 235, 159, 255, 252,
2189 241, 16, 244, 194, 233, 68, 170, 7, 27, 189, 211, 250, 189, 104, 143, 255, 241, 159, 66,
2190 221, 121, 169, 73, 98, 189, 90, 107, 78, 23, 29, 229, 215, 96, 54, 27, 153, 183, 182, 223,
2191 155, 223, 55, 127, 24, 227, 219, 167, 217, 237, 188, 31, 74, 154, 226, 255, 115, 220, 169,
2192 143, 165, 231, 135, 95, 245, 217, 118, 174, 201, 134, 255, 88, 121, 92, 132, 129, 123, 251,
2193 175, 250, 204, 189, 169, 46, 138, 78, 144, 126, 109, 161, 10, 170, 240, 32, 248, 27, 83,
2194 156, 172, 6, 162, 90, 233, 223, 85, 151, 214, 182, 21, 88, 37, 79, 242, 238, 149, 118, 143,
2195 152, 10, 131, 232, 111, 91, 209, 2, 85, 113, 239, 29, 99, 178, 37, 170, 76, 210, 173, 232,
2196 201, 104, 84, 183, 230, 220, 122, 26, 248, 138, 209, 149, 148, 167, 65, 73, 111, 91, 50,
2197 249, 194, 153, 85, 241, 204, 245, 218, 242, 215, 130, 121, 31, 84, 188, 192, 171, 13, 36,
2198 165, 54, 152, 43, 55, 157, 123, 85, 176, 228, 69, 203, 242, 197, 187, 129, 210, 10, 71,
2199 155, 231, 50, 25, 45, 147, 161, 185, 46, 15, 228, 201, 243, 56, 156, 56, 214, 92, 9, 212,
2200 134, 103, 142, 230, 53, 203, 157, 71, 83, 211, 242, 126, 118, 55, 196,
2201 ];
2202
2203 let b = FrameVec::new(&a).unwrap();
2204 assert!(b.frames.len() == 1);
2205 assert!(b.frames[0].is_fragment());
2206}
2207
2208#[tokio::test]
2209async fn test_client_packet2() {
2210 let p0 = [
2212 132, 0, 0, 0, 64, 0, 144, 0, 0, 0, 9, 162, 70, 235, 28, 218, 182, 26, 192, 0, 0, 0, 0, 16,
2213 151, 43, 113, 0,
2214 ];
2215 let p1 = [
2217 132, 1, 0, 0, 64, 0, 144, 0, 0, 0, 9, 162, 70, 235, 28, 218, 182, 26, 192, 0, 0, 0, 0, 16,
2218 151, 43, 113, 0,
2219 ];
2220 let p2 = [
2222 132, 2, 0, 0, 96, 9, 64, 1, 0, 0, 0, 0, 0, 0, 19, 4, 83, 237, 234, 82, 74, 188, 6, 23, 0,
2223 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 196, 178, 112, 86, 5, 59, 97, 219, 15, 0,
2224 0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 188, 210, 59, 150, 246,
2225 167, 182, 213, 33, 0, 0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0,
2226 132, 194, 47, 23, 175, 46, 78, 138, 23, 0, 0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128,
2227 0, 0, 0, 0, 0, 0, 136, 219, 85, 240, 191, 125, 172, 233, 10, 0, 0, 0, 6, 23, 0, 225, 138,
2228 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 80, 211, 212, 44, 191, 227, 124, 40, 13, 0, 0, 0,
2229 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 77, 13, 19, 149, 102, 140, 134,
2230 77, 16, 0, 0, 0, 4, 83, 237, 239, 254, 225, 138, 4, 63, 87, 214, 254, 225, 138, 4, 83, 236,
2231 159, 254, 225, 138, 4, 63, 87, 46, 254, 225, 138, 4, 63, 87, 56, 128, 225, 138, 4, 63, 87,
2232 56, 123, 225, 138, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255,
2233 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255,
2234 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 0, 0, 0, 0, 0, 0, 0,
2235 0, 0, 0, 0, 0, 16, 151, 56, 146, 0, 0, 72, 0, 0, 0, 0, 0, 16, 151, 56, 146,
2236 ];
2237 let p3 = [132, 3, 0, 0, 0, 0, 72, 0, 0, 0, 0, 0, 16, 151, 56, 161];
2239 let p4 = [
2241 140, 4, 0, 0, 112, 44, 192, 2, 0, 0, 1, 0, 0, 0, 0, 0, 0, 19, 0, 0, 0, 0, 0, 0, 254, 236,
2242 189, 203, 114, 234, 202, 214, 239, 249, 157, 94, 85, 61, 198, 119, 186, 181, 79, 72, 2,
2243 188, 77, 245, 140, 145, 48, 24, 137, 137, 208, 5, 169, 162, 98, 7, 32, 246, 4, 36, 97, 77,
2244 155, 201, 173, 162, 158, 231, 188, 81, 53, 171, 89, 47, 80, 189, 211, 170, 148, 148, 194,
2245 164, 156, 96, 16, 248, 178, 214, 252, 55, 126, 177, 194, 107, 122, 88, 82, 94, 198, 200,
2246 113, 201, 204, 255, 231, 127, 252, 215, 255, 242, 31, 255, 241, 95, 254, 223, 255, 251,
2247 127, 252, 215, 255, 254, 63, 255, 199, 127, 252, 159, 255, 57, 154, 12, 166, 243, 255, 252,
2248 223, 254, 247, 255, 28, 111, 90, 147, 97, 99, 52, 237, 76, 91, 138, 185, 85, 203, 218, 125,
2249 243, 165, 57, 255, 41, 122, 189, 230, 77, 211, 23, 155, 118, 223, 83, 186, 86, 208, 236,
2250 73, 213, 65, 175, 31, 253, 82, 101, 173, 102, 90, 65, 163, 183, 213, 21, 221, 84, 234, 61,
2251 83, 119, 93, 65, 81, 172, 70, 107, 233, 25, 138, 210, 179, 149, 23, 79, 18, 205, 142, 41,
2252 8, 163, 134, 179, 177, 77, 87, 182, 131, 86, 121, 80, 106, 213, 135, 125, 179, 162, 139,
2253 122, 203, 241, 131, 138, 109, 4, 93, 51, 168, 173, 135, 74, 75, 29, 214, 29, 97, 104, 4,
2254 55, 222, 182, 86, 26, 205, 189, 242, 192, 127, 217, 90, 219, 174, 48, 40, 233, 115, 171,
2255 49, 105, 141, 109, 189, 63, 182, 91, 63, 70, 141, 167, 237, 120, 166, 148, 213, 173, 92,
2256 30, 110, 149, 78, 91, 12, 66, 93, 241, 5, 93, 208, 45, 45, 156, 56, 143, 130, 18, 90, 194,
2257 147, 104, 55, 116, 209, 8, 163, 103, 211, 94, 151, 134, 91, 189, 50, 174, 235, 191, 135,
2258 138, 213, 237, 218, 21, 211, 182, 252, 233, 191, 187, 79, 255, 141, 124, 247, 204, 237,
2259 183, 132, 129, 237, 70, 142, 164, 8, 174, 169, 136, 94, 99, 178, 28, 133, 129, 48, 38, 223,
2260 238, 61, 180, 68, 183, 183, 154, 186, 253, 201, 170, 57, 123, 90, 107, 179, 110, 169, 83,
2261 119, 214, 154, 33, 175, 219, 247, 173, 200, 109, 88, 191, 189, 6, 249, 93, 171, 38, 58,
2262 225, 58, 114, 132, 69, 48, 62, 179, 205, 58, 178, 165, 14, 164, 160, 60, 54, 215, 51, 79,
2263 90, 15, 70, 243, 192, 50, 109, 77, 84, 45, 93, 50, 229, 234, 162, 103, 180, 238, 181, 146,
2264 235, 116, 234, 218, 111, 183, 81, 233, 155, 129, 53, 177, 27, 66, 73, 123, 240, 26, 110,
2265 40, 139, 238, 180, 250, 226, 137, 74, 199, 106, 184, 27, 67, 113, 155, 142, 209, 234, 14,
2266 109, 235, 247, 72, 246, 90, 154, 31, 61, 245, 76, 209, 234, 133, 74, 223, 158, 183, 126,
2267 13, 77, 241, 87, 199, 168, 117, 134, 194, 162, 163, 7, 90, 167, 59, 183, 218, 110, 67, 40,
2268 143, 130, 32, 178, 31, 180, 208, 233, 63, 109, 123, 91, 85, 26, 223, 223, 174, 45, 163, 41,
2269 245, 30, 106, 15, 170, 18, 149, 123, 155, 170, 173, 90, 206, 148, 124, 243, 111, 39, 116,
2270 166, 157, 153, 44, 105, 245, 145, 212, 169, 255, 148, 52, 67, 169, 222, 255, 252, 111, 237,
2271 167, 127, 74, 191, 234, 254, 100, 169, 140, 42, 171, 201, 92, 173, 142, 252, 127, 77, 86,
2272 247, 77, 251, 95, 254, 175, 127, 215, 127, 223, 246, 42, 255, 26, 253, 235, 165, 108, 133,
2273 255, 246, 151, 203, 202, 211, 250, 151, 175, 12, 203, 193, 124, 221, 155, 255, 67, 126,
2274 190, 109, 253, 99, 242, 243, 95, 211, 167, 219, 187, 237, 157, 107, 254, 43, 24, 107, 81,
2275 79, 22, 171, 225, 38, 252, 183, 240, 24, 4, 255, 114, 31, 21, 213, 235, 11, 195, 81, 239,
2276 159, 253, 161, 97, 169, 222, 77, 215, 41, 173, 164, 231, 254, 68, 157, 151, 199, 131, 31,
2277 15, 51, 229, 63, 255, 215, 120, 20, 151, 181, 190, 25, 143, 98, 173, 167, 4, 37, 210, 202,
2278 110, 79, 94, 44, 109, 63, 184, 25, 204, 106, 245, 174, 165, 184, 186, 255, 34, 232, 166,
2279 85, 235, 10, 129, 108, 219, 94, 77, 55, 38, 138, 209, 88, 68, 227, 135, 64, 117, 74, 222,
2280 11, 105, 165, 138, 101, 41, 51, 50, 138, 77, 47, 168, 253, 24, 154, 254, 166, 75, 250, 197,
2281 174, 255, 220, 14, 67, 253, 183, 37, 182, 44, 199, 154, 44, 180, 173, 94, 210, 67, 119,
2282 162, 217, 238, 168, 45, 173, 75, 166, 89, 49, 221, 121, 75, 181, 252, 201, 68, 13, 106, 11,
2283 215, 174, 172, 116, 193, 151, 122, 225, 164, 101, 88, 214, 163, 99, 121, 131, 81, 24, 45,
2284 12, 251, 169, 162, 202, 149, 101, 79, 168, 52, 76, 161, 114, 239, 153, 11, 127, 104, 76,
2285 54, 182, 29, 56, 35, 201, 157, 140, 103, 90, 91, 221, 186, 229, 126, 190, 7, 196, 213, 210,
2286 154, 41, 118, 115, 186, 154, 58, 246, 122, 78, 70, 227, 84, 183, 212, 109, 167, 222, 37,
2287 35, 57, 30, 200, 156, 142, 145, 201, 36, 38, 19, 59, 212, 151, 67, 179, 178, 28, 134, 90,
2288 64, 196, 74, 234, 204, 41, 107, 51, 191, 172, 197, 19, 125, 214, 20, 200, 127, 5, 242, 251,
2289 66, 60, 160, 71, 37, 53, 105, 190, 161, 20, 17, 185, 81, 220, 225, 193, 248, 225, 46, 251,
2290 187, 228, 247, 101, 81, 141, 255, 46, 249, 91, 163, 185, 30, 185, 97, 48, 115, 250, 122,
2291 208, 237, 91, 194, 160, 81, 221, 12, 250, 122, 165, 57, 139, 132, 209, 220, 10, 226, 191,
2292 231, 244, 187, 251, 239, 84, 74, 101, 3, 159, 52, 143, 16, 255, 174, 249, 96, 77, 135, 141,
2293 96, 214, 147, 172, 74, 252, 73, 134, 57, 113, 61, 193, 170, 217, 230, 164, 61, 20, 163,
2294 214, 56, 124, 90, 117, 5, 69, 39, 10, 164, 173, 201, 150, 210, 53, 181, 150, 174, 4, 243,
2295 174, 105, 173, 134, 134, 86, 215, 238, 23, 45, 53, 208, 156, 81, 99, 97, 116, 172, 213,
2296 114, 40, 142, 54, 174, 229, 149, 188, 185, 190, 238, 9, 81, 89, 51, 69, 215, 182, 38, 134,
2297 107, 212, 108, 93, 146, 203, 35, 210, 197, 158, 111, 13, 6, 194, 122, 57, 18, 91, 226, 64,
2298 138, 106, 166, 160, 77, 6, 210, 66, 232, 17, 165, 52, 242, 29, 113, 220, 183, 212, 145,
2299 229, 170, 158, 82, 171, 140, 251, 110, 191, 107, 47, 38, 93, 59, 90, 143, 103, 147, 173,
2300 166, 232, 237, 158, 229, 106, 221, 112, 210, 115, 231, 234, 115, 135, 76, 126, 199, 22,
2301 213, 113, 99, 210, 28, 62, 68, 79, 182, 226, 174, 201, 179, 230, 214, 67, 107, 101, 217,
2302 238, 141, 19, 40, 125, 67, 80, 227, 46, 50, 255, 61, 31, 119, 199, 195, 137, 182, 49, 141,
2303 167, 40, 108, 255, 238, 169, 189, 127, 140, 171, 255, 124, 232, 206, 102, 155, 95, 102,
2304 235, 95, 255, 186, 123, 12, 122, 225, 116, 34, 213, 131, 141, 222, 105, 253, 90, 180, 199,
2305 255, 248, 167, 80, 119, 126, 215, 36, 177, 94, 173, 53, 167, 139, 142, 242, 115, 48, 155,
2306 141, 204, 91, 219, 239, 205, 31, 154, 119, 198, 248, 246, 121, 118, 59, 239, 135, 146, 166,
2307 248, 255, 26, 119, 234, 99, 233, 229, 241, 103, 125, 182, 157, 107, 178, 225, 63, 85, 158,
2308 22, 97, 224, 222, 254, 187, 62, 115, 111, 170, 139, 162, 19, 164, 95, 91, 168, 130, 42, 60,
2309 10, 254, 198, 20, 39, 171, 129, 168, 86, 250, 247, 213, 165, 181, 109, 5, 86, 201, 147,
2310 188, 7, 165, 221, 35, 166, 194, 176, 2, 215, 86, 180, 64, 85, 220, 7, 199, 152, 108, 137,
2311 42, 147, 116, 43, 122, 54, 26, 213, 173, 57, 183, 158, 7, 190, 98, 116, 37, 229, 153, 168,
2312 243, 182, 37, 147, 47, 156, 89, 21, 207, 92, 175, 45, 127, 45, 152, 15, 65, 197, 11, 188,
2313 218, 64, 82, 106, 131, 185, 114, 211, 121, 80, 5, 75, 94, 180, 44, 95, 188, 31, 40, 173,
2314 112, 180, 121, 41, 147, 209, 50, 25, 154, 235, 242, 64, 158, 188, 140, 137, 154, 183, 230,
2315 74, 160, 54, 60, 115, 52, 175, 89, 238, 60, 154, 154, 150, 247, 163, 187,
2316 ];
2317 let p5 = [
2319 140, 5, 0, 0, 96, 9, 64, 1, 0, 0, 0, 0, 0, 0, 19, 4, 83, 237, 234, 82, 74, 188, 6, 23, 0,
2320 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 196, 178, 112, 86, 5, 59, 97, 219, 15, 0,
2321 0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 188, 210, 59, 150, 246,
2322 167, 182, 213, 33, 0, 0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0,
2323 132, 194, 47, 23, 175, 46, 78, 138, 23, 0, 0, 0, 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128,
2324 0, 0, 0, 0, 0, 0, 136, 219, 85, 240, 191, 125, 172, 233, 10, 0, 0, 0, 6, 23, 0, 225, 138,
2325 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 80, 211, 212, 44, 191, 227, 124, 40, 13, 0, 0, 0,
2326 6, 23, 0, 225, 138, 0, 0, 0, 0, 254, 128, 0, 0, 0, 0, 0, 0, 77, 13, 19, 149, 102, 140, 134,
2327 77, 16, 0, 0, 0, 4, 83, 237, 239, 254, 225, 138, 4, 63, 87, 214, 254, 225, 138, 4, 83, 236,
2328 159, 254, 225, 138, 4, 63, 87, 46, 254, 225, 138, 4, 63, 87, 56, 128, 225, 138, 4, 63, 87,
2329 56, 123, 225, 138, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255,
2330 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255,
2331 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 4, 255, 255, 255, 255, 0, 0, 0, 0, 0, 0, 0, 0, 0,
2332 0, 0, 0, 0, 0, 16, 151, 56, 146,
2333 ];
2334 let p6 = [
2336 140, 6, 0, 0, 112, 44, 192, 2, 0, 0, 1, 0, 0, 0, 0, 0, 0, 19, 0, 0, 0, 0, 0, 0, 254, 236,
2337 189, 203, 114, 234, 202, 214, 239, 249, 157, 94, 85, 61, 198, 119, 186, 181, 79, 72, 2,
2338 188, 77, 245, 140, 145, 48, 24, 137, 137, 208, 5, 169, 162, 98, 7, 32, 246, 4, 36, 97, 77,
2339 155, 201, 173, 162, 158, 231, 188, 81, 53, 171, 89, 47, 80, 189, 211, 170, 148, 148, 194,
2340 164, 156, 96, 16, 248, 178, 214, 252, 55, 126, 177, 194, 107, 122, 88, 82, 94, 198, 200,
2341 113, 201, 204, 255, 231, 127, 252, 215, 255, 242, 31, 255, 241, 95, 254, 223, 255, 251,
2342 127, 252, 215, 255, 254, 63, 255, 199, 127, 252, 159, 255, 57, 154, 12, 166, 243, 255, 252,
2343 223, 254, 247, 255, 28, 111, 90, 147, 97, 99, 52, 237, 76, 91, 138, 185, 85, 203, 218, 125,
2344 243, 165, 57, 255, 41, 122, 189, 230, 77, 211, 23, 155, 118, 223, 83, 186, 86, 208, 236,
2345 73, 213, 65, 175, 31, 253, 82, 101, 173, 102, 90, 65, 163, 183, 213, 21, 221, 84, 234, 61,
2346 83, 119, 93, 65, 81, 172, 70, 107, 233, 25, 138, 210, 179, 149, 23, 79, 18, 205, 142, 41,
2347 8, 163, 134, 179, 177, 77, 87, 182, 131, 86, 121, 80, 106, 213, 135, 125, 179, 162, 139,
2348 122, 203, 241, 131, 138, 109, 4, 93, 51, 168, 173, 135, 74, 75, 29, 214, 29, 97, 104, 4,
2349 55, 222, 182, 86, 26, 205, 189, 242, 192, 127, 217, 90, 219, 174, 48, 40, 233, 115, 171,
2350 49, 105, 141, 109, 189, 63, 182, 91, 63, 70, 141, 167, 237, 120, 166, 148, 213, 173, 92,
2351 30, 110, 149, 78, 91, 12, 66, 93, 241, 5, 93, 208, 45, 45, 156, 56, 143, 130, 18, 90, 194,
2352 147, 104, 55, 116, 209, 8, 163, 103, 211, 94, 151, 134, 91, 189, 50, 174, 235, 191, 135,
2353 138, 213, 237, 218, 21, 211, 182, 252, 233, 191, 187, 79, 255, 141, 124, 247, 204, 237,
2354 183, 132, 129, 237, 70, 142, 164, 8, 174, 169, 136, 94, 99, 178, 28, 133, 129, 48, 38, 223,
2355 238, 61, 180, 68, 183, 183, 154, 186, 253, 201, 170, 57, 123, 90, 107, 179, 110, 169, 83,
2356 119, 214, 154, 33, 175, 219, 247, 173, 200, 109, 88, 191, 189, 6, 249, 93, 171, 38, 58,
2357 225, 58, 114, 132, 69, 48, 62, 179, 205, 58, 178, 165, 14, 164, 160, 60, 54, 215, 51, 79,
2358 90, 15, 70, 243, 192, 50, 109, 77, 84, 45, 93, 50, 229, 234, 162, 103, 180, 238, 181, 146,
2359 235, 116, 234, 218, 111, 183, 81, 233, 155, 129, 53, 177, 27, 66, 73, 123, 240, 26, 110,
2360 40, 139, 238, 180, 250, 226, 137, 74, 199, 106, 184, 27, 67, 113, 155, 142, 209, 234, 14,
2361 109, 235, 247, 72, 246, 90, 154, 31, 61, 245, 76, 209, 234, 133, 74, 223, 158, 183, 126,
2362 13, 77, 241, 87, 199, 168, 117, 134, 194, 162, 163, 7, 90, 167, 59, 183, 218, 110, 67, 40,
2363 143, 130, 32, 178, 31, 180, 208, 233, 63, 109, 123, 91, 85, 26, 223, 223, 174, 45, 163, 41,
2364 245, 30, 106, 15, 170, 18, 149, 123, 155, 170, 173, 90, 206, 148, 124, 243, 111, 39, 116,
2365 166, 157, 153, 44, 105, 245, 145, 212, 169, 255, 148, 52, 67, 169, 222, 255, 252, 111, 237,
2366 167, 127, 74, 191, 234, 254, 100, 169, 140, 42, 171, 201, 92, 173, 142, 252, 127, 77, 86,
2367 247, 77, 251, 95, 254, 175, 127, 215, 127, 223, 246, 42, 255, 26, 253, 235, 165, 108, 133,
2368 255, 246, 151, 203, 202, 211, 250, 151, 175, 12, 203, 193, 124, 221, 155, 255, 67, 126,
2369 190, 109, 253, 99, 242, 243, 95, 211, 167, 219, 187, 237, 157, 107, 254, 43, 24, 107, 81,
2370 79, 22, 171, 225, 38, 252, 183, 240, 24, 4, 255, 114, 31, 21, 213, 235, 11, 195, 81, 239,
2371 159, 253, 161, 97, 169, 222, 77, 215, 41, 173, 164, 231, 254, 68, 157, 151, 199, 131, 31,
2372 15, 51, 229, 63, 255, 215, 120, 20, 151, 181, 190, 25, 143, 98, 173, 167, 4, 37, 210, 202,
2373 110, 79, 94, 44, 109, 63, 184, 25, 204, 106, 245, 174, 165, 184, 186, 255, 34, 232, 166,
2374 85, 235, 10, 129, 108, 219, 94, 77, 55, 38, 138, 209, 88, 68, 227, 135, 64, 117, 74, 222,
2375 11, 105, 165, 138, 101, 41, 51, 50, 138, 77, 47, 168, 253, 24, 154, 254, 166, 75, 250, 197,
2376 174, 255, 220, 14, 67, 253, 183, 37, 182, 44, 199, 154, 44, 180, 173, 94, 210, 67, 119,
2377 162, 217, 238, 168, 45, 173, 75, 166, 89, 49, 221, 121, 75, 181, 252, 201, 68, 13, 106, 11,
2378 215, 174, 172, 116, 193, 151, 122, 225, 164, 101, 88, 214, 163, 99, 121, 131, 81, 24, 45,
2379 12, 251, 169, 162, 202, 149, 101, 79, 168, 52, 76, 161, 114, 239, 153, 11, 127, 104, 76,
2380 54, 182, 29, 56, 35, 201, 157, 140, 103, 90, 91, 221, 186, 229, 126, 190, 7, 196, 213, 210,
2381 154, 41, 118, 115, 186, 154, 58, 246, 122, 78, 70, 227, 84, 183, 212, 109, 167, 222, 37,
2382 35, 57, 30, 200, 156, 142, 145, 201, 36, 38, 19, 59, 212, 151, 67, 179, 178, 28, 134, 90,
2383 64, 196, 74, 234, 204, 41, 107, 51, 191, 172, 197, 19, 125, 214, 20, 200, 127, 5, 242, 251,
2384 66, 60, 160, 71, 37, 53, 105, 190, 161, 20, 17, 185, 81, 220, 225, 193, 248, 225, 46, 251,
2385 187, 228, 247, 101, 81, 141, 255, 46, 249, 91, 163, 185, 30, 185, 97, 48, 115, 250, 122,
2386 208, 237, 91, 194, 160, 81, 221, 12, 250, 122, 165, 57, 139, 132, 209, 220, 10, 226, 191,
2387 231, 244, 187, 251, 239, 84, 74, 101, 3, 159, 52, 143, 16, 255, 174, 249, 96, 77, 135, 141,
2388 96, 214, 147, 172, 74, 252, 73, 134, 57, 113, 61, 193, 170, 217, 230, 164, 61, 20, 163,
2389 214, 56, 124, 90, 117, 5, 69, 39, 10, 164, 173, 201, 150, 210, 53, 181, 150, 174, 4, 243,
2390 174, 105, 173, 134, 134, 86, 215, 238, 23, 45, 53, 208, 156, 81, 99, 97, 116, 172, 213,
2391 114, 40, 142, 54, 174, 229, 149, 188, 185, 190, 238, 9, 81, 89, 51, 69, 215, 182, 38, 134,
2392 107, 212, 108, 93, 146, 203, 35, 210, 197, 158, 111, 13, 6, 194, 122, 57, 18, 91, 226, 64,
2393 138, 106, 166, 160, 77, 6, 210, 66, 232, 17, 165, 52, 242, 29, 113, 220, 183, 212, 145,
2394 229, 170, 158, 82, 171, 140, 251, 110, 191, 107, 47, 38, 93, 59, 90, 143, 103, 147, 173,
2395 166, 232, 237, 158, 229, 106, 221, 112, 210, 115, 231, 234, 115, 135, 76, 126, 199, 22,
2396 213, 113, 99, 210, 28, 62, 68, 79, 182, 226, 174, 201, 179, 230, 214, 67, 107, 101, 217,
2397 238, 141, 19, 40, 125, 67, 80, 227, 46, 50, 255, 61, 31, 119, 199, 195, 137, 182, 49, 141,
2398 167, 40, 108, 255, 238, 169, 189, 127, 140, 171, 255, 124, 232, 206, 102, 155, 95, 102,
2399 235, 95, 255, 186, 123, 12, 122, 225, 116, 34, 213, 131, 141, 222, 105, 253, 90, 180, 199,
2400 255, 248, 167, 80, 119, 126, 215, 36, 177, 94, 173, 53, 167, 139, 142, 242, 115, 48, 155,
2401 141, 204, 91, 219, 239, 205, 31, 154, 119, 198, 248, 246, 121, 118, 59, 239, 135, 146, 166,
2402 248, 255, 26, 119, 234, 99, 233, 229, 241, 103, 125, 182, 157, 107, 178, 225, 63, 85, 158,
2403 22, 97, 224, 222, 254, 187, 62, 115, 111, 170, 139, 162, 19, 164, 95, 91, 168, 130, 42, 60,
2404 10, 254, 198, 20, 39, 171, 129, 168, 86, 250, 247, 213, 165, 181, 109, 5, 86, 201, 147,
2405 188, 7, 165, 221, 35, 166, 194, 176, 2, 215, 86, 180, 64, 85, 220, 7, 199, 152, 108, 137,
2406 42, 147, 116, 43, 122, 54, 26, 213, 173, 57, 183, 158, 7, 190, 98, 116, 37, 229, 153, 168,
2407 243, 182, 37, 147, 47, 156, 89, 21, 207, 92, 175, 45, 127, 45, 152, 15, 65, 197, 11, 188,
2408 218, 64, 82, 106, 131, 185, 114, 211, 121, 80, 5, 75, 94, 180, 44, 95, 188, 31, 40, 173,
2409 112, 180, 121, 41, 147, 209, 50, 25, 154, 235, 242, 64, 158, 188, 140, 137, 154, 183, 230,
2410 74, 160, 54, 60, 115, 52, 175, 89, 238, 60, 154, 154, 150, 247, 163, 187,
2411 ];
2412 let ps: Vec<Vec<u8>> = vec![
2413 p0.to_vec(),
2414 p1.to_vec(),
2415 p2.to_vec(),
2416 p3.to_vec(),
2417 p4.to_vec(),
2418 p5.to_vec(),
2419 p6.to_vec(),
2420 ];
2421
2422 let mut n = 0;
2423
2424 let mut rq = RecvQ::new();
2425 for i in ps {
2426 let v = FrameVec::new(&i).unwrap();
2427 for i in v.frames {
2428 rq.insert(i).unwrap();
2429 if !rq.flush(&"0.0.0.0:0".parse().unwrap()).is_empty() {
2430 n += 1;
2431 }
2432 }
2433 }
2434
2435 assert_eq!(n, 4);
2437}
2438
2439#[cfg(test)]
2440#[path = "arq_tests.rs"]
2441mod regressions;
2442
2443#[test]
2444fn rejects_messages_exceeding_the_receivers_fragment_limit() {
2445 let sendq = SendQ::new(576);
2446 let limit = crate::fragment::MAX_FRAGMENTS * (576 - 60);
2447 assert!(
2448 sendq
2449 .required_bytes(Reliability::ReliableOrdered, limit)
2450 .is_ok()
2451 );
2452 assert!(matches!(
2453 sendq.required_bytes(Reliability::ReliableOrdered, limit + 1),
2454 Err(RaknetError::PacketSizeExceedMTU)
2455 ));
2456}
2457
2458#[test]
2459fn exact_mtu_payload_is_not_marked_as_split() {
2460 let mut sendq = SendQ::new(576);
2461 let payload = vec![0xfe; 516];
2462 sendq
2463 .insert(Reliability::ReliableOrdered, &payload)
2464 .unwrap();
2465 let frames = sendq.flush(0, &"127.0.0.1:19132".parse().unwrap());
2466 assert_eq!(frames.len(), 1);
2467 assert_eq!(frames[0].flags & 16, 0);
2468 assert_eq!(frames[0].data.as_ref(), payload);
2469 sendq.ack(frames[0].sequence_number, 1);
2470 assert!(sendq.is_empty());
2471 assert_eq!(sendq.buffered_bytes, 0);
2472}
2473
2474#[cfg(test)]
2475mod allocation_regressions {
2476 use super::*;
2477
2478 #[test]
2479 fn frame_decoder_reuses_storage_and_discards_previous_frames() {
2480 let packet = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0xfe; 800]);
2481 let encoded = packet.serialize().unwrap();
2482 let mut frames = Vec::new();
2483 FrameVec::decode_into(&encoded, &mut frames).unwrap();
2484 let capacity = frames.capacity();
2485 assert_eq!(frames[0].data.as_ref(), &[0xfe; 800]);
2486 frames.clear();
2487 FrameVec::decode_into(&encoded, &mut frames).unwrap();
2488 assert_eq!(frames.len(), 1);
2489 assert_eq!(frames.capacity(), capacity);
2490 assert!(FrameVec::decode_into(&[0x84], &mut frames).is_err());
2491 assert!(frames.is_empty());
2492 }
2493
2494 #[test]
2495 fn ordered_delivery_retains_the_receive_queue_allocation() {
2496 let mut queue = RecvQ::new();
2497 let mut delivered = Vec::new();
2498 for index in 0..16 {
2499 let mut frame = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0xfe]);
2500 frame.sequence_number = index;
2501 frame.reliable_frame_index = index;
2502 frame.ordered_frame_index = index;
2503 queue.insert(frame).unwrap();
2504 queue.flush_into(&mut delivered);
2505 assert_eq!(delivered.len(), 1);
2506 delivered.clear();
2507 assert!(queue.ready_ordered.capacity() > 0);
2508 }
2509 }
2510}
2511
2512#[cfg(test)]
2513mod serialization_reuse_tests {
2514 use super::*;
2515
2516 #[test]
2517 fn reused_fragment_buffer_preserves_the_wire_format() {
2518 let mut frame = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0xfe; 16]);
2519 frame.flags |= 16;
2520 frame.sequence_number = 0xabcdef;
2521 frame.reliable_frame_index = 0x123456;
2522 frame.ordered_frame_index = 0x654321;
2523 frame.order_channel = 7;
2524 frame.compound_size = 3;
2525 frame.compound_id = 0x1234;
2526 frame.fragment_index = 1;
2527 let mut expected = vec![
2528 0x8c, 0xef, 0xcd, 0xab, 0x70, 0, 0x80, 0x56, 0x34, 0x12, 0x21, 0x43, 0x65, 7, 0, 0, 0,
2529 3, 0x12, 0x34, 0, 0, 0, 1,
2530 ];
2531 expected.extend_from_slice(&[0xfe; 16]);
2532 let mut buffer = Vec::with_capacity(128);
2533 buffer.extend_from_slice(&[0xff; 64]);
2534 for _ in 0..100 {
2535 frame.serialize_into(&mut buffer).unwrap();
2536 assert_eq!(buffer, expected);
2537 assert_eq!(buffer.capacity(), 128);
2538 }
2539 frame.compound_size = 0;
2540 assert!(frame.serialize_into(&mut buffer).is_err());
2541 assert_eq!(buffer, expected);
2542 }
2543}
2544
2545#[cfg(test)]
2546mod shared_fragment_tests {
2547 use super::*;
2548
2549 #[test]
2550 fn retransmitted_fragment_accepts_old_ack_without_releasing_the_budget_twice() {
2551 let peer = "127.0.0.1:19132".parse().unwrap();
2552 let mut queue = SendQ::new(1400);
2553 queue
2554 .insert(Reliability::ReliableOrdered, &[0xfe; 4096])
2555 .unwrap();
2556 let frames = queue.flush(0, &peer);
2557 let old_id = frames[3].sequence_number;
2558 queue.nack(old_id, 1);
2559 let retry = queue.flush(1, &peer);
2560 assert_eq!(retry.len(), 1);
2561 let new_id = retry[0].sequence_number;
2562 queue.ack(old_id, 2);
2563 let budget = queue.buffered_bytes;
2564 queue.ack(old_id, 2);
2565 queue.ack(new_id, 2);
2566 assert_eq!(queue.buffered_bytes, budget);
2567 for frame in &frames[..3] {
2568 queue.ack(frame.sequence_number, 2);
2569 }
2570 assert_eq!(queue.buffered_bytes, 0);
2571 }
2572
2573 #[test]
2574 fn fragment_slices_share_one_allocation_and_keep_its_budget_until_final_ack() {
2575 let peer = "127.0.0.1:19132".parse().unwrap();
2576 let mut queue = SendQ::new(1400);
2577 let payload = vec![0xfe; 4096];
2578 queue
2579 .insert(Reliability::ReliableOrdered, &payload)
2580 .unwrap();
2581 let frames = queue.flush(0, &peer);
2582 assert_eq!(frames.len(), 4);
2583 let owner = frames[0].shared_payload.as_ref().unwrap();
2584 for frame in &frames {
2585 assert!(Arc::ptr_eq(owner, frame.shared_payload.as_ref().unwrap()));
2586 }
2587 let restored: Vec<_> = frames
2588 .iter()
2589 .flat_map(|frame| frame.data.iter().copied())
2590 .collect();
2591 assert_eq!(restored, payload);
2592 let initial = queue.buffered_bytes;
2593 for (index, frame) in frames.iter().enumerate() {
2594 queue.ack(frame.sequence_number, 1);
2595 queue.ack(frame.sequence_number, 1);
2596 if index + 1 < frames.len() {
2597 assert_eq!(
2598 queue.buffered_bytes,
2599 initial - (index + 1) * SendQ::FRAME_BUDGET
2600 );
2601 }
2602 }
2603 assert_eq!(queue.buffered_bytes, 0);
2604 assert_eq!(owner.remaining.load(Ordering::Relaxed), 0);
2605 }
2606}
2607
2608#[cfg(test)]
2609mod flight_order_tests {
2610 use super::*;
2611
2612 #[test]
2613 fn retries_and_sequence_wrap_keep_the_existing_numeric_order() {
2614 let peer = "127.0.0.1:19132".parse().unwrap();
2615 let mut queue = SendQ::new(1400);
2616 queue.sequence_number = sequence::MASK;
2617 queue.insert(Reliability::ReliableOrdered, &[0xfe]).unwrap();
2618 queue.insert(Reliability::ReliableOrdered, &[0xfe]).unwrap();
2619 let initial = queue.flush(0, &peer);
2620 assert_eq!(
2621 initial
2622 .iter()
2623 .map(|frame| frame.sequence_number)
2624 .collect::<Vec<_>>(),
2625 [sequence::MASK, 0]
2626 );
2627 assert!(queue.flight_order_dirty);
2628 assert!(queue.flush(1, &peer).is_empty());
2629 assert_eq!(
2630 queue
2631 .sent_packet
2632 .iter()
2633 .map(|packet| packet.0.sequence_number)
2634 .collect::<Vec<_>>(),
2635 [0, sequence::MASK]
2636 );
2637 assert!(!queue.flight_order_dirty);
2638 queue.nack(sequence::MASK, 2);
2639 assert!(queue.flight_order_dirty);
2640 let retry = queue.flush(2, &peer);
2641 assert_eq!(retry.len(), 1);
2642 assert_eq!(retry[0].sequence_number, 1);
2643 assert!(!queue.flight_order_dirty);
2644 assert_eq!(
2645 queue
2646 .sent_packet
2647 .iter()
2648 .map(|packet| packet.0.sequence_number)
2649 .collect::<Vec<_>>(),
2650 [0, 1]
2651 );
2652 queue.ack(sequence::MASK, 3);
2653 queue.ack(0, 3);
2654 assert!(queue.is_empty());
2655 assert_eq!(queue.buffered_bytes, 0);
2656 }
2657}
2658
2659#[cfg(test)]
2660mod owned_decode_tests {
2661 use super::*;
2662
2663 #[test]
2664 fn a_single_frame_moves_the_udp_allocation_into_its_payload() {
2665 let frame = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0xfe; 800]);
2666 let packet = frame.serialize().unwrap();
2667 let pointer = packet.as_ptr();
2668 let mut decoded = Vec::new();
2669 FrameVec::decode_owned_into(packet, &mut decoded).unwrap();
2670 assert_eq!(decoded.len(), 1);
2671 assert_eq!(decoded[0].data.as_ptr(), pointer);
2672 assert_eq!(decoded[0].data.as_ref(), frame.data.as_ref());
2673 let data: Vec<u8> = decoded.pop().unwrap().data.into();
2674 assert_eq!(data.as_ptr(), pointer);
2675 }
2676
2677 #[test]
2678 fn multiple_frames_and_invalid_lengths_keep_the_borrowed_decoder_contract() {
2679 let first = FrameSetPacket::new(Reliability::ReliableOrdered, vec![0xfe; 800]);
2680 let second = FrameSetPacket::new(Reliability::Unreliable, vec![0xfe; 64]);
2681 let mut packet = first.serialize().unwrap();
2682 packet.extend_from_slice(&second.serialize().unwrap()[4..]);
2683 let mut expected = Vec::new();
2684 FrameVec::decode_into(&packet, &mut expected).unwrap();
2685 let mut actual = Vec::new();
2686 FrameVec::decode_owned_into(packet.clone(), &mut actual).unwrap();
2687 assert_eq!(actual.len(), 2);
2688 for (actual, expected) in actual.iter().zip(&expected) {
2689 assert_eq!(actual.serialize().unwrap(), expected.serialize().unwrap());
2690 }
2691 for length in 0..packet.len() {
2692 let mut borrowed = Vec::new();
2693 let mut owned = Vec::new();
2694 assert_eq!(
2695 FrameVec::decode_into(&packet[..length], &mut borrowed).is_ok(),
2696 FrameVec::decode_owned_into(packet[..length].to_vec(), &mut owned).is_ok()
2697 );
2698 }
2699 }
2700}
2701
2702#[cfg(test)]
2703mod scheduling_regressions {
2704 use super::*;
2705
2706 #[test]
2707 fn cached_deadlines_match_a_full_scan_across_ack_loss_and_wrap() {
2708 let peer = "127.0.0.1:19132".parse().unwrap();
2709 let mut cached = SendQ::new(1400);
2710 cached.sequence_number = sequence::MASK - 80;
2711 let mut scanned = SendQ::new(1400);
2712 scanned.sequence_number = cached.sequence_number;
2713 let mut rng = 0x12345678u32;
2714 for tick in 0..4000 {
2715 rng = rng.wrapping_mul(1664525).wrapping_add(1013904223);
2716 if rng & 3 == 0 {
2717 for queue in [&mut cached, &mut scanned] {
2718 queue
2719 .insert(Reliability::ReliableOrdered, &[0xfe, 42])
2720 .unwrap();
2721 }
2722 }
2723 if !cached.sent_packet.is_empty() && rng & 7 == 1 {
2724 let index = (rng as usize >> 8) % cached.sent_packet.len();
2725 let sequence = cached.sent_packet[index].0.sequence_number;
2726 for queue in [&mut cached, &mut scanned] {
2727 if rng & 16 == 0 {
2728 queue.ack(sequence, tick);
2729 } else {
2730 queue.nack(sequence, tick);
2731 }
2732 }
2733 }
2734 scanned.next_retry = 0;
2736 scanned.retries_pending = true;
2737 let actual: Vec<_> = cached
2738 .flush(tick, &peer)
2739 .iter()
2740 .map(|f| f.serialize().unwrap())
2741 .collect();
2742 let expected: Vec<_> = scanned
2743 .flush(tick, &peer)
2744 .iter()
2745 .map(|f| f.serialize().unwrap())
2746 .collect();
2747 assert_eq!(actual, expected, "tick {tick}");
2748 assert_eq!(cached.buffered_bytes, scanned.buffered_bytes);
2749 }
2750 }
2751
2752 #[test]
2753 fn lower_rto_invalidates_the_deadline_and_retries_keep_the_backoff() {
2754 let peer = "127.0.0.1:19132".parse().unwrap();
2755 let mut queue = SendQ::new(1400);
2756 queue.srtt = 133;
2757 queue.rto = 200;
2758 for _ in 0..2 {
2759 queue.insert(Reliability::ReliableOrdered, &[0xfe]).unwrap();
2760 }
2761 queue.flush(0, &peer);
2762 queue.ack(0, 50);
2763 assert_eq!(queue.rto, 174);
2764 assert!(queue.flush(173, &peer).is_empty());
2765 assert_eq!(queue.flush(174, &peer).len(), 1);
2766 assert!(queue.flush(434, &peer).is_empty());
2767 assert_eq!(queue.flush(435, &peer).len(), 1);
2768 queue.ack(1, 436);
2769 assert!(queue.is_empty());
2770 }
2771
2772 #[test]
2773 fn ack_storage_is_recycled_without_replaying_old_ranges() {
2774 let mut set = ACKSet::default();
2775 let mut output = Vec::new();
2776 for index in 0..100 {
2777 set.insert(index);
2778 set.take_ack_into(&mut output);
2779 assert_eq!(output, [(index, index)]);
2780 }
2781 set.take_ack_into(&mut output);
2782 assert!(output.is_empty());
2783 assert!(set.ack.capacity() > 0);
2784 }
2785}
2786
2787#[cfg(test)]
2788mod owned_payload_tests {
2789 use super::*;
2790
2791 #[test]
2792 fn owned_sends_match_borrowed_wire_bytes_and_share_retry_storage() {
2793 let peer = "127.0.0.1:19132".parse().unwrap();
2794 for mode in [
2795 Reliability::Unreliable,
2796 Reliability::UnreliableSequenced,
2797 Reliability::Reliable,
2798 Reliability::ReliableOrdered,
2799 Reliability::ReliableSequenced,
2800 ] {
2801 for size in [64, 800, 4096] {
2802 let data = bytes::Bytes::from(vec![0xfe; size]);
2803 let mut owned = SendQ::new(1400);
2804 let mut borrowed = SendQ::new(1400);
2805 let a = owned.insert_bytes(mode, &data, 7);
2806 let b = borrowed.insert_with_order_channel(mode, &data, 7);
2807 assert_eq!(a.is_ok(), b.is_ok());
2808 if a.is_err() {
2809 continue;
2810 }
2811 let frames = owned.flush(0, &peer);
2812 let expected = borrowed.flush(0, &peer);
2813 assert_eq!(
2814 frames
2815 .iter()
2816 .map(|f| f.serialize().unwrap())
2817 .collect::<Vec<_>>(),
2818 expected
2819 .iter()
2820 .map(|f| f.serialize().unwrap())
2821 .collect::<Vec<_>>()
2822 );
2823 let mut offset = 0;
2824 for frame in &frames {
2825 assert_eq!(frame.data.as_ptr(), data[offset..].as_ptr());
2826 offset += frame.data.len();
2827 }
2828 if mode != Reliability::Unreliable && mode != Reliability::UnreliableSequenced {
2829 owned.nack(frames[0].sequence_number, 1);
2830 let retry = owned.flush(1, &peer);
2831 assert_eq!(retry[0].data.as_ptr(), data.as_ptr());
2832 owned.ack_ranges(&[(0, (frames.len() - 1) as u32)], 2);
2834 assert!(owned.is_empty());
2835 assert_eq!(owned.buffered_bytes, 0);
2836 }
2837 }
2838 }
2839 }
2840}