flatland_client_lib/
remote.rs

1use std::net::SocketAddr;
2
3use flatland_protocol::{
4    frame::{read_server_message, write_client_message},
5    ClientMessage, Intent, ServerMessage,
6};
7use tokio::io::AsyncWriteExt;
8use tokio::net::TcpStream;
9use tokio::sync::mpsc;
10
11use crate::session::{PlayConnection, SessionEvent};
12
13pub struct RemoteSession {
14    pub session_id: flatland_protocol::SessionId,
15    pub entity_id: flatland_protocol::EntityId,
16    cmd_tx: mpsc::UnboundedSender<ClientMessage>,
17    event_rx: mpsc::UnboundedReceiver<SessionEvent>,
18}
19
20#[async_trait::async_trait]
21impl PlayConnection for RemoteSession {
22    fn session_id(&self) -> flatland_protocol::SessionId {
23        self.session_id
24    }
25
26    fn entity_id(&self) -> flatland_protocol::EntityId {
27        self.entity_id
28    }
29
30    async fn submit_intent(&self, intent: Intent) -> anyhow::Result<()> {
31        self.cmd_tx
32            .send(ClientMessage::Intent(intent))
33            .map_err(|_| anyhow::anyhow!("session closed"))?;
34        Ok(())
35    }
36
37    fn try_next_event(&mut self) -> Option<SessionEvent> {
38        self.event_rx.try_recv().ok()
39    }
40
41    async fn next_event(&mut self) -> Option<SessionEvent> {
42        self.event_rx.recv().await
43    }
44
45    fn disconnect(&self) {
46        let _ = self.cmd_tx.send(ClientMessage::Disconnect);
47    }
48}
49
50impl Drop for RemoteSession {
51    fn drop(&mut self) {
52        self.disconnect();
53    }
54}
55
56pub async fn connect_tcp(
57    addr: SocketAddr,
58    hello: &flatland_protocol::Hello,
59) -> anyhow::Result<RemoteSession> {
60    let stream = TcpStream::connect(addr).await?;
61    let (mut reader, mut writer) = stream.into_split();
62
63    write_client_message(&mut writer, &ClientMessage::Hello(hello.clone())).await?;
64
65    let welcome = match read_server_message(&mut reader).await? {
66        ServerMessage::Welcome(welcome) => welcome,
67        other => anyhow::bail!("expected welcome, got {other:?}"),
68    };
69
70    let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<ClientMessage>();
71    let (event_tx, event_rx) = mpsc::unbounded_channel::<SessionEvent>();
72
73    let _ = event_tx.send(SessionEvent::Welcome {
74        session_id: welcome.session_id,
75        entity_id: welcome.entity_id,
76        snapshot: welcome.snapshot,
77    });
78
79    tokio::spawn(async move {
80        while let Some(msg) = cmd_rx.recv().await {
81            if write_client_message(&mut writer, &msg).await.is_err() {
82                break;
83            }
84            if matches!(msg, ClientMessage::Disconnect) {
85                break;
86            }
87        }
88        let _ = writer.shutdown().await;
89    });
90
91    tokio::spawn(async move {
92        loop {
93            match read_server_message(&mut reader).await {
94                Ok(ServerMessage::Tick(delta)) => {
95                    if event_tx.send(SessionEvent::Tick(delta)).is_err() {
96                        break;
97                    }
98                }
99                Ok(ServerMessage::IntentAck {
100                    entity_id,
101                    seq,
102                    tick,
103                }) => {
104                    if event_tx
105                        .send(SessionEvent::IntentAck {
106                            entity_id,
107                            seq,
108                            tick,
109                        })
110                        .is_err()
111                    {
112                        break;
113                    }
114                }
115                Ok(ServerMessage::Welcome(_)) => {}
116                Err(_) => {
117                    let _ = event_tx.send(SessionEvent::Disconnected);
118                    break;
119                }
120            }
121        }
122    });
123
124    Ok(RemoteSession {
125        session_id: welcome.session_id,
126        entity_id: welcome.entity_id,
127        cmd_tx,
128        event_rx,
129    })
130}
flatland_client_lib/remote.rs

flatland_client_lib/
remote.rs
