msrt 0.1.1

Portable MSRT protocol implementation.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

//! Incoming wire byte handling.

use crate::core::{Error, ErrorKind};
use crate::reliability::{Dedup, DedupDecision};
use crate::wire::{Crc16, StreamDecodeOutcome};

use crate::engine::{
    Engine, EngineOutput, ReceiveReport,
    runtime::{
        outgoing::queue_ack,
        packet::{PacketBytes, PacketDecode, decode_packet_bytes},
    },
};

pub(crate) fn receive(engine: &mut Engine, bytes: &[u8]) -> ReceiveReport {
    receive_bytes(engine, bytes)
}

fn receive_bytes(engine: &mut Engine, bytes: &[u8]) -> ReceiveReport {
    let mut input = bytes;
    let mut report = ReceiveReport::Incomplete { needed: None };

    loop {
        let outcome = match engine.ingress.feed(input, &Crc16) {
            Ok(outcome) => outcome,
            Err(error) => return ReceiveReport::Error(error),
        };
        input = &[];

        match outcome {
            StreamDecodeOutcome::Packet {
                packet_bytes,
                consumed: _,
            } => {
                let packet = match PacketBytes::try_from_slice(packet_bytes) {
                    Ok(packet) => packet,
                    Err(error) => return ReceiveReport::Error(error),
                };
                report = receive_complete_packet(engine, packet.as_bytes());

                if matches!(report, ReceiveReport::Error(_)) {
                    return report;
                }
            }
            StreamDecodeOutcome::NeedMore { additional } => {
                return match report {
                    ReceiveReport::Incomplete { .. } => {
                        ReceiveReport::Incomplete { needed: additional }
                    }
                    other => other,
                };
            }
            StreamDecodeOutcome::Noise { skipped } => {
                report = ReceiveReport::Noise { skipped };
            }
            StreamDecodeOutcome::Corrupted { consumed: _ } => {
                report = ReceiveReport::Corrupted;
            }
            StreamDecodeOutcome::Resync { skipped } => {
                report = ReceiveReport::Noise { skipped };
            }
        }
    }
}

fn receive_complete_packet(engine: &mut Engine, bytes: &[u8]) -> ReceiveReport {
    match decode_packet_bytes(bytes) {
        PacketDecode::Data(fragment) => {
            let packet_number = fragment.packet_number;
            let ack_eliciting = fragment.ack_eliciting;

            if ack_eliciting && queue_ack(engine, packet_number).is_err() {
                return ReceiveReport::Error(Error::new(ErrorKind::Engine));
            }

            if engine.dedup.observe(packet_number) == DedupDecision::Duplicate {
                return ReceiveReport::Duplicate { packet_number };
            }

            match engine.reassembly.observe(fragment, engine.now_ms) {
                Ok(Some(mut message)) => {
                    message.profile = engine.channel_profile(message.channel_id);
                    if engine.events.push(EngineOutput::Message(message)).is_err() {
                        return ReceiveReport::Error(Error::new(ErrorKind::Engine));
                    }
                    ReceiveReport::Packet { packet_number }
                }
                Ok(None) => ReceiveReport::Packet { packet_number },
                Err(error) => ReceiveReport::Error(error),
            }
        }
        PacketDecode::Ack(ack) => {
            engine.in_flight.ack_frame(ack.frame);
            ReceiveReport::Ack {
                packet_number: ack.frame.largest_acknowledged,
            }
        }
        PacketDecode::Malformed => ReceiveReport::Error(Error::malformed()),
    }
}