use std::marker::Unpin;
use crate::codec::Codec;
use crate::{ClientMessage, Envelope, PostcardCodec, ServerMessage, PROTOCOL_VERSION};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
const MAX_FRAME_BYTES: usize = 1_048_576;
pub const FRAME_MAGIC: [u8; 4] = *b"FL3\x01";
#[derive(Debug, thiserror::Error)]
pub enum FrameError {
#[error("frame too large: {0} bytes")]
TooLarge(usize),
#[error("frame magic mismatch: expected FL3\\x01, got {got:?} (TCP frame read was interrupted — file a bug)")]
BadMagic { got: [u8; 4] },
#[error("io error: {0}")]
Io(#[from] std::io::Error),
#[error("codec error: {0}")]
Codec(#[from] crate::codec::CodecError),
#[error("protocol version mismatch: expected {PROTOCOL_VERSION}, got {0}")]
VersionMismatch(u16),
}
pub async fn write_client_message<W>(writer: &mut W, message: &ClientMessage) -> Result<(), FrameError>
where
W: AsyncWrite + Unpin + Send,
{
write_payload(writer, &Envelope::new(message.clone())).await
}
pub async fn read_client_message<R>(reader: &mut R) -> Result<ClientMessage, FrameError>
where
R: AsyncRead + Unpin + Send,
{
let envelope: Envelope<ClientMessage> = read_payload(reader).await?;
Ok(envelope.payload)
}
pub async fn write_server_message<W>(writer: &mut W, message: &ServerMessage) -> Result<(), FrameError>
where
W: AsyncWrite + Unpin + Send,
{
write_payload(writer, &Envelope::new(message.clone())).await
}
pub async fn read_server_message<R>(reader: &mut R) -> Result<ServerMessage, FrameError>
where
R: AsyncRead + Unpin + Send,
{
let envelope: Envelope<ServerMessage> = read_payload(reader).await?;
Ok(envelope.payload)
}
async fn write_payload<W, T>(writer: &mut W, envelope: &Envelope<T>) -> Result<(), FrameError>
where
W: AsyncWrite + Unpin + Send,
T: serde::Serialize,
{
let codec = PostcardCodec;
let bytes = codec.encode(envelope)?;
if bytes.len() > MAX_FRAME_BYTES {
return Err(FrameError::TooLarge(bytes.len()));
}
let len = u32::try_from(bytes.len()).map_err(|_| FrameError::TooLarge(bytes.len()))?;
writer.write_all(&FRAME_MAGIC).await?;
writer.write_u32_le(len).await?;
writer.write_all(&bytes).await?;
writer.flush().await?;
Ok(())
}
async fn read_payload<R, T>(reader: &mut R) -> Result<Envelope<T>, FrameError>
where
R: AsyncRead + Unpin + Send,
T: serde::de::DeserializeOwned,
{
let mut magic = [0_u8; 4];
reader.read_exact(&mut magic).await?;
if magic != FRAME_MAGIC {
return Err(FrameError::BadMagic { got: magic });
}
let len = reader.read_u32_le().await? as usize;
if len > MAX_FRAME_BYTES {
return Err(FrameError::TooLarge(len));
}
let mut buf = vec![0_u8; len];
reader.read_exact(&mut buf).await?;
let codec = PostcardCodec;
let envelope: Envelope<T> = codec.decode(&buf)?;
if envelope.protocol_version != PROTOCOL_VERSION {
return Err(FrameError::VersionMismatch(envelope.protocol_version));
}
Ok(envelope)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::types::EntityState;
use crate::{AuthCredential, ClientMessage, Hello, PROTOCOL_VERSION};
use uuid::Uuid;
#[tokio::test]
async fn roundtrip_client_message() {
let msg = ClientMessage::Hello(Hello {
client_name: "test".into(),
protocol_version: PROTOCOL_VERSION,
auth: Default::default(),
character_id: None,
});
let mut buf = Vec::new();
write_client_message(&mut buf, &msg).await.unwrap();
let parsed = read_client_message(&mut buf.as_slice()).await.unwrap();
assert_eq!(msg, parsed);
}
#[tokio::test]
async fn roundtrip_full_world_tick() {
use crate::types::{
PlayerSkills, PlayerVitals, PrimaryAttributes, ResourceNodeState, ResourceNodeView,
ServerMessage, TickDelta, Transform, Velocity2D, WorldClock, WorldCoord,
};
let delta = TickDelta {
tick: 42,
entities: vec![EntityState {
id: 1,
label: "Traveler".into(),
transform: Transform {
position: WorldCoord::surface(128.0, 128.0),
yaw: 0.0,
velocity: Velocity2D { vx: 0.0, vy: 0.0 },
},
vitals: Some(PlayerVitals::default()),
attributes: Some(PrimaryAttributes::default()),
skills: Some(PlayerSkills::default()),
inside_building: None,
}],
resource_nodes: vec![ResourceNodeView {
id: "spawn-oak-1".into(),
label: "Oak tree".into(),
x: 126.0,
y: 134.0,
z: 0.0,
item_template: "oak_log".into(),
state: ResourceNodeState::Available,
blocking: true,
blocking_radius_m: 0.8,
}],
buildings: Vec::new(),
doors: Vec::new(),
npcs: Vec::new(),
inventory: Vec::new(),
blueprints: Vec::new(),
world_clock: WorldClock::default(),
ground_drops: Vec::new(),
placed_containers: Vec::new(),
combat: None,
};
let msg = ServerMessage::Tick(delta);
let mut buf = Vec::new();
write_server_message(&mut buf, &msg).await.unwrap();
let parsed = read_server_message(&mut buf.as_slice()).await.unwrap();
assert_eq!(msg, parsed);
}
#[tokio::test]
async fn roundtrip_session_auth_hello() {
let msg = ClientMessage::Hello(Hello {
client_name: "traveler".into(),
protocol_version: PROTOCOL_VERSION,
auth: AuthCredential::Session {
token: "flat_sess_test".into(),
},
character_id: Some(Uuid::parse_str("019f399e-2b4e-7001-87a2-2e99c005e79d").unwrap()),
});
let mut buf = Vec::new();
write_client_message(&mut buf, &msg).await.unwrap();
let parsed = read_client_message(&mut buf.as_slice()).await.unwrap();
assert_eq!(msg, parsed);
}
#[tokio::test]
async fn rejects_garbage_without_length_desync() {
let mut buf = b"GET /health HTTP/1.1\r\n".as_slice();
let err = read_client_message(&mut buf).await.unwrap_err();
assert!(matches!(err, FrameError::BadMagic { .. }));
}
#[tokio::test]
async fn multiplexed_frames_stay_aligned() {
use crate::types::{TickDelta, Transform, WorldClock, WorldCoord};
use crate::{Intent, ServerMessage};
let (mut client_io, mut server_io) = tokio::io::duplex(64 * 1024);
for seq in 1..=200u32 {
write_client_message(
&mut client_io,
&ClientMessage::Intent(Intent::Stop {
entity_id: 1,
seq,
}),
)
.await
.unwrap();
write_server_message(
&mut server_io,
&ServerMessage::Tick(TickDelta {
tick: seq as u64,
entities: vec![EntityState {
id: 1,
label: "p".into(),
transform: Transform {
position: WorldCoord::surface(128.0, 128.0),
yaw: 0.0,
velocity: crate::types::Velocity2D { vx: 0.0, vy: 0.0 },
},
vitals: None,
attributes: None,
skills: None,
inside_building: None,
}],
resource_nodes: vec![],
buildings: vec![],
doors: vec![],
npcs: vec![],
inventory: vec![],
blueprints: vec![],
world_clock: WorldClock::default(),
ground_drops: vec![],
placed_containers: vec![],
combat: None,
}),
)
.await
.unwrap();
}
for seq in 1..=200u32 {
let intent = read_client_message(&mut server_io).await.unwrap();
assert_eq!(
intent,
ClientMessage::Intent(Intent::Stop {
entity_id: 1,
seq,
})
);
let ServerMessage::Tick(delta) = read_server_message(&mut client_io).await.unwrap() else {
panic!("expected tick");
};
assert_eq!(delta.tick, seq as u64);
}
}
}