flatland-client-lib 0.1.0

use std::net::SocketAddr;

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>,
}

#[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<()> {
        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) {
        let _ = self.cmd_tx.send(ClientMessage::Disconnect);
    }
}

impl Drop for RemoteSession {
    fn drop(&mut self) {
        self.disconnect();
    }
}

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? {
        ServerMessage::Welcome(welcome) => welcome,
        other => anyhow::bail!("expected welcome, got {other:?}"),
    };

    let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<ClientMessage>();
    let (event_tx, event_rx) = mpsc::unbounded_channel::<SessionEvent>();

    let _ = event_tx.send(SessionEvent::Welcome {
        session_id: welcome.session_id,
        entity_id: welcome.entity_id,
        snapshot: welcome.snapshot,
    });

    tokio::spawn(async move {
        while let Some(msg) = cmd_rx.recv().await {
            if write_client_message(&mut writer, &msg).await.is_err() {
                break;
            }
            if matches!(msg, ClientMessage::Disconnect) {
                break;
            }
        }
        let _ = writer.shutdown().await;
    });

    tokio::spawn(async move {
        loop {
            match read_server_message(&mut reader).await {
                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::Welcome(_)) => {}
                Err(_) => {
                    let _ = event_tx.send(SessionEvent::Disconnected);
                    break;
                }
            }
        }
    });

    Ok(RemoteSession {
        session_id: welcome.session_id,
        entity_id: welcome.entity_id,
        cmd_tx,
        event_rx,
    })
}