use std::io;

use futures_util::{
    SinkExt,
    StreamExt,
    stream::{
        SplitSink,
        SplitStream,
    },
};

use tokio::{
    time::{
        sleep,
        Duration,
    },
    net::{
        TcpListener,
        TcpStream,
    },
};

use tokio_tungstenite::{
    accept_async_with_config,
    WebSocketStream,
    tungstenite::{
        Message,
        Result,
        Error as WsError,
        protocol::WebSocketConfig,
    },
};

type Tx = SplitSink<WebSocketStream<tokio::net::TcpStream>, Message>;
type Rx = SplitStream<WebSocketStream<tokio::net::TcpStream>>;

async fn accept(tcp_stream: TcpStream) -> Result<WebSocketStream<TcpStream>, WsError> {
    let wsconfig = WebSocketConfig {
        max_send_queue: Some(5),
        max_message_size: None,
        max_frame_size: None,
        accept_unmasked_frames: false,
    };
    accept_async_with_config(tcp_stream, Some(wsconfig)).await
}

async fn handle(mut rx: Rx) -> anyhow::Result<()> {
    println!("enter incoming loop");
    while let Some(msg) = rx.next().await {
        msg.map(|x| { println!("Got: {x:?}"); })?;
    }
    Ok(())
}

async fn serve(mut tx: Tx) -> anyhow::Result<()> {
    println!("enter send loop");
    let mut c = 0;
    loop {
        sleep(Duration::from_millis(1000)).await;
        c += 1;
        let txt = format!("bla bla {c}");
        println!("{txt}");
        tx.send(Message::Text(txt)).await?;
    }
}

#[tokio::main]
async fn main() -> io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;

    while let Ok((tcp_stream, addr)) = listener.accept().await {
        println!("Got a connection from: {addr}");

        let ws_result = accept(tcp_stream).await;
        let (tx, rx) = ws_result.map(|ws| ws.split()).unwrap();
        tokio::select! {
            res = handle(rx) => println!("Done waiting for messages. Reason: {res:?}"),
            _ = serve(tx) => println!("Done serving locations.")
        }
    }

    Ok(())
}