1use std::marker::Unpin;
7
8use crate::codec::Codec;
9use crate::{ClientMessage, Envelope, PostcardCodec, ServerMessage, PROTOCOL_VERSION};
10use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
11
12const MAX_FRAME_BYTES: usize = 1_048_576;
13pub const FRAME_MAGIC: [u8; 4] = *b"FL3\x01";
15
16#[derive(Debug, thiserror::Error)]
17pub enum FrameError {
18 #[error("frame too large: {0} bytes")]
19 TooLarge(usize),
20 #[error("frame magic mismatch: expected FL3\\x01, got {got:?} (TCP frame read was interrupted — file a bug)")]
21 BadMagic { got: [u8; 4] },
22 #[error("io error: {0}")]
23 Io(#[from] std::io::Error),
24 #[error("codec error: {0}")]
25 Codec(#[from] crate::codec::CodecError),
26 #[error("protocol version mismatch: expected {PROTOCOL_VERSION}, got {0}")]
27 VersionMismatch(u16),
28}
29
30pub async fn write_client_message<W>(writer: &mut W, message: &ClientMessage) -> Result<(), FrameError>
31where
32 W: AsyncWrite + Unpin + Send,
33{
34 write_payload(writer, &Envelope::new(message.clone())).await
35}
36
37pub async fn read_client_message<R>(reader: &mut R) -> Result<ClientMessage, FrameError>
38where
39 R: AsyncRead + Unpin + Send,
40{
41 let envelope: Envelope<ClientMessage> = read_payload(reader).await?;
42 Ok(envelope.payload)
43}
44
45pub async fn write_server_message<W>(writer: &mut W, message: &ServerMessage) -> Result<(), FrameError>
46where
47 W: AsyncWrite + Unpin + Send,
48{
49 write_payload(writer, &Envelope::new(message.clone())).await
50}
51
52pub async fn read_server_message<R>(reader: &mut R) -> Result<ServerMessage, FrameError>
53where
54 R: AsyncRead + Unpin + Send,
55{
56 let envelope: Envelope<ServerMessage> = read_payload(reader).await?;
57 Ok(envelope.payload)
58}
59
60async fn write_payload<W, T>(writer: &mut W, envelope: &Envelope<T>) -> Result<(), FrameError>
61where
62 W: AsyncWrite + Unpin + Send,
63 T: serde::Serialize,
64{
65 let codec = PostcardCodec;
66 let bytes = codec.encode(envelope)?;
67 if bytes.len() > MAX_FRAME_BYTES {
68 return Err(FrameError::TooLarge(bytes.len()));
69 }
70 let len = u32::try_from(bytes.len()).map_err(|_| FrameError::TooLarge(bytes.len()))?;
71 writer.write_all(&FRAME_MAGIC).await?;
72 writer.write_u32_le(len).await?;
73 writer.write_all(&bytes).await?;
74 writer.flush().await?;
75 Ok(())
76}
77
78async fn read_payload<R, T>(reader: &mut R) -> Result<Envelope<T>, FrameError>
79where
80 R: AsyncRead + Unpin + Send,
81 T: serde::de::DeserializeOwned,
82{
83 let mut magic = [0_u8; 4];
84 reader.read_exact(&mut magic).await?;
85 if magic != FRAME_MAGIC {
86 return Err(FrameError::BadMagic { got: magic });
87 }
88 let len = reader.read_u32_le().await? as usize;
89 if len > MAX_FRAME_BYTES {
90 return Err(FrameError::TooLarge(len));
91 }
92 let mut buf = vec![0_u8; len];
93 reader.read_exact(&mut buf).await?;
94 let codec = PostcardCodec;
95 let envelope: Envelope<T> = codec.decode(&buf)?;
96 if envelope.protocol_version != PROTOCOL_VERSION {
97 return Err(FrameError::VersionMismatch(envelope.protocol_version));
98 }
99 Ok(envelope)
100}
101
102#[cfg(test)]
103mod tests {
104 use super::*;
105 use crate::types::EntityState;
106 use crate::{AuthCredential, ClientMessage, Hello, PROTOCOL_VERSION};
107 use uuid::Uuid;
108
109 #[tokio::test]
110 async fn roundtrip_client_message() {
111 let msg = ClientMessage::Hello(Hello {
112 client_name: "test".into(),
113 protocol_version: PROTOCOL_VERSION,
114 auth: Default::default(),
115 character_id: None,
116 });
117 let mut buf = Vec::new();
118 write_client_message(&mut buf, &msg).await.unwrap();
119 let parsed = read_client_message(&mut buf.as_slice()).await.unwrap();
120 assert_eq!(msg, parsed);
121 }
122
123 #[tokio::test]
124 async fn roundtrip_full_world_tick() {
125 use crate::types::{
126 PlayerSkills, PlayerVitals, PrimaryAttributes, ResourceNodeState, ResourceNodeView,
127 ServerMessage, TickDelta, Transform, Velocity2D, WorldClock, WorldCoord,
128 };
129
130 let delta = TickDelta {
131 tick: 42,
132 entities: vec![EntityState {
133 id: 1,
134 label: "Traveler".into(),
135 transform: Transform {
136 position: WorldCoord::surface(128.0, 128.0),
137 yaw: 0.0,
138 velocity: Velocity2D { vx: 0.0, vy: 0.0 },
139 },
140 vitals: Some(PlayerVitals::default()),
141 attributes: Some(PrimaryAttributes::default()),
142 skills: Some(PlayerSkills::default()),
143 inside_building: None,
144 tile_id: None,
145 presentation_state: None,
146 sprite_mode: None,
147 }],
148 resource_nodes: vec![ResourceNodeView {
149 id: "spawn-oak-1".into(),
150 label: "Oak tree".into(),
151 x: 126.0,
152 y: 134.0,
153 z: 0.0,
154 item_template: "oak_log".into(),
155 state: ResourceNodeState::Available,
156 blocking: true,
157 blocking_radius_m: 0.8,
158 tile_id: None,
159 sprite_mode: None,
160 presentation_state: None,
161 }],
162 buildings: Vec::new(),
163 doors: Vec::new(),
164 npcs: Vec::new(),
165 inventory: Vec::new(),
166 blueprints: Vec::new(),
167 world_clock: WorldClock::default(),
168 ground_drops: Vec::new(),
169 placed_containers: Vec::new(),
170 combat: None,
171 interior_map: None,
172 quest_log: Vec::new(),
173 interactables: Vec::new(),
174 };
175 let msg = ServerMessage::Tick(delta);
176 let mut buf = Vec::new();
177 write_server_message(&mut buf, &msg).await.unwrap();
178 let parsed = read_server_message(&mut buf.as_slice()).await.unwrap();
179 assert_eq!(msg, parsed);
180 }
181
182 #[tokio::test]
183 async fn roundtrip_session_auth_hello() {
184 let msg = ClientMessage::Hello(Hello {
185 client_name: "traveler".into(),
186 protocol_version: PROTOCOL_VERSION,
187 auth: AuthCredential::Session {
188 token: "flat_sess_test".into(),
189 },
190 character_id: Some(Uuid::parse_str("019f399e-2b4e-7001-87a2-2e99c005e79d").unwrap()),
191 });
192 let mut buf = Vec::new();
193 write_client_message(&mut buf, &msg).await.unwrap();
194 let parsed = read_client_message(&mut buf.as_slice()).await.unwrap();
195 assert_eq!(msg, parsed);
196 }
197
198 #[tokio::test]
199 async fn rejects_garbage_without_length_desync() {
200 let mut buf = b"GET /health HTTP/1.1\r\n".as_slice();
201 let err = read_client_message(&mut buf).await.unwrap_err();
202 assert!(matches!(err, FrameError::BadMagic { .. }));
203 }
204
205 #[tokio::test]
206 async fn multiplexed_frames_stay_aligned() {
207 use crate::types::{TickDelta, Transform, WorldClock, WorldCoord};
208 use crate::{Intent, ServerMessage};
209
210 let (mut client_io, mut server_io) = tokio::io::duplex(64 * 1024);
211
212 for seq in 1..=200u32 {
213 write_client_message(
214 &mut client_io,
215 &ClientMessage::Intent(Intent::Stop {
216 entity_id: 1,
217 seq,
218 }),
219 )
220 .await
221 .unwrap();
222 write_server_message(
223 &mut server_io,
224 &ServerMessage::Tick(TickDelta {
225 tick: seq as u64,
226 entities: vec![EntityState {
227 id: 1,
228 label: "p".into(),
229 transform: Transform {
230 position: WorldCoord::surface(128.0, 128.0),
231 yaw: 0.0,
232 velocity: crate::types::Velocity2D { vx: 0.0, vy: 0.0 },
233 },
234 vitals: None,
235 attributes: None,
236 skills: None,
237 inside_building: None,
238 tile_id: None,
239 presentation_state: None,
240 sprite_mode: None,
241 }],
242 resource_nodes: vec![],
243 buildings: vec![],
244 doors: vec![],
245 npcs: vec![],
246 inventory: vec![],
247 blueprints: vec![],
248 world_clock: WorldClock::default(),
249 ground_drops: vec![],
250 placed_containers: vec![],
251 combat: None,
252 interior_map: None,
253 quest_log: vec![],
254 interactables: vec![],
255 }),
256 )
257 .await
258 .unwrap();
259 }
260
261 for seq in 1..=200u32 {
262 let intent = read_client_message(&mut server_io).await.unwrap();
263 assert_eq!(
264 intent,
265 ClientMessage::Intent(Intent::Stop {
266 entity_id: 1,
267 seq,
268 })
269 );
270 let ServerMessage::Tick(delta) = read_server_message(&mut client_io).await.unwrap() else {
271 panic!("expected tick");
272 };
273 assert_eq!(delta.tick, seq as u64);
274 }
275 }
276}