fws_actors 0.1.0

use bytes::Bytes;
use fastwebsockets::{handshake, WebSocket};

use http_body_util::Empty;
use hyper::{body::Incoming, header::{CONNECTION, UPGRADE}, upgrade::Upgraded, Request, Response};

use hyper_util::rt::TokioIo;

use tokio::net::TcpStream;
use url::Url;

use anyhow::{Error, Result};

struct SpawnExecutor;

impl<Fut> hyper::rt::Executor<Fut> for SpawnExecutor
    where Fut: Future + Send + 'static,
          Fut::Output: Send + 'static
{

    //I presume this will be called (if at all) in a scope after an EnterGuard has been created via a Tokio Runtime or Handle.

    //https://docs.rs/tokio/latest/tokio/runtime/struct.EnterGuard.html

    fn execute(&self, fut: Fut)
    {

        tokio::task::spawn(fut);
        
    }

}

pub async fn connect_to_server(host_and_port_buffer: &mut String, url: &Url) -> Result<(WebSocket<TokioIo<Upgraded>>, Response<Incoming>)>
{

    //Get from cache...

    //let mut host_and_port = String::with_capacity(100); //(50);

    host_and_port_buffer.clear();

    match url.host_str()
    {

        Some(the_host) =>
        {

            host_and_port_buffer.push_str(the_host);

        }
        None =>
        {

            return Result::Err(Error::msg("Host section not found in the provided URL."));

        }

    }

    match url.port()
    {

        Some(port) =>
        {

            host_and_port_buffer.push(':');

            host_and_port_buffer.push_str(&port.to_string());

        }
        None =>
        {

            //Assume port 80 if no port number has been provided as part of the URL. 

            host_and_port_buffer.push_str(":80");

        }

    }

    let host_and_port_buffer_str = host_and_port_buffer.as_str();

    //May just have to be host and port number...

    let connection_stream = TcpStream::connect(host_and_port_buffer_str).await?; //&*host_and_port_buffer).await?; //&host_and_port).await?; //url.as_str()).await?; //&host_and_port).await?;

    let request = Request::builder()
        .method("GET")
        .uri(url.as_str())
        .header("Host", host_and_port_buffer_str) //&*host_and_port_buffer) //host_str) //host_and_port)
        .header(UPGRADE, "websocket")
        .header(CONNECTION, "upgrade")
        .header("Sec-WebSocket-Key", handshake::generate_key())
        .header("Sec-WebSocket-Version", "13") //"Sec-WebSocket-Verion" OMG!!!!!!
        .body(Empty::<Bytes>::new())?;

    let (ws, res) = handshake::client(&SpawnExecutor, request, connection_stream).await?;

    Ok((ws, res))

}

