flatland_client_lib/
remote.rs

1use std::net::SocketAddr;
2use std::sync::atomic::{AtomicBool, Ordering};
3use std::sync::Arc;
4
5use flatland_protocol::{
6    frame::{read_server_message, write_client_message},
7    ClientMessage, Intent, ServerMessage,
8};
9use tokio::io::AsyncWriteExt;
10use tokio::net::TcpStream;
11use tokio::sync::mpsc;
12
13use crate::session::{PlayConnection, SessionEvent};
14
15pub struct RemoteSession {
16    pub session_id: flatland_protocol::SessionId,
17    pub entity_id: flatland_protocol::EntityId,
18    cmd_tx: mpsc::UnboundedSender<ClientMessage>,
19    event_rx: mpsc::UnboundedReceiver<SessionEvent>,
20    closing: Arc<AtomicBool>,
21}
22
23#[async_trait::async_trait]
24impl PlayConnection for RemoteSession {
25    fn session_id(&self) -> flatland_protocol::SessionId {
26        self.session_id
27    }
28
29    fn entity_id(&self) -> flatland_protocol::EntityId {
30        self.entity_id
31    }
32
33    async fn submit_intent(&self, intent: Intent) -> anyhow::Result<()> {
34        self.cmd_tx
35            .send(ClientMessage::Intent(intent))
36            .map_err(|_| anyhow::anyhow!("session closed"))?;
37        Ok(())
38    }
39
40    fn try_next_event(&mut self) -> Option<SessionEvent> {
41        self.event_rx.try_recv().ok()
42    }
43
44    async fn next_event(&mut self) -> Option<SessionEvent> {
45        self.event_rx.recv().await
46    }
47
48    fn disconnect(&self) {
49        self.closing.store(true, Ordering::SeqCst);
50        let _ = self.cmd_tx.send(ClientMessage::Disconnect);
51    }
52}
53
54impl Drop for RemoteSession {
55    fn drop(&mut self) {
56        if !self.closing.load(Ordering::SeqCst) {
57            self.disconnect();
58        }
59    }
60}
61
62fn disconnect_reason_from_io(err: &flatland_protocol::FrameError, closing: &AtomicBool) -> Option<String> {
63    if closing.load(Ordering::SeqCst) {
64        return None;
65    }
66    match err {
67        flatland_protocol::FrameError::Io(io)
68            if io.kind() == std::io::ErrorKind::ConnectionReset
69                || io.kind() == std::io::ErrorKind::BrokenPipe
70                || io.kind() == std::io::ErrorKind::UnexpectedEof =>
71        {
72            Some("server closed the connection".into())
73        }
74        flatland_protocol::FrameError::BadMagic { .. } => {
75            Some("protocol desync — restart client and server".into())
76        }
77        flatland_protocol::FrameError::TooLarge(len) => {
78            Some(format!(
79                "frame too large ({len} bytes) — stream likely desynced; restart client and server"
80            ))
81        }
82        flatland_protocol::FrameError::Codec(inner) => {
83            Some(format!(
84                "server message decode failed: {inner} — client/server likely out of sync; \
85                 run `make dev-restart` and restart `make dev-server` + `make dev-control-plane`"
86            ))
87        }
88        other => Some(other.to_string()),
89    }
90}
91
92pub async fn connect_tcp(
93    addr: SocketAddr,
94    hello: &flatland_protocol::Hello,
95) -> anyhow::Result<RemoteSession> {
96    let stream = TcpStream::connect(addr).await?;
97    let (mut reader, mut writer) = stream.into_split();
98
99    write_client_message(&mut writer, &ClientMessage::Hello(hello.clone())).await?;
100
101    let welcome = match read_server_message(&mut reader).await {
102        Ok(ServerMessage::Welcome(welcome)) => welcome,
103        Ok(other) => anyhow::bail!("expected welcome, got {other:?}"),
104        Err(flatland_protocol::FrameError::Codec(err)) => {
105            anyhow::bail!(
106                "codec error: {err}\n\
107                 Client and server are out of sync — restart both after `make dev-restart`:\n\
108                   1. make dev-stop\n\
109                   2. make dev-control-plane   (terminal 1)\n\
110                   3. make dev-server          (terminal 2)\n\
111                   4. make dev-play NAME=…"
112            )
113        }
114        Err(err) => return Err(err.into()),
115    };
116
117    let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<ClientMessage>();
118    let (event_tx, event_rx) = mpsc::unbounded_channel::<SessionEvent>();
119    let closing = Arc::new(AtomicBool::new(false));
120
121    let _ = event_tx.send(SessionEvent::Welcome {
122        session_id: welcome.session_id,
123        entity_id: welcome.entity_id,
124        snapshot: welcome.snapshot,
125    });
126
127    let writer_events = event_tx.clone();
128    let closing_writer = Arc::clone(&closing);
129    tokio::spawn(async move {
130        while let Some(msg) = cmd_rx.recv().await {
131            if let Err(err) = write_client_message(&mut writer, &msg).await {
132                if !closing_writer.load(Ordering::SeqCst) {
133                    tracing::warn!(error = %err, "client write failed");
134                    let _ = writer_events.send(SessionEvent::Disconnected {
135                        reason: Some(err.to_string()),
136                    });
137                }
138                break;
139            }
140            if matches!(msg, ClientMessage::Disconnect) {
141                break;
142            }
143        }
144        let _ = writer.shutdown().await;
145    });
146
147    let closing_reader = Arc::clone(&closing);
148    tokio::spawn(async move {
149        loop {
150            match read_server_message(&mut reader).await {
151                Ok(ServerMessage::Tick(delta)) => {
152                    if event_tx.send(SessionEvent::Tick(delta)).is_err() {
153                        break;
154                    }
155                }
156                Ok(ServerMessage::IntentAck {
157                    entity_id,
158                    seq,
159                    tick,
160                }) => {
161                    if event_tx
162                        .send(SessionEvent::IntentAck {
163                            entity_id,
164                            seq,
165                            tick,
166                        })
167                        .is_err()
168                    {
169                        break;
170                    }
171                }
172                Ok(ServerMessage::Chat(msg)) => {
173                    if event_tx.send(SessionEvent::Chat(msg)).is_err() {
174                        break;
175                    }
176                }
177                Ok(ServerMessage::HarvestResult(result)) => {
178                    if event_tx.send(SessionEvent::HarvestResult(result)).is_err() {
179                        break;
180                    }
181                }
182                Ok(ServerMessage::CraftResult(result)) => {
183                    if event_tx.send(SessionEvent::CraftResult(result)).is_err() {
184                        break;
185                    }
186                }
187                Ok(ServerMessage::Death(notice)) => {
188                    if event_tx.send(SessionEvent::Death(notice)).is_err() {
189                        break;
190                    }
191                }
192                Ok(ServerMessage::Interaction(notice)) => {
193                    if event_tx.send(SessionEvent::Interaction(notice)).is_err() {
194                        break;
195                    }
196                }
197                Ok(ServerMessage::Welcome(_)) => {}
198                Err(err) => {
199                    let reason = disconnect_reason_from_io(&err, &closing_reader);
200                    if let Some(ref r) = reason {
201                        tracing::warn!(error = %r, "server read failed");
202                    }
203                    let _ = event_tx.send(SessionEvent::Disconnected { reason });
204                    break;
205                }
206            }
207        }
208    });
209
210    Ok(RemoteSession {
211        session_id: welcome.session_id,
212        entity_id: welcome.entity_id,
213        cmd_tx,
214        event_rx,
215        closing,
216    })
217}
flatland_client_lib/remote.rs

flatland_client_lib/
remote.rs
