//! WebSocket stream connection.

use std::time::Duration;

use futures::{channel::mpsc::UnboundedSender, select, stream::StreamExt, SinkExt};
use websocket_lite::{ClientBuilder, Message, Opcode};

use super::{Error, Parser, Protocol, Subscription};

/// Connects to WebSocket stream and subscribes.
pub async fn subscribe<P, I>(
    subscription: &Subscription,
    sender: UnboundedSender<I>,
) -> Result<(), Error>
where
    P: Protocol<I> + 'static,
{
    let (cmd, mut parser) = P::subscription(subscription)?;

    let mut client = ClientBuilder::new(cmd.address())?
        .async_connect()
        .await?
        .fuse();

    if let Some(commands) = cmd.commands() {
        for command in commands {
            client.send(Message::text(command)).await?;
        }
    }

    let mut interval = tokio::time::interval(Duration::from_secs(5)).fuse();

    loop {
        select! {
            _ = interval.select_next_some() => {
                trace!("Sending unsolicited ping message.");
                let _ = client.send(Message::pong("")).await?;
            },
            message = client.select_next_some() => {
                match message {
                    Ok(message) => {
                        match message.opcode() {
                            Opcode::Text => {
                                if let Some(event) = parser.parse(message.as_text()?)? {
                                    sender.unbounded_send(event)?;
                                }
                            },
                            Opcode::Ping => {
                                trace!("Received ping message, sending pong.");
                                client.send(Message::pong(message.into_data())).await?
                            },
                            Opcode::Close => {
                                trace!("WebSocket was closed.");
                                client.send(Message::close(None)).await?
                            },
                            Opcode::Pong | Opcode::Binary => (),
                        }
                    },
                    Err(err) => {
                        error!("Error: {:?}", err);
                        let _ = client.send(Message::close(None)).await;
                        break;
                    }
                }
            }
            complete => break,
        }
    }

    Ok(())
}