use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use flatland_protocol::{
frame::{read_server_message, write_client_message},
ClientMessage, Intent, ServerMessage,
};
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
use tokio::sync::mpsc;
use crate::session::{PlayConnection, SessionEvent};
pub struct RemoteSession {
pub session_id: flatland_protocol::SessionId,
pub entity_id: flatland_protocol::EntityId,
cmd_tx: mpsc::UnboundedSender<ClientMessage>,
event_rx: mpsc::UnboundedReceiver<SessionEvent>,
closing: Arc<AtomicBool>,
}
#[async_trait::async_trait]
impl PlayConnection for RemoteSession {
fn session_id(&self) -> flatland_protocol::SessionId {
self.session_id
}
fn entity_id(&self) -> flatland_protocol::EntityId {
self.entity_id
}
async fn submit_intent(&self, intent: Intent) -> anyhow::Result<()> {
if let Intent::Harvest {
entity_id,
ref node_id,
seq,
} = intent
{
crate::harvest_trace!(
entity_id,
node_id = %node_id,
seq,
"remote session enqueue harvest intent"
);
}
self.cmd_tx
.send(ClientMessage::Intent(intent))
.map_err(|_| anyhow::anyhow!("session closed"))?;
Ok(())
}
fn try_next_event(&mut self) -> Option<SessionEvent> {
self.event_rx.try_recv().ok()
}
async fn next_event(&mut self) -> Option<SessionEvent> {
self.event_rx.recv().await
}
fn disconnect(&self) {
self.closing.store(true, Ordering::SeqCst);
let _ = self.cmd_tx.send(ClientMessage::Disconnect);
}
}
impl Drop for RemoteSession {
fn drop(&mut self) {
if !self.closing.load(Ordering::SeqCst) {
self.disconnect();
}
}
}
fn disconnect_reason_from_io(err: &flatland_protocol::FrameError, closing: &AtomicBool) -> Option<String> {
if closing.load(Ordering::SeqCst) {
return None;
}
match err {
flatland_protocol::FrameError::Io(io)
if io.kind() == std::io::ErrorKind::ConnectionReset
|| io.kind() == std::io::ErrorKind::BrokenPipe
|| io.kind() == std::io::ErrorKind::UnexpectedEof =>
{
Some("server closed the connection".into())
}
flatland_protocol::FrameError::BadMagic { .. } => {
Some("protocol desync — restart client and server".into())
}
flatland_protocol::FrameError::TooLarge(len) => {
Some(format!(
"frame too large ({len} bytes) — stream likely desynced; restart client and server"
))
}
flatland_protocol::FrameError::Codec(inner) => {
Some(format!(
"server message decode failed: {inner} — client/server likely out of sync; \
run `make dev-restart` and restart `make dev-server` + `make dev-control-plane`"
))
}
other => Some(other.to_string()),
}
}
pub async fn connect_tcp(
addr: SocketAddr,
hello: &flatland_protocol::Hello,
) -> anyhow::Result<RemoteSession> {
let stream = TcpStream::connect(addr).await?;
let (mut reader, mut writer) = stream.into_split();
write_client_message(&mut writer, &ClientMessage::Hello(hello.clone())).await?;
let welcome = match read_server_message(&mut reader).await {
Ok(ServerMessage::Welcome(welcome)) => welcome,
Ok(other) => anyhow::bail!("expected welcome, got {other:?}"),
Err(flatland_protocol::FrameError::Codec(err)) => {
anyhow::bail!(
"codec error: {err}\n\
Client and server are out of sync — restart both after `make dev-restart`:\n\
1. make dev-stop\n\
2. make dev-control-plane (terminal 1)\n\
3. make dev-server (terminal 2)\n\
4. make dev-play NAME=…"
)
}
Err(err) => return Err(err.into()),
};
let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<ClientMessage>();
let (event_tx, event_rx) = mpsc::unbounded_channel::<SessionEvent>();
let closing = Arc::new(AtomicBool::new(false));
let _ = event_tx.send(SessionEvent::Welcome {
session_id: welcome.session_id,
entity_id: welcome.entity_id,
snapshot: welcome.snapshot,
});
let writer_events = event_tx.clone();
let closing_writer = Arc::clone(&closing);
tokio::spawn(async move {
while let Some(msg) = cmd_rx.recv().await {
if let Err(err) = write_client_message(&mut writer, &msg).await {
if !closing_writer.load(Ordering::SeqCst) {
tracing::warn!(error = %err, "client write failed");
let _ = writer_events.send(SessionEvent::Disconnected {
reason: Some(err.to_string()),
});
}
break;
}
if matches!(msg, ClientMessage::Disconnect) {
break;
}
}
let _ = writer.shutdown().await;
});
let closing_reader = Arc::clone(&closing);
tokio::spawn(async move {
loop {
match read_server_message(&mut reader).await {
Ok(ServerMessage::ContentUpdated(snapshot)) => {
if event_tx.send(SessionEvent::ContentUpdated { snapshot }).is_err() {
break;
}
}
Ok(ServerMessage::Tick(delta)) => {
if event_tx.send(SessionEvent::Tick(delta)).is_err() {
break;
}
}
Ok(ServerMessage::IntentAck {
entity_id,
seq,
tick,
}) => {
if event_tx
.send(SessionEvent::IntentAck {
entity_id,
seq,
tick,
})
.is_err()
{
break;
}
}
Ok(ServerMessage::Chat(msg)) => {
if event_tx.send(SessionEvent::Chat(msg)).is_err() {
break;
}
}
Ok(ServerMessage::HarvestResult(result)) => {
crate::harvest_trace!(
node_id = %result.node_id,
template = %result.item_template,
quantity = result.quantity,
"remote session received harvest result from TCP"
);
if event_tx.send(SessionEvent::HarvestResult(result)).is_err() {
break;
}
}
Ok(ServerMessage::CraftResult(result)) => {
if event_tx.send(SessionEvent::CraftResult(result)).is_err() {
break;
}
}
Ok(ServerMessage::Death(notice)) => {
if event_tx.send(SessionEvent::Death(notice)).is_err() {
break;
}
}
Ok(ServerMessage::Interaction(notice)) => {
if event_tx.send(SessionEvent::Interaction(notice)).is_err() {
break;
}
}
Ok(ServerMessage::ShopOpened(catalog)) => {
if event_tx.send(SessionEvent::ShopOpened(catalog)).is_err() {
break;
}
}
Ok(ServerMessage::UseResult(result)) => {
if event_tx.send(SessionEvent::UseResult(result)).is_err() {
break;
}
}
Ok(ServerMessage::Welcome(_)) => {}
Err(err) => {
let reason = disconnect_reason_from_io(&err, &closing_reader);
if let Some(ref r) = reason {
tracing::warn!(error = %r, "server read failed");
}
let _ = event_tx.send(SessionEvent::Disconnected { reason });
break;
}
}
}
});
Ok(RemoteSession {
session_id: welcome.session_id,
entity_id: welcome.entity_id,
cmd_tx,
event_rx,
closing,
})
}