roowho2_lib/server/rwhod/
packet_receiver.rs

1use std::sync::Arc;
2
3use anyhow::Context;
4use chrono::{Timelike, Utc};
5use tokio::net::UdpSocket;
6
7use crate::{
8    proto::{Whod, WhodStatusUpdate},
9    server::rwhod::RwhodStatusStore,
10};
11
12pub async fn rwhod_packet_receiver_task(
13    socket: Arc<UdpSocket>,
14    whod_status_store: RwhodStatusStore,
15) -> anyhow::Result<()> {
16    let mut buf = [0u8; Whod::MAX_SIZE];
17
18    loop {
19        let (len, src) = socket.recv_from(&mut buf).await?;
20
21        tracing::debug!("Received rwhod packet of length {} bytes from {}", len, src);
22
23        if len < Whod::HEADER_SIZE {
24            tracing::error!(
25                "Received too short packet from {src}: {len} bytes (needs to be at least {} bytes)",
26                Whod::HEADER_SIZE
27            );
28            continue;
29        }
30
31        let result = Whod::from_bytes(&buf[..len])
32            .context("Failed to parse whod packet")?
33            .try_into()
34            .map(|mut status_update: WhodStatusUpdate| {
35                let timestamp = Utc::now().with_nanosecond(0).unwrap_or(Utc::now());
36                status_update.recvtime = Some(timestamp);
37                status_update
38            })
39            .map_err(|e| anyhow::anyhow!("Invalid whod packet: {}", e));
40
41        match result {
42            Ok(status_update) => {
43                tracing::debug!("Processed whod packet from {src}: {:?}", status_update);
44
45                if status_update.boot_time > status_update.sendtime {
46                    tracing::warn!(
47                        "Received whod packet from {src} with boot time {} after send time {}",
48                        status_update.boot_time,
49                        status_update.sendtime
50                    );
51                }
52
53                if let Some(recvtime) = status_update.recvtime
54                    && recvtime < status_update.sendtime
55                {
56                    tracing::warn!(
57                        "Received whod packet from {src} with recv time {} before send time {}",
58                        recvtime,
59                        status_update.sendtime
60                    );
61                }
62
63                let mut store = whod_status_store.write().await;
64                store.upsert(status_update);
65            }
66            Err(err) => {
67                tracing::error!("Error processing whod packet from {src}: {err}");
68            }
69        }
70    }
71}
roowho2_lib/server/rwhod/packet_receiver.rs

roowho2_lib/server/rwhod/
packet_receiver.rs
