literustlib_server 0.4.0

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

pub struct DataSender<D: literustlib::packet::PacketData> {
    packager: literustlib::serdes::PacketPackager,
    void_sequence: std::sync::atomic::AtomicU16,
    _d: std::marker::PhantomData<fn() -> D>,
}

impl <D: literustlib::packet::PacketData> DataSender<D> {
    pub fn new(mtu: u16) -> Self {
        Self {
            packager: literustlib::serdes::PacketPackager::new(mtu),
            void_sequence: std::sync::atomic::AtomicU16::new(0),
            _d: Default::default(),
        }
    }

    pub async fn send_merged(&self, data: &[(literustlib::packet::Property, D)], conn: &super::Connection<D>) -> std::io::Result<usize> {
        /*
        let mut start_i = 0;
        while start_i + 1 < packet.data.len() {
            let size = u16::from_le_bytes([packet.data[start_i], packet.data[start_i + 1]]);
            start_i += 2;
            let end = start_i + size as usize;
            if end > packet.data.len() {
                log::warn!("Merged packet tried to read more than the packet");
                break;
            }
            self.handle_packet(&packet.data[start_i..end], from).await;
            start_i = end;
        }
        */
        let mut merged_bytes = bytes::BytesMut::new();
        let mut buffer = Vec::new();
        for (prop, datum) in data.iter() {
            let bytes = datum.dump();
            let channel_opt = conn.channel_by_prop(*prop);
            let sequence = if let Some(chann) = &channel_opt {
                &chann.local_sequence
            } else {
                &self.void_sequence
            };
            let packets = self.packager.package(bytes, *prop, sequence);
            if packets.len() != 1 {
                return Err(std::io::Error::new(std::io::ErrorKind::StorageFull, "Cannot merge already-fragmented packet"));
            }
            let mut writer = std::io::Cursor::new(&mut buffer);
            let size = packets[0].dump(&mut writer)?;
            #[cfg(debug_assertions)]
            assert_eq!(size, buffer.len(), "Total dumped bytes and total written bytes do not match!");
            merged_bytes.extend_from_slice(&(size as u16).to_le_bytes());
            merged_bytes.extend_from_slice(buffer.as_slice());
            buffer.clear();
        }
        log::trace!("Packet merge buffer grew to {} by the end", buffer.capacity());
        let merged_bytes = merged_bytes.freeze();
        self.send_to(merged_bytes, literustlib::packet::Property::Merged, conn).await
        //Ok(merged_bytes.len())
    }

    pub async fn send_data(&self, data: D, property: literustlib::packet::Property, conn: &super::Connection<D>) -> std::io::Result<usize> {
        let data = data.dump();
        self.send_to(data, property, conn).await
    }

    pub(crate) async fn send_to(&self, data: bytes::Bytes, property: literustlib::packet::Property, conn: &super::Connection<D>) -> std::io::Result<usize> {
        #[cfg(debug_assertions)]
        let mut total_packet_bytes = 0;
        let mut total_sent_bytes = 0;
        let socket = conn.socket.clone();
        let addr = conn.addr;
        let mut channel_opt = conn.channel_by_prop(property);
        let sequence = if let Some(chann) = &channel_opt {
            &chann.local_sequence
        } else {
            &self.void_sequence
        };
        let packets = self.packager.package(data, property, sequence);
        let mut buf = Vec::new();
        for packet in packets {
            if let Some(chann) = &mut channel_opt {
                let mut chann_lock = chann.pending.lock().await;
                let is_window_full = chann.is_window_full(&chann_lock);

                if chann.is_reliable() {
                    chann_lock.push_back(packet.clone());
                }
                if is_window_full {
                    continue;
                }
            }
            #[cfg(debug_assertions)]
            let total_dumped_bytes_now;
            #[cfg(debug_assertions)]
            {
                total_dumped_bytes_now = packet.dump(&mut buf)?;
                total_packet_bytes += total_dumped_bytes_now;
            }
            #[cfg(not(debug_assertions))]
            {
                packet.dump(&mut buf)?;
            }
            log::trace!("Sending packet to {} ({}b) {:?}", addr, buf.len(), buf.as_slice());
            let total_sent_bytes_now = socket.send_to(&buf, addr).await?;
            #[cfg(debug_assertions)]
            assert_eq!(total_sent_bytes_now, total_dumped_bytes_now);
            total_sent_bytes += total_sent_bytes_now;
            buf.clear();
        }
        #[cfg(debug_assertions)]
        assert_eq!(total_sent_bytes, total_packet_bytes, "Total sent and total packet bytes do not match!");
        Ok(total_sent_bytes)
    }

    // assumed to be lockless
    pub(crate) async fn raw_send_to(&self, packet: &literustlib::packet::Packet, conn: &super::Connection<D>) -> std::io::Result<usize> {
        let mut buf = Vec::new();
        let total_packet_bytes = packet.dump(&mut buf)?;
        log::trace!("Sending raw packet to {} ({}) {:?}", conn.addr, buf.len(), buf.as_slice());
        let total_sent_bytes = conn.socket.send_to(&buf, conn.addr).await?;
        assert_eq!(total_sent_bytes, total_packet_bytes);
        Ok(total_sent_bytes)
    }
}