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

flatland_client_lib/
remote.rs