/*

async fn prepare_for_new_connection_and_connect(&mut self, url: SendableText) -> InternalConnectionResult<WebSocketWriteHalf> //FswWebSocketWriterActorInternalMessage //CLEROrConnected //String) -> CLEROrConnected
{

    //Check if a zero length String has been provided for the connection URL.

    if url.is_empty()
    {

        return InternalConnectionResult::Err(self.on_non_connection_error(NonConnectionError::EmptyUrlProvided).await);

        /*
        let ided_message = self.connection_state_id.connection_state_message(WebSocketActorOutputMessage::NonConnectionError(NonConnectionError::EmptyUrlProvided)); //(SendableText::Str(ERROR_EMPTY_URL_PROVIDED)));

        //let counted_message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::ClientMessage(WebSocketActorOutputClientMessage::ConnectionError(SendableText::Str(ERROR_EMPTY_URL_PROVIDED))));

        if let Err(_) = self.io_server.output_sender_ref().send(ided_message).await
        //if let Err(_) = self.read_frame_processor_actor_input_sender.send(counted_message).await
        {

            return FswWebSocketWriterActorInternalMessage::ActorClientSenderClosed;

            //return CLEROrConnected::CLER(Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped));

        }

        return CLEROrConnected::CLER(Some(ConnectedLoopExitReason::InvalidInput));
        */

    }

    let parsed_url;
    
    match Url::parse(&url)
    {

        Ok(res) =>
        {

            parsed_url = res;

        }
        Err(err) =>
        {

            return Err(self.on_non_connection_error(NonConnectionError::UrlParseError(Arc::from(err.to_string()))).await);
            
            /*
            let ided_message = self.connection_state_id.connection_state_message(WebSocketActorOutputMessage::NonConnectionError(NonConnectionError::UrlParseError(Arc::from(err.to_string()))));

            //let counted_message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::ClientMessage(WebSocketActorOutputClientMessage::NotConnected(SendableText::String(err.to_string()))));

            if let Err(_) = self.io_server.output_sender_ref().send(ided_message).await
            //if let Err(_) = self.read_frame_processor_actor_input_sender.send(counted_message).await
            {
    
                return CLEROrConnected::CLER(Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped));
    
            }

            return CLEROrConnected::CLER(Some(ConnectedLoopExitReason::InvalidInput));
            */

        }

    }

    match self.connect_to_server(&parsed_url).await
    {

        Ok(res) => 
        {

            //Increment the connection_state_id

            let current_id = self.connection_state_id.next();

            //Split the stream here

            let (read, write) = res.0.split(tokio::io::split);

            //Setup the read-half of the websocket stream in the WebSocketReader and send it to the WebSocketClientReaderActor.

            //The WebSocketClientReaderActor is responsible for notifying WebSocketClientActor message receivers that a new connection has been established.

            let reader = WebSocketReader::FragmentCollectorRead(FragmentCollectorRead::new(read));

            let ided_message = current_id.connection_state_message(ReadWebSocketActorInputMessage::NewConnection(reader));

            if let Err(_) = self.read_web_socket_actor_io_client.input_sender_ref().send(ided_message).await //::Connected(reader)).await
            {

                return Err(WebSocketWriterActorInternalMessage::WebSocketClientReaderActorChannelSenderClosed); //ActorClientSenderClosed);

                //return CLEROrConnected::CLER(Some(ConnectedLoopExitReason::ReadWebSocketActorDropped));

            }

            self.url = Some(parsed_url);

            //Connected!

            /*
            let ided_message = current_id.connection_state_message(WebSocketActorOutputMessage::NewConnection);

            //let counted_message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::ClientMessage(WebSocketActorOutputClientMessage::ConnectionSucceed(SendableText::Str(CONNECTION_SUCCEEDED))));

            if let Err(_) = self.io_server.output_sender_ref().
            //if let Err(_) = self.read_frame_processor_actor_input_sender.send(counted_message).await
            {

                return CLEROrConnected::CLER(Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped));

            }
            */

            //Return the writer side.

            return Ok(write);

            //return CLEROrConnected::Connected(write);

        },
        Err(err) =>
        {
            
            let err_string = err.to_string();

            //Send Error message to the actor-client

            let message = WebSocketActorOutputMessage::NonConnectionError(NonConnectionError::HandshakeError(Arc::from(err_string)));

            let counted_message = self.connection_state_id.connection_state_message(message);

            //let counted_message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::ClientMessage(WebSocketActorOutputClientMessage::ConnectionError(SendableText::String(err_string))));

            if let Err(_) = self.io_server.output_sender_ref().send(counted_message).await
            //if let Err(_) = self.read_frame_processor_actor_input_sender.send(counted_message).await
            {

                return Err(WebSocketWriterActorInternalMessage::ActorClientChannelSenderClosed);

                //return CLEROrConnected::CLER(Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped));

            }

            return Err(WebSocketWriterActorInternalMessage::NonConnectionError);

            //return CLEROrConnected::CLER(Some(ConnectedLoopExitReason::ServerDisconnectedOrConnectionError));

        }

    }

}

 */