use std::sync::{Arc, RwLock};
use std::time::Duration;
use flatland_cli::game_session;
use flatland_client_lib::{
connect, load_client_settings, AutoNavigator, ConnectOptions, GameClient, RemoteSession,
};
use flatland_client_ui::{InputAction, UiKeyEvent};
use flatland_protocol::ChatChannel;
use tokio::sync::mpsc;
use tokio::time::{self, MissedTickBehavior};
use tracing::info;
use crate::actions::{dispatch_action, dispatch_menu_key, ActionCtx};
use crate::play::GfxPlayArgs;
const RECONNECT_INITIAL: Duration = Duration::from_secs(1);
const RECONNECT_MAX: Duration = Duration::from_secs(30);
const RECONNECT_SESSION_COOLDOWN: Duration = Duration::from_millis(500);
pub struct GfxNet {
state: Arc<RwLock<Option<flatland_client_lib::GameState>>>,
cmd_tx: mpsc::UnboundedSender<NetCmd>,
}
pub enum NetCmd {
Say(ChatChannel, String),
MenuKey(UiKeyEvent),
KeychainSelect(usize),
Action(InputAction),
Movement {
forward: f32,
strafe: f32,
vertical: f32,
sprint: bool,
},
AutoNavTo { x: f32, y: f32 },
CancelAutoNav,
StopMovement,
Esc,
Shutdown,
}
impl GfxNet {
pub fn spawn(args: GfxPlayArgs) -> anyhow::Result<Self> {
let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::<anyhow::Result<()>>(1);
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
let state = Arc::new(RwLock::new(None));
std::thread::Builder::new()
.name("flatland-gfx-net".into())
.spawn({
let state = Arc::clone(&state);
move || {
let rt = match tokio::runtime::Runtime::new() {
Ok(rt) => rt,
Err(err) => {
let _ = ready_tx.send(Err(err.into()));
return;
}
};
if let Err(err) = rt.block_on(net_thread(args, cmd_rx, state, ready_tx)) {
tracing::error!(error = %err, "gfx network thread exited");
}
}
})?;
ready_rx
.recv()
.map_err(|_| anyhow::anyhow!("gfx network thread exited before ready"))??;
Ok(Self { state, cmd_tx })
}
pub fn send(&self, cmd: NetCmd) {
let _ = self.cmd_tx.send(cmd);
}
pub fn snapshot(&self) -> flatland_client_lib::GameState {
self.state
.read()
.ok()
.and_then(|g| g.clone())
.expect("gfx network thread not connected")
}
}
impl Drop for GfxNet {
fn drop(&mut self) {
let _ = self.cmd_tx.send(NetCmd::Shutdown);
}
}
fn sync_state(state: &Arc<RwLock<Option<flatland_client_lib::GameState>>>, client: &GameClient<RemoteSession>) {
if let Ok(mut guard) = state.write() {
*guard = Some(client.state.clone());
}
}
fn auto_reconnect_enabled(no_reconnect: bool) -> bool {
if no_reconnect {
return false;
}
match std::env::var("FLATLAND_AUTO_RECONNECT").as_deref() {
Ok("0" | "false" | "FALSE" | "no" | "NO") => false,
_ => true,
}
}
async fn connect_client(opts: &ConnectOptions) -> anyhow::Result<GameClient<RemoteSession>> {
let character_id = opts.character_id;
let session = connect(opts.clone()).await?;
let mut client = GameClient::new(session);
client.state.character_id = character_id;
client.wait_until_ready().await?;
Ok(client)
}
async fn wait_for_server(
opts: &ConnectOptions,
state: &Arc<RwLock<Option<flatland_client_lib::GameState>>>,
cmd_rx: &mut mpsc::UnboundedReceiver<NetCmd>,
) -> anyhow::Result<GameClient<RemoteSession>> {
let mut backoff = RECONNECT_INITIAL;
let mut attempt = 0u32;
loop {
attempt += 1;
if let Ok(mut guard) = state.write() {
if let Some(s) = guard.as_mut() {
s.push_log(format!(
"Reconnecting to {} (attempt {attempt})…",
opts.server
));
}
}
match connect_client(opts).await {
Ok(client) => return Ok(client),
Err(err) => {
if let Ok(mut guard) = state.write() {
if let Some(s) = guard.as_mut() {
s.push_log(format!(
"Reconnect failed: {err} — retry in {:.1}s",
backoff.as_secs_f32()
));
}
}
}
}
let sleep = tokio::time::sleep(backoff);
tokio::pin!(sleep);
tokio::select! {
_ = &mut sleep => {}
cmd = cmd_rx.recv() => {
match cmd {
Some(NetCmd::Shutdown) | None => {
anyhow::bail!("quit while reconnecting");
}
Some(_) => {}
}
}
}
backoff = (backoff * 2).min(RECONNECT_MAX);
}
}
async fn net_thread(
args: GfxPlayArgs,
mut cmd_rx: mpsc::UnboundedReceiver<NetCmd>,
state: Arc<RwLock<Option<flatland_client_lib::GameState>>>,
ready_tx: std::sync::mpsc::SyncSender<anyhow::Result<()>>,
) -> anyhow::Result<()> {
let auto_reconnect = auto_reconnect_enabled(args.no_reconnect);
let opts = game_session::prepare_remote_play(args.name.as_deref(), args.server).await?;
let mut client = match connect_client(&opts).await {
Ok(c) => c,
Err(err) => {
let _ = ready_tx.send(Err(err));
return Ok(());
}
};
client.state.logs.retain(|line| {
!line.starts_with("gfx:") && !line.contains("not wired")
});
client
.state
.push_log(format!("Gfx client v{} ready.", env!("CARGO_PKG_VERSION")));
info!(entity_id = client.entity_id(), "gfx connected — ? help, Ctrl+C quit");
sync_state(&state, &client);
ready_tx
.send(Ok(()))
.map_err(|_| anyhow::anyhow!("gfx play loop not waiting for ready"))?;
let keys = load_client_settings().keys;
'sessions: loop {
let mut was_moving = false;
let mut intent = time::interval(Duration::from_millis(100));
intent.set_missed_tick_behavior(MissedTickBehavior::Skip);
intent.tick().await;
let mut drain = time::interval(Duration::from_millis(33));
drain.set_missed_tick_behavior(MissedTickBehavior::Skip);
drain.tick().await;
let mut move_forward = 0.0f32;
let mut move_strafe = 0.0f32;
let mut move_vertical = 0.0f32;
let mut move_sprint = false;
let mut auto_nav: Option<AutoNavigator> = None;
let mut block_active = false;
let mut last_move_err: Option<String> = None;
let mut shutdown = false;
'session: loop {
tokio::select! {
cmd = cmd_rx.recv() => {
let Some(cmd) = cmd else {
shutdown = true;
break 'session;
};
match cmd {
NetCmd::Shutdown => {
shutdown = true;
break 'session;
}
NetCmd::Esc => {
if client.state.show_npc_chat {
let _ = client.npc_talk_close().await;
} else {
let _ = client.back_on_esc();
}
}
NetCmd::Say(channel, text) => {
if let Err(err) = client.say(channel, &text).await {
client.state.push_log(format!("Chat failed: {err}"));
}
}
NetCmd::MenuKey(key) => {
let _ = dispatch_menu_key(&mut client, key).await;
}
NetCmd::KeychainSelect(idx) => {
let n = client.state.keychain_entries().len();
if n > 0 {
client.state.keychain_menu_index = idx.min(n - 1);
}
}
NetCmd::Action(action) => {
let mut ctx = ActionCtx {
block_active: &mut block_active,
auto_nav: &mut auto_nav,
};
dispatch_action(&mut client, &keys, action, &mut ctx).await;
}
NetCmd::Movement { forward, strafe, vertical, sprint } => {
move_forward = forward;
move_strafe = strafe;
move_vertical = vertical;
move_sprint = sprint;
}
NetCmd::AutoNavTo { x, y } => {
auto_nav = AutoNavigator::plan(&client.state, x, y);
}
NetCmd::CancelAutoNav => {
auto_nav = None;
}
NetCmd::StopMovement => {
move_forward = 0.0;
move_strafe = 0.0;
move_vertical = 0.0;
auto_nav = None;
if was_moving {
if let Err(err) = client.stop().await {
client.state.push_log(format!("Stop failed: {err}"));
}
was_moving = false;
}
}
}
client.drain_events();
sync_state(&state, &client);
if !client.state.connected {
if let Some(reason) = &client.state.disconnect_reason {
info!(%reason, "gfx session ended");
} else {
info!("gfx session ended (server disconnected)");
}
break 'session;
}
}
_ = drain.tick() => {
client.drain_events();
sync_state(&state, &client);
if !client.state.connected {
if let Some(reason) = &client.state.disconnect_reason {
info!(%reason, "gfx session ended");
} else {
info!("gfx session ended (server disconnected)");
}
break 'session;
}
}
_ = intent.tick() => {
let (px, py, pz) = client.state.player_position_with_z();
let (forward, strafe, vertical, sprint) = if let Some(ref mut nav) = auto_nav {
match nav.steer(px, py, pz, &client.state) {
Some(v) => v,
None => {
auto_nav = None;
client.state.push_log("Arrived at map target.");
(0.0, 0.0, 0.0, false)
}
}
} else {
(move_forward, move_strafe, move_vertical, move_sprint)
};
let moving = forward.abs() > f32::EPSILON
|| strafe.abs() > f32::EPSILON
|| vertical.abs() > f32::EPSILON;
if moving {
if let Err(err) = client.move_by(forward, strafe, vertical, sprint).await {
let msg = format!("Move failed: {err}");
if last_move_err.as_deref() != Some(&msg) {
client.state.push_log(msg.clone());
last_move_err = Some(msg);
}
} else {
last_move_err = None;
}
was_moving = true;
} else if was_moving {
if let Err(err) = client.stop().await {
client.state.push_log(format!("Stop failed: {err}"));
}
was_moving = false;
}
client.drain_events();
sync_state(&state, &client);
if !client.state.connected {
if let Some(reason) = &client.state.disconnect_reason {
info!(%reason, "gfx session ended");
} else {
info!("gfx session ended (server disconnected)");
}
break 'session;
}
}
}
}
let shutdown_requested = shutdown;
client.disconnect();
if shutdown_requested || !auto_reconnect {
break 'sessions;
}
info!("gfx disconnected — reconnecting");
if let Ok(mut guard) = state.write() {
if let Some(s) = guard.as_mut() {
s.connected = false;
s.push_log("Disconnected — reconnecting…");
}
}
drop(client);
tokio::time::sleep(RECONNECT_SESSION_COOLDOWN).await;
client = match wait_for_server(&opts, &state, &mut cmd_rx).await {
Ok(new_client) => {
new_client
}
Err(err) => {
tracing::warn!(error = %err, "gfx reconnect aborted");
break 'sessions;
}
};
client.close_overlays();
client
.state
.push_log(format!("Reconnected as entity {}.", client.entity_id()));
info!(entity_id = client.entity_id(), "gfx reconnected");
sync_state(&state, &client);
}
if let Ok(mut guard) = state.write() {
*guard = None;
}
Ok(())
}