fws_actors 0.1.0

use act_rs::{impl_pre_run_async, impl_post_run_async, impl_pre_and_post_run_async, impl_mac_task_actor, impl_mac_task_actor_built_state, ActorState};

use crossbeam::queue::ArrayQueue;

use fastwebsockets::{handshake, FragmentCollector, FragmentCollectorRead, Frame, OpCode, WebSocket, WebSocketError, WebSocketRead, WebSocketWrite};

//use corlib::text::SendableText;

//use gtk_estate::should_continue;

use highly_sendable::text::SendableText;

use hyper::body::Incoming;

use hyper::Response;

use libsync::std::{CountedPipelineMessage, IncrementedPipelineMessageCounter, CountedPipelineMessageMut, PipelineMessageCounter};

use libsync::{ReceiveError, ReceiveResult};

use tokio::io::{self, AsyncWriteExt, ReadHalf, WriteHalf};

use tokio::select;

use url::Url;

use std::future::Future;

use std::sync::atomic::{AtomicI32, AtomicUsize};

use std::{marker::PhantomData, sync::Arc};

use tokio::runtime::Runtime;

use std::collections::HashMap;

use tokio::time::{Duration, Instant};

use tokio::net::TcpStream;

use hyper::{Request, body::Bytes, upgrade::Upgraded, header::{UPGRADE, CONNECTION}};

use http_body_util::Empty;

use anyhow::{Error, Result};

use hyper_util::rt::TokioIo;

use tokio::task::JoinHandle;

//use crate::actors::websockets::fastwebsockets::{OwnedFrame, ReadFrameProcessorActorInputMessage};

//use crate::ReadFrameProcessorActorInputMessage;

//use super::ReadFrameProcessorActorInputMessage;

//ConnectionError,

use super::{InternalConnectionResult, NonConnectionError, OwnedFrame, ReadWebSocketActorInputMessage, ReadWebSocketActorOutputMessage, WebSocketActorInputMessage, WebSocketActorOutputMessage, WebSocketWriterActorInternalMessage}; //WebSocketClientActorInputMessage, //ConnectedOrOther,

use super::websocket_read_and_write::{WebSocketReadHalf, WebSocketWriteHalf};

//use super::{ReadFrameProcessorActorOutputMessage, WebSocketActorInputMessage, WebSocketActorOutputClientMessage};

//use crate::{ReadFrameProcessorActorOutputMessage, WebSocketActorInputMessage, WebSocketActorOutputClientMessage};

//use super::{ReadFrameProcessorActorOutputMessage, WebSocketActorInputMessage, WebSocketActorOutputClientMessage};

//use super::{WebSocketActorInputMessage, WebSocketActorOutputMessage};

use paste::paste;

use tokio::time::timeout_at;

use libsync::crossbeam::mpmc::tokio::array_queue::{Sender, Receiver, channel, io_channels::{IOClient, IOServer, io_channels, io_channel_both}};

//use crate::actors::websockets::fastwebsockets::websocket_read_and_write::WebSocketReader;

//use crate::fastwebsockets::websocket_read_and_write::WebSocketReader;

use super::WebSocketReader;

use crate::{ConnectionStateId, ConnectionStateMessage};

//use super::array_queue::{ActorIOClient, ActorIOServer, actor_io};

static CONNECTION_SUCCEEDED: &str = "Connection succeeded!";

static ERROR_EMPTY_URL_PROVIDED: &str = "Error: Empty URL provided.";

static DISCONNECTED_FROM_SERVER: &str = "Disconnected from server.";

static ERROR_NO_SERVER_CONNECTED: &str = "Error: No server connected.";

static CLOSE_FRAME_SENT: &str = "Close frame Sent.";

static TIME_ELAPSED_FORCED_CLOSURE_NOTICE: &str = "Close Connection Response Time Has Elapsed: Forcing closure of connection.";

static SERVER_DISCONNECTION_FORCED: &str = "Forced server disconnection";

static PING_FRAME_RECEIVED: &str = "Ping frame received - pong frame already sent";

static PING_FRAME_SENT: &str = "Ping frame sent.";

static PONG_FRAME_RECEIVED: &str = "Pong frame received.";

static CLOSE_FRAME_RECEIVED: &str = "Close frame received - Close frame already sent.";

/*
enum ConnectedLoopExitReason
{

    ReadFrameProcessorActorDropped,
    ReadWebSocketActorDropped,
    ServerDisconnectedOrConnectionError,
    InvalidInput

}

impl ConnectedLoopExitReason
{

    pub fn should_continue(&self) -> bool
    {

        match self
        {

            ConnectedLoopExitReason::ReadFrameProcessorActorDropped | ConnectedLoopExitReason::ReadWebSocketActorDropped => false,
            ConnectedLoopExitReason::ServerDisconnectedOrConnectionError | ConnectedLoopExitReason::InvalidInput => true,

        }

    }

}

enum ContinueOrConnected
{

    ShouldContinue(bool),
    Connected(WebSocketWriteHalf)

}

///
/// CLER: ConnectedLoopExitReason
/// 
enum CLEROrConnected
{

    CLER(Option<ConnectedLoopExitReason>),
    Connected(WebSocketWriteHalf)

}
*/

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 type FswWebSocketWriterActorInternalMessage = WebSocketWriterActorInternalMessage<WebSocketWriteHalf>;

//WriteFrameProcessorActor -> WebSocketActor <-> ReadWebSocketActor -> ReadFrameProcessorActor

//WebSocketActor -> ReadFrameProcessorActor

//WebSocketActors input queue/channel can be accessed via WriteFrameProcessorActors ActorIOClient directly, allowing it to be bypassed.

//Writing WebSocket Actor

pub struct WebSocketActorState
{

    //input_receiver: Receiver<WebSocketActorInputMessage>, //From WriteFrameProcessorActor
    //url: Option<Url>,
    //pipeline_message_counter: PipelineMessageCounter,
    //read_web_socket_actor_io_client: ActorIOClient<ReadWebSocketActorInputMessage, WebSocketActorInputMessage>, //To/From the ReadWebSocketActor (handles the read side of the stream).
    //read_web_socket_actor_io_client: IOClient<ReadWebSocketActorInputMessage, WebSocketActorInputMessage>,
    //read_frame_processor_actor_input_sender: Sender<CountedPipelineMessageMut<ReadFrameProcessorActorInputMessage>>, //To the ReadFrameProcessorActor, bypassing the associated ReadWebSocketActor for connection status notifications, erors etc.
    //host_and_port: String,
    read_web_socket_actor_io_client: IOClient<ConnectionStateMessage<ReadWebSocketActorInputMessage>, ConnectionStateMessage<ReadWebSocketActorOutputMessage>>, //web_socket_client_reader_actor_io_client: IOClient<ConnectionStateMessage<ReadWebSocketActorInputMessage>, ConnectionStateMessage<ReadWebSocketActorOutputMessage>>,
    io_server: IOServer<ConnectionStateMessage<WebSocketActorInputMessage>, ConnectionStateMessage<WebSocketActorOutputMessage>>,
    connection_state_id: ConnectionStateId
}

impl WebSocketActorState
{

    pub fn new(io_server: IOServer<ConnectionStateMessage<WebSocketActorInputMessage>, ConnectionStateMessage<WebSocketActorOutputMessage>>) -> Self
    {

        //Spawn the ReadWebSocketActor

        let read_web_socket_actor_io_client = WebSocketReaderActorState::spawn(io_server.output_sender_ref());

        Self
        {

            //url: None,
            //host_and_port: String::with_capacity(100),
            read_web_socket_actor_io_client,
            io_server,
            connection_state_id: ConnectionStateId::new()

        }

    }

    //read_frame_processor_actor_input_sender: The sender for the output channel for the next phase.

    pub fn spawn() -> IOClient<ConnectionStateMessage<WebSocketActorInputMessage>, ConnectionStateMessage<WebSocketActorOutputMessage>>
    {

        let (io_client, io_server) = io_channel_both(50);

        let state = WebSocketActorState::new(io_server);

        WebSocketActor::spawn(state);

        io_client

    }

    /*
    pub fn new(input_receiver: Receiver<WebSocketActorInputMessage>, pipeline_message_counter: &PipelineMessageCounter, read_frame_processor_actor_input_sender: &Sender<CountedPipelineMessageMut<ReadFrameProcessorActorInputMessage>>) -> Self
    {

        //Spawn the ReadWebSocketActor

        let read_web_socket_actor_io_client = ReadWebSocketClientActorState::spawn(read_frame_processor_actor_input_sender, &pipeline_message_counter);

        Self
        {

            input_receiver,
            url: None,
            pipeline_message_counter: pipeline_message_counter.clone(),
            read_web_socket_actor_io_client: read_web_socket_actor_io_client,
            read_frame_processor_actor_input_sender: read_frame_processor_actor_input_sender.clone(),
            host_and_port: String::with_capacity(100)

        }

    }

    //read_frame_processor_actor_input_sender: The sender for the output channel for the next phase.

    pub fn spawn(pipeline_message_counter: &PipelineMessageCounter, read_frame_processor_actor_input_sender: &Sender<CountedPipelineMessageMut<ReadFrameProcessorActorInputMessage>>) -> Sender<WebSocketActorInputMessage>
    {

        let (input_sender, input_receiver) = channel(50);

        let state = WebSocketClientActorState::new(input_receiver, pipeline_message_counter, read_frame_processor_actor_input_sender);

        WebSocketClientActor::spawn(state);

        input_sender

    }
    */

    impl_pre_and_post_run_async!();

    //Default on_enter_async and on_exit_async implementations.

    //impl_default_start_and_end_async!();

    //The non-connected loop

    async fn run_async(&mut self) -> bool
    {

        if let Ok(message) = self.io_server.input_receiver_ref().recv().await
        //if let Ok(message) = self.input_receiver.recv().await
        {

            let res = self.process_received_actor_input_message(message).await;

            match res
            {

                Ok(write_half) =>
                {

                    self.connected_loop(write_half).await.should_continue()

                }
                Err(err) =>
                {

                    err.should_continue()

                }

            }

            /*
            match res
            {

                ContinueOrConnected::ShouldContinue(should_continue) => should_continue,
                ContinueOrConnected::Connected(web_socket_writer) =>
                {

                    //Connected loop
    
                    if !self.connected_loop(web_socket_writer).await
                    {
    
                        return false;
    
                    }

                    true
    
                }

            }
            */

        }
        else
        {

            false

        }

    }
    
    /*
    async fn connect_to_server(&mut self, url: &Url) -> Result<(WebSocket<TokioIo<Upgraded>>, Response<Incoming>)>
    {

        //Get from cache...

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

        self.host_and_port.clear();

        match url.host_str()
        {

            Some(the_host) =>
            {

                self.host_and_port.push_str(the_host);

            }
            None =>
            {

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

            }

        }

        match url.port()
        {

            Some(port) =>
            {

                self.host_and_port.push(':');

                self.host_and_port.push_str(&port.to_string());

            }
            None =>
            {

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

                self.host_and_port.push_str(":80");

            }

        }

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

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

        /*
        let host_str;

        if let Some(value) = url.host_str()
        {

            host_str = value;

        }
        else
        {

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

        let request = Request::builder()
            .method("GET")
            .uri(url.as_str())
            .header("Host", &self.host_and_port) //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))

    }
    */

    //disconnect_from_server

    ///
    /// Shuts down and drops the connection. The WebSocket connection closure process should've been completed by this point.
    /// 
    async fn disconnect(&mut self, mut web_socket_writer: WebSocketWriteHalf, received_close_frame: bool, sent_close_frame: bool) -> WebSocketWriterActorInternalMessage //Option<ConnectedLoopExitReason>
    {

        //If received_close_frame is false, send the close response frame, otherwise initiate the connection closure process.

        //Report that we're disconnecting here?

        //Notify the reader actor that we're disconnecting.

        if let Err(_) = self.read_web_socket_actor_io_client.input_sender_ref().send(self.connection_state_id.connection_state_message(ReadWebSocketActorInputMessage::Disconnecting)).await
        {

            return WebSocketWriterActorInternalMessage::WebSocketReaderActorChannelSenderClosed;

            //return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

        }

        //Notify the actor-client that we're disconnecting.

        if let Err(_) = self.io_server.output_sender_ref().send(self.connection_state_id.connection_state_message(WebSocketActorOutputMessage::Disconnecting)).await
        {

            return WebSocketWriterActorInternalMessage::WebSocketReaderActorChannelSenderClosed;

            //return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

        }

        if !sent_close_frame
        {

            if let Err(err) = self.send_close_frame(&mut web_socket_writer).await
            {

                return self.on_web_socket_error_report_only(err).await;
    
                /*
                let opt_res = self.on_web_socket_error_report_only(err).await;
    
                if let Some(res) = opt_res
                {

                    return res;

                }
                */

                /*
                if !Self::continue_or_not(&res)
                {
    
                    return res;
    
                }
                */

                /*
                let message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::ClientMessage(WebSocketActorOutputClientMessage::CloseFrameReceived(SendableText::Str(CLOSE_FRAME_SENT))));
        
                if let Err(_err) = self.read_frame_processor_actor_input_sender.send(message).await
                {
                    
                    return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

                }
                */

            }

            if let Err(_) = self.io_server.output_sender_ref().send(self.connection_state_id.connection_state_message(WebSocketActorOutputMessage::CloseFrameSent)).await
            {

                return WebSocketWriterActorInternalMessage::WebSocketReaderActorChannelSenderClosed;

                //return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

            }

        }

        //Receiving frames only.

        if !received_close_frame
        {

            //Wait fot the close frame response.

            let now = Instant::now();

            let soon = now.checked_add(Duration::from_secs(10)).expect("Error: Instant problems");

            //Disconnection loop
            
            //Loop because you don't know if the next frame is the close frame.

            loop
            {
                
                let output_receiver_recv = self.read_web_socket_actor_io_client.output_receiver_ref().recv();

                match timeout_at(soon, output_receiver_recv).await
                {

                    Ok(res) =>
                    {

                        match res
                        {

                            Ok(connected_state_message) =>
                            {

                                //let message_id = connected_state_message.id();

                                match connected_state_message.take_message()
                                {

                                    ReadWebSocketActorOutputMessage::ReadFrame(owned_frame) =>
                                    {

                                        //The actor-client should already have received the read close frame so there's no need to send a notification message here.

                                        if owned_frame.opcode == OpCode::Close
                                        {

                                            //The close frame has been received.

                                            break;

                                        }

                                        //Cache frame

                                        /*
                                        if let Err(_err) = self.io_server.output_sender_ref().send(self.connection_state_id.connection_state_message(WebSocketActorOutputMessage::ReadFrame(owned_frame))).await
                                        {

                                            //Correct message?

                                            return Some(ConnectedLoopExitReason::ServerDisconnectedOrConnectionError);

                                        }
                                        */

                                    }
                                    ReadWebSocketActorOutputMessage::Disconnected =>
                                    {

                                        break;

                                        //return WebSocketWriterActorInternalMessage::Disconnected;

                                        //This should be valid.

                                        //return None;

                                    }

                                }

                            }
                            Err(_) =>
                            {

                                return WebSocketWriterActorInternalMessage::ActorChannelReceiverClosed;

                                //return Some(ConnectedLoopExitReason::ReadWebSocketActorDropped);

                            }

                        }
                        
                        /*
                        match res
                        {

                            Ok(message) =>
                            {

                                match message
                                {

                                    WebSocketActorInputMessage::ConnectTo(url) =>
                                    {

                                        let error_message = format!("Error: Cannot connect to {url} as the client is disconnecting.");

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

                                        if let Err(_err) = self.read_frame_processor_actor_input_sender.send(counted_message).await
                                        {
                                
                                            return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);
                    
                                        }

                                    }
                                    WebSocketActorInputMessage::Disconnect =>
                                    {

                                        return Some(ConnectedLoopExitReason::ReadWebSocketActorDropped);

                                    }
                                    WebSocketActorInputMessage::WriteFrame(frame) =>
                                    {

                                        match frame.opcode
                                        {
        
                                            OpCode::Close =>
                                            {

                                                //Close frame response received.
        
                                                let message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::FrameAndMessage(frame, WebSocketActorOutputClientMessage::CloseFrameReceived(SendableText::Str(CLOSE_FRAME_RECEIVED))));
        
                                                if let Err(_err) = self.read_frame_processor_actor_input_sender.send(message).await
                                                {
                                                    
                                                    return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);
                            
                                                }

                                                //Tell the ReadWebSocketActor to drop its side of the WebSocket stream.

                                                if let Err(_) = self.read_web_socket_actor_io_client.input_sender_ref().send(ReadWebSocketActorInputMessage::Disconnect).await
                                                {

                                                    return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

                                                }
                                                
                                                break;
        
                                            }
                                            OpCode::Continuation | OpCode::Text | OpCode::Binary | OpCode::Pong =>
                                            {
        
                                                let message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::Frame(frame));
        
                                                if let Err(_err) = self.read_frame_processor_actor_input_sender.send(message).await
                                                {
                                        
                                                    return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);
                            
                                                }

                                            }
                                            OpCode::Ping =>
                                            {
        
                                                //From cache...

                                                let mut of = OwnedFrame::new();

                                                of.copy_payload_from_other(&frame);

                                                of.pong_setup();
                                                
                                                let message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::FrameAndMessage(frame, WebSocketActorOutputClientMessage::PingFrameReceived(SendableText::Str(PING_FRAME_RECEIVED))));

                                                if let Err(_err) = self.read_frame_processor_actor_input_sender.send(message).await
                                                {
                                        
                                                    return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);
                            
                                                }

                                                if let Err(_) = self.write_frame(&mut web_socket_writer, of).await
                                                {

                                                    return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

                                                }
        
                                            }
                    
                                        }

                                    }

                                }

                            }
                            Err(_) =>
                            {

                                return Some(ConnectedLoopExitReason::ReadWebSocketActorDropped);

                            }

                        }
                        */

                    },
                    Err(_err) =>
                    {

                        let new_id = self.connection_state_id.next();

                        //Tell the ReadWebSocketActor to drop its side of the WebSocket stream.

                        if let Err(_) = self.read_web_socket_actor_io_client.input_sender_ref().send(new_id.connection_state_message(ReadWebSocketActorInputMessage::Disconnect)).await
                        {

                            return WebSocketWriterActorInternalMessage::ActorChannelReceiverClosed;

                            //return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

                        }

                        //let message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::ClientMessage(WebSocketActorOutputClientMessage::Disconnected(SendableText::Str(TIME_ELAPSED_FORCED_CLOSURE_NOTICE))));

                        //Report the disconnection to the actor-client. CloseWindowElapsed == Disconnected

                        if let Err(_) = self.io_server.output_sender_ref().send(new_id.connection_state_message(WebSocketActorOutputMessage::CloseWindowElapsed)).await
                        //if let Err(_) = self.read_frame_processor_actor_input_sender.send(message).await
                        {

                            return WebSocketWriterActorInternalMessage::ActorChannelReceiverClosed;

                            //return Some(ConnectedLoopExitReason::ServerDisconnectedOrConnectionError);
                
                            //return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);
                
                        }

                        //self.url = None;

                        return WebSocketWriterActorInternalMessage::Disconnected;

                        //return None;

                        //break;

                    }

                }

            }
            
        }
        /*
        else
        {

            let new_id = self.connection_state_id.next();

            //Tell the ReadWebSocketActor to drop its side of the WebSocket stream.

            if let Err(_) = self.read_web_socket_actor_io_client.input_sender_ref().send(new_id.connection_state_message(ReadWebSocketActorInputMessage::Disconnect)).await
            //if let Err(_) = self.read_web_socket_actor_io_client.input_sender_ref().send(ReadWebSocketActorInputMessage::Disconnect).await
            {

                return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

            }
            
        }
        */

        //Update the connection_state_id.

        let new_id = self.connection_state_id.next();

        //Send error or other message.

        //Make sure to drop the URL as well.

        //self.url = None;

        //let message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::ClientMessage(WebSocketActorOutputClientMessage::Disconnected(SendableText::Str(DISCONNECTED_FROM_SERVER))));

        if let Err(_) = self.read_web_socket_actor_io_client.input_sender_ref().send(new_id.connection_state_message(ReadWebSocketActorInputMessage::Disconnect)).await
        //if let Err(_) = self.read_frame_processor_actor_input_sender.send(message).await
        {

            return WebSocketWriterActorInternalMessage::ActorChannelSenderClosed;

            //return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

        }

        //Notify the actor-client that we've diconnected.

        if let Err(_) = self.io_server.output_sender_ref().send(new_id.connection_state_message(WebSocketActorOutputMessage::Disconnected)).await
        {

            return WebSocketWriterActorInternalMessage::WebSocketReaderActorChannelSenderClosed;

        }

        WebSocketWriterActorInternalMessage::Disconnected

        //None

    }

    //disconnect_from_server_now

    async fn disconnect_now(&mut self) -> WebSocketWriterActorInternalMessage
    {

        let new_id = self.connection_state_id.next();

        //Notify the reader actor that we're disconnecting.

        if let Err(_) = self.read_web_socket_actor_io_client.input_sender_ref().send(new_id.connection_state_message(ReadWebSocketActorInputMessage::Disconnect)).await
        {

            return WebSocketWriterActorInternalMessage::WebSocketReaderActorChannelSenderClosed;

        }

        //Notify the actor-client that we've diconnected.

        if let Err(_) = self.io_server.output_sender_ref().send(new_id.connection_state_message(WebSocketActorOutputMessage::Disconnected)).await
        {

            return WebSocketWriterActorInternalMessage::WebSocketReaderActorChannelSenderClosed;

        }

        WebSocketWriterActorInternalMessage::Disconnected

    }

    /*
    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));

            }

        }

    }
    */

    async fn new_connection(&mut self, web_socket: WebSocket<TokioIo<Upgraded>>) -> InternalConnectionResult<WebSocketWriteHalf> 
    {

        let current_id = self.connection_state_id.next();

        //Split the stream here

        let (read, write) = web_socket.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::WebSocketReaderActorChannelSenderClosed); //ActorClientSenderClosed);

        }

        InternalConnectionResult::Ok(write)

    }

    //Disconnected: Not connected to a server, should continue?

    async fn process_received_actor_input_message(&mut self, ided_message: ConnectionStateMessage<WebSocketActorInputMessage>) -> InternalConnectionResult<WebSocketWriteHalf> //ConnectionStateMessage<WebSocketClientActorInputMessage> //ConnectedOrOther<WebSocketReadHalf> //WebSocketWriterActorInternalMessage //ContinueOrConnected
    {

        //Ignore the id

        let message = ided_message.take_message();

        //Not connected

        match message
        {

            //WebSocketClientActorInputMessage::ConnectTo(url) =>
            WebSocketActorInputMessage::NewConnection(websocket) =>
            {

                self.new_connection(websocket).await

                //self.prepare_for_new_connection_and_connect(url).await

                //match self.prepare_for_new_connection_and_connect(url).await
                //{

                    /*
                    CLEROrConnected::CLER(res) =>
                    {

                        return ContinueOrConnected::ShouldContinue(Self::continue_or_not(&res));

                    }
                    CLEROrConnected::Connected(connection) =>
                    {

                        //Connected to a server, return the connection.

                        return ContinueOrConnected::Connected(connection);

                    }
                    */
                    
                //}

            }
            WebSocketActorInputMessage::Disconnect | WebSocketActorInputMessage::DisconnectNow | WebSocketActorInputMessage::WriteFrame(_) =>
            {

                Err(self.report_not_connected().await)

                //return self.report_not_connected().await;

            }

        }

    }

    async fn report_not_connected(&mut self) -> WebSocketWriterActorInternalMessage //Option<WebSocketWriterActorInternalMessage> //ContinueOrConnected
    {

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

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

        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 WebSocketWriterActorInternalMessage::ActorChannelSenderClosed;

            //return ContinueOrConnected::ShouldContinue(false);

        }

        WebSocketWriterActorInternalMessage::NonConnectionError

        //None

        //ContinueOrConnected::ShouldContinue(true)

    }

    /*
    fn continue_or_not(opt_cler: &Option<ConnectedLoopExitReason>) -> bool
    {

        match opt_cler
        {

            Some(res) =>
            {

                return res.should_continue();

            },
            None =>
            {

                return true;
                
            }

        }

    }
    */

    //The loop for when connected to a server, should the actor continue after this?

    async fn connected_loop(&mut self, mut web_socket_writer: WebSocketWriteHalf) -> WebSocketWriterActorInternalMessage //bool
    {

        enum ConnectedInput
        {

            WebSocketClientReaderActorOuput(Result<ConnectionStateMessage<ReadWebSocketActorOutputMessage>, ReceiveError>),
            InputReceiverInput(Result<ConnectionStateMessage<WebSocketActorInputMessage>, ReceiveError>)

        }

        loop
        {

            let res;

            let read_web_socket_actor_io_client_output_receiver_recv = self.read_web_socket_actor_io_client.output_receiver_ref().recv();

            let input_receiver_recv = self.io_server.input_receiver_ref().recv(); //self.input_receiver.recv();

            //et res2;
    
            //let _ = self.read_web_socket_actor_io_client.output_receiver().try_recv();

            //let from_server;

            select!
            {
                
                biased;

                connected_res = read_web_socket_actor_io_client_output_receiver_recv =>
                {

                    //From the associated ReadWebSocketActor.

                    res = ConnectedInput::WebSocketClientReaderActorOuput(connected_res);

                    //from_server = true;

                    //res2 = Some(self.read_web_socket_actor_io_client.output_receiver().try_recv());

                }
                connected_res = input_receiver_recv =>
                {

                    //From the GUI.

                    res = ConnectedInput::InputReceiverInput(connected_res);

                    //from_server = false;

                    //res2 = None;

                }
                else =>
                {

                    return WebSocketWriterActorInternalMessage::ActorChannelReceiverClosed;

                }
    
            }

            match res
            {

                ConnectedInput::WebSocketClientReaderActorOuput(reader_ouput_res) =>
                {

                    match reader_ouput_res
                    {

                        Ok(connection_state_message) =>
                        {

                            let message_id = connection_state_message.id();

                            match connection_state_message.take_message()
                            {

                                ReadWebSocketActorOutputMessage::ReadFrame(owned_frame) =>
                                {

                                    if self.connection_state_id != message_id
                                    {

                                        //Cache frame

                                        continue;

                                    }
                                    
                                    /* 
                                    if owned_frame.opcode == OpCode::Close
                                    {

                                        return self.disconnect(web_socket_writer, true, false).await;

                                        //disconnect_from_server

                                        //return Self::continue_or_not( &self.disconnect_from_server(web_socket_writer, true, false).await);

                                    }
                                    else
                                    */

                                    {

                                        let opcode = owned_frame.opcode;

                                        if let Err(err) = self.write_frame(&mut web_socket_writer, owned_frame).await
                                        {

                                            return self.on_web_socket_error(err, web_socket_writer).await

                                            /*
                                            if let Some(internal_message) = self.on_web_socket_error(err, web_socket_writer).await
                                            {

                                                return internal_message;

                                            }
                                            else
                                            {

                                                return WebSocketWriterActorInternalMessage::ConnectionError;
                                                
                                            }
                                            */

                                            //(SendableText::ArcStr(Arc::from(err.to_string())));

                                            //return false;

                                        }

                                        if opcode == OpCode::Close
                                        {

                                            return self.disconnect(web_socket_writer, true, false).await;

                                        }
                                        else if opcode == OpCode::Pong
                                        {

                                            if let Err(_) = self.io_server.output_sender_ref().send(self.connection_state_id.connection_state_message(WebSocketActorOutputMessage::PongFrameSent)).await
                                            {

                                                return WebSocketWriterActorInternalMessage::ActorChannelSenderClosed;

                                                //return false;

                                            }

                                        }
                                        
                                    }

                                }
                                ReadWebSocketActorOutputMessage::Disconnected =>
                                {

                                    //Make sure this is in regards to the right connection. 

                                    if self.connection_state_id != message_id
                                    {

                                        continue;

                                    }

                                    //Is this always an error?

                                    let error_text = SendableText::Str("Unexpected reader error: connection closed.");

                                    if let Err(_) = self.io_server.output_sender_ref().send(self.connection_state_id.connection_state_message(WebSocketActorOutputMessage::ConnectionError(error_text))).await
                                    {

                                        return WebSocketWriterActorInternalMessage::ActorChannelSenderClosed;

                                        //return false;

                                    }

                                    return WebSocketWriterActorInternalMessage::Disconnected;

                                    //return true;

                                }

                            }

                        }
                        Err(_) =>
                        {

                            return WebSocketWriterActorInternalMessage::WebSocketReaderActorChannelReceiverClosed;

                            //return false;

                        }

                    }

                },
                ConnectedInput::InputReceiverInput(input_receiver_res) =>
                {

                    match input_receiver_res
                    {

                        Ok(connection_state_message) =>
                        {

                            let message_id = connection_state_message.id();

                            match connection_state_message.take_message()
                            {

                                WebSocketActorInputMessage::NewConnection(web_socket) =>
                                {

                                    match self.new_connection(web_socket).await
                                    {

                                        Ok(res) =>
                                        {

                                            web_socket_writer = res;

                                        }
                                        Err(err) =>
                                        {

                                            return err;

                                        }

                                    }

                                }
                                /*
                                WebSocketActorInputMessage::ConnectTo(url) =>
                                {

                                    //Ignore any ConnectTo messages from the ReadWebSocketActor.

                                    /*
                                    if from_server
                                    {

                                        continue;

                                    }
                                    */

                                    /*
                                    if !Self::continue_or_not(&self.disconnect_from_server(web_socket_writer, false, false).await)
                                    {

                                        return false;

                                    }
                                    */

                                    let res = self.prepare_for_new_connection_and_connect(url).await;

                                    match res
                                    {

                                        Ok(new_web_socket_writer) =>
                                        {

                                            web_socket_writer = new_web_socket_writer;

                                            //Output to actor client here?

                                        }
                                        Err(err) =>
                                        {

                                            return err;

                                        }

                                        /*
                                        CLEROrConnected::CLER(should_continue) =>
                                        {

                                            return Self::continue_or_not(&should_continue);

                                        },
                                        CLEROrConnected::Connected(new_web_socket_writer) =>
                                        {

                                            web_socket_writer = new_web_socket_writer;

                                            //The new connection is set, stay in the loop. 

                                        }
                                        */

                                    }

                                }
                                */
                                WebSocketActorInputMessage::Disconnect =>
                                {

                                    return self.disconnect(web_socket_writer, false, false).await;

                                    //disconnect_from_server

                                    //return Self::continue_or_not(&self.disconnect_from_server(web_socket_writer, false, false).await);

                                }
                                WebSocketActorInputMessage::WriteFrame(frame) =>
                                {

                                    //Is the message_id valid?

                                    if self.connection_state_id != message_id
                                    {

                                        //Cache frame

                                        continue;

                                    }

                                    let opcode = frame.opcode;

                                    if let Err(err) = self.write_frame(&mut web_socket_writer, frame).await
                                    {

                                        return self.on_web_socket_error(err, web_socket_writer).await;

                                        /*
                                        if let Some(err) = self.on_web_socket_error(err, web_socket_writer).await
                                        {

                                            return err;

                                        }
                                        else
                                        {

                                            return WebSocketWriterActorInternalMessage::ConnectionError;
                                            
                                        }
                                        */

                                        //return Self::continue_or_not(&self.on_web_socket_error(err, web_socket_writer).await);

                                    }

                                    if opcode == OpCode::Close
                                    {

                                        //Initiate the disconnection process if the frame sent was a close frame.

                                        return self.disconnect(web_socket_writer, false, true).await;

                                    }

                                    //Did the UI send a close frame? Then from_server = false.

                                    //Did the server send the close frame? Then from_server = true.

                                    /*
                                    if opcode == OpCode::Close
                                    {

                                        let message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::ClientMessage(WebSocketActorOutputClientMessage::CloseFrameReceived(SendableText::Str(CLOSE_FRAME_SENT))));

                                        if let Err(_err) = self.read_frame_processor_actor_input_sender.send(message).await
                                        {
                
                                            return false;

                                        }                

                                        return Self::continue_or_not( &self.disconnect_from_server(web_socket_writer, from_server, true).await);

                                    }
                                    */

                                }
                                WebSocketActorInputMessage::DisconnectNow =>
                                {

                                    return self.disconnect_now().await;

                                    //return self.disconnect_from_server_now().await;

                                }

                            }

                        }
                        Err(_) =>
                        {

                            return WebSocketWriterActorInternalMessage::ActorChannelReceiverClosed;

                            //return false;

                        }

                    }

                }
                
            }

        }

    }

    async fn write_frame(&self, web_socket_writer: &mut WebSocketWriteHalf, mut of: OwnedFrame) -> Result<(), WebSocketError>
    {

        let frame = of.new_frame_to_be_written();

        web_socket_writer.write_frame(frame).await?;

        //Put OwnedFrame into cache...

        Ok(())

    }

    async fn send_close_frame(&self, web_socket_writer: &mut WebSocketWriteHalf) -> Result<(), WebSocketError>
    {

        println!("--> send_close_frame(_) <--");

        web_socket_writer.write_frame(Frame::close_raw(vec![].into())).await

    }

    async fn on_web_socket_error(&mut self, error: WebSocketError, _web_socket_writer: WebSocketWriteHalf) -> WebSocketWriterActorInternalMessage //Option<WebSocketWriterActorInternalMessage> //Option<ConnectedLoopExitReason>
    {

        //Make sure to drop the URL as well.

        //self.url = None;

        self.on_web_socket_error_report_only(error).await

        /*
        let res = self.on_web_socket_error_report_only(error).await;
        
        if !Self::continue_or_not(&res)
        {

            return res;

        }

        res
        */

    }

    async fn on_web_socket_error_report_only(&mut self, error: WebSocketError) -> WebSocketWriterActorInternalMessage //Option<WebSocketWriterActorInternalMessage> //Option<ConnectedLoopExitReason>
    {

        if let Err(_) = self.io_server.output_sender_ref().send(self.connection_state_id.connection_state_message(WebSocketActorOutputMessage::ConnectionError(SendableText::String(error.to_string())))).await
        {

            return WebSocketWriterActorInternalMessage::ActorChannelSenderClosed;

            //return Some(WebSocketWriterActorInternalMessage::ActorClientChannelSenderClosed); //ActorClientReceiverClosed);

            //return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

        }

        //let message = self.pipeline_message_counter.increment_with_message_mut(ReadFrameProcessorActorInputMessage::ClientMessage(WebSocketActorOutputClientMessage::ConnectionError(SendableText::String(error.to_string()))));

        /*
        if let Err(_) = self.read_frame_processor_actor_input_sender.send(message).await
        {

            return Some(ConnectedLoopExitReason::ReadFrameProcessorActorDropped);

        }
        */

        WebSocketWriterActorInternalMessage::ConnectionError

        //None

    }

    async fn on_non_connection_error(&mut self, non_connection_error: NonConnectionError) -> WebSocketWriterActorInternalMessage //InternalConnectionResult<WebSocketWriteHalf>
    {

        let ided_message = self.connection_state_id.connection_state_message(WebSocketActorOutputMessage::NonConnectionError(non_connection_error));

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

            return WebSocketWriterActorInternalMessage::ActorChannelSenderClosed; //ActorClientReceiverClosed;

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

        }

         WebSocketWriterActorInternalMessage::NonConnectionError

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

    }

}

//Setup the macro generated Task actor.

impl_mac_task_actor!(WebSocketActor);

//The ReadWebSocketActor

//Back-channel message

/*
enum ReadWebSocketActorInputMessage
{

    Connected(WebSocketReader),
    Disconnect

}
*/

//Subordinate actor, beginning of read-frame pipeline.

struct WebSocketReaderActorState
{

    write_frame_processor_actor_io_sender: Sender<ConnectionStateMessage<WebSocketActorOutputMessage>>, //Probably have two of these one here and the other in the writer actor.
    connection_state_id: ConnectionStateId,
    //read_frame_processor_actor_io_sender: Sender<CountedPipelineMessageMut<ReadFrameProcessorActorInputMessage>>, //To the ReadFrameProcessorActor (the next stage).
    //pipeline_message_counter: PipelineMessageCounter,
    obligated_send_frame_holder: Arc<ArrayQueue<OwnedFrame>>, //A handy work-around. May be swapped with an Arc<UnSafeCell>.
    //io_server: ActorIOServer<ReadWebSocketActorInputMessage, WebSocketActorInputMessage> //Outputs to the "contaning" WebSocketActor.
    io_server: IOServer<ConnectionStateMessage<ReadWebSocketActorInputMessage>, ConnectionStateMessage<ReadWebSocketActorOutputMessage>>
    //io_server: IOServer<ReadWebSocketActorInputMessage, WebSocketActorInputMessage>

}

impl WebSocketReaderActorState
{

    pub fn new(write_frame_processor_actor_io_sender: &Sender<ConnectionStateMessage<WebSocketActorOutputMessage>>) -> (IOClient<ConnectionStateMessage<ReadWebSocketActorInputMessage>, ConnectionStateMessage<ReadWebSocketActorOutputMessage>>, Self)
    {

        let (io_client, io_server) = io_channels(4, 10); //(2, 10);

        (io_client,
        Self
        {

            write_frame_processor_actor_io_sender: write_frame_processor_actor_io_sender.clone(),
            connection_state_id: ConnectionStateId::new(),
            obligated_send_frame_holder: Arc::new(ArrayQueue::new(1)),
            io_server

        })

    }

    pub fn spawn(write_frame_processor_actor_io_sender: &Sender<ConnectionStateMessage<WebSocketActorOutputMessage>>) -> IOClient<ConnectionStateMessage<ReadWebSocketActorInputMessage>, ConnectionStateMessage<ReadWebSocketActorOutputMessage>> //ActorIOClient<ReadWebSocketActorInputMessage, WebSocketActorInputMessage>
    {

        let (io_client, state) = WebSocketReaderActorState::new(write_frame_processor_actor_io_sender);

        WebSocketReaderActor::spawn(state);

        //ReadWebSocketClientActor::spawn(state);

        io_client

    }

    /*
    pub fn new(read_frame_processor_actor_io_sender: &Sender<CountedPipelineMessageMut<ReadFrameProcessorActorInputMessage>>, pipeline_message_counter: &PipelineMessageCounter) -> (IOClient<ReadWebSocketActorInputMessage, WebSocketActorInputMessage>, Self) //(ActorIOClient<ReadWebSocketActorInputMessage, WebSocketActorInputMessage>, Self)
    {

        let (io_client, io_server) = io_channels(2, 10); //actor_io(2, 10);

        (io_client,
        Self
        {

            read_frame_processor_actor_io_sender: read_frame_processor_actor_io_sender.clone(),
            pipeline_message_counter: pipeline_message_counter.clone(),
            obligated_send_frame_holder: Arc::new(ArrayQueue::new(1)),
            io_server

        })

    }

    pub fn spawn(read_frame_processor_actor_io_sender: &Sender<CountedPipelineMessageMut<ReadFrameProcessorActorInputMessage>>, pipeline_message_counter: &PipelineMessageCounter) -> IOClient<ReadWebSocketActorInputMessage, WebSocketActorInputMessage> //ActorIOClient<ReadWebSocketActorInputMessage, WebSocketActorInputMessage>
    {

        let (io_client, state) = ReadWebSocketClientActorState::new(read_frame_processor_actor_io_sender, pipeline_message_counter);

        ReadWebSocketClientActor::spawn(state);

        io_client

    }
    */

    impl_pre_and_post_run_async!();

    //impl_default_start_and_end_async!();

    //The non-connected loop.

    async fn run_async(&mut self) -> bool
    {

        match self.io_server.input_receiver_ref().recv().await
        {

            Ok(message) =>
            {

                //Set the connection_state_id so that ConnectionStateMessage<WebSocketActorOutputMessage> messages can get the proper connection id.

                //self.connection_state_id = message.id();

                match message.set_id_take_message(&mut self.connection_state_id) //message.take_message()
                {

                    ReadWebSocketActorInputMessage::NewConnection(reader) =>
                    //ReadWebSocketActorInputMessage::Connected(reader) =>
                    {

                        //Send the new connection notification.

                        let message = WebSocketActorOutputMessage::NewConnection;

                        let ided_message = self.connection_state_id.connection_state_message(message);
    
                        if let Err(_err) = self.write_frame_processor_actor_io_sender.send(ided_message).await
                        {

                            return false;

                        }

                        self.connected_loop(reader).await

                        /*
                        if !self.connected_loop(reader).await
                        {

                            return false;

                        }
                        else
                        {

                            true

                        }
                        */

                    }
                    ReadWebSocketActorInputMessage::Disconnect | ReadWebSocketActorInputMessage::Disconnecting => { /* Continue */ true },

                }

            }
            Err(_) => //return false
            {

                //println!("ReadWebSocketActor Exiting!");

                return false

            }

        }

    }

    //The connected loop

    async fn connected_loop(&mut self, mut reader: WebSocketReader) -> bool
    {

        enum SelectResult<'f>
        {

            ReadFrame(Result<Frame<'f>, WebSocketError>),
            //Connection(Option<ReadWebSocketActorInputMessage>)
            //Connection(ReceiveResult<ReadWebSocketActorInputMessage>)
            Connection(ReceiveResult<ConnectionStateMessage<ReadWebSocketActorInputMessage>>)

        }

        let mut is_disconnecting = false;

        let obligated_send_frame_holder = self.obligated_send_frame_holder.clone();

        let mut is_obligated_send_frame = false;

        let mut send_fn = |obligated_send_frame: Frame|
        {

            is_obligated_send_frame = true;

            //Take the obligated_send_fram out of this closure,

            //Get from cache

            let mut of = OwnedFrame::new();

            of.copy_all_from_read_frame(&obligated_send_frame);

            let _ = obligated_send_frame_holder.push(of).expect("Error: The obligated_send_frame_holder should've been checked.");

            async
            {

                Result::<(), WebSocketError>::Ok(())

            }

        };

        //println!("ReadWebSocketActor: before connected_loop");

        loop
        {

            //From the WebSocketActor

            let connection_future = self.io_server.input_receiver_ref().recv();

            //From the WebSocketReadHalf or FragmentCollectorRead.

            let read_frame_future = reader.read_frame(&mut send_fn);

            let select_result;

            //Doesn't look like reader.read_frame is cancel-safe.
            
            //Should only loose one frame at most however.

            select!
            {

                biased;

                res = connection_future =>
                {

                    //This select invocation is biased so that the actor will disconnect or change connection when told. 

                    select_result = SelectResult::Connection(res);

                }
                res = read_frame_future =>
                {

                    //Read from WebSocketReadHalf or FragmentCollectorRead.

                    select_result = SelectResult::ReadFrame(res);

                }
                else =>
                {

                    //println!("ReadWebSocketActor: The input receiver has definitely closed in this case.");

                    //The input receiver has definitely closed in this case.

                    return false;

                }

            }

            //Does an obligated frame need to be sent?

            if let Some(mut of) = obligated_send_frame_holder.pop()
            {

                println!("WebSocketReaderActor: connected_loop - obligated_send_frame_holder.pop()");

                println!("{:?}", of);

                match of.opcode
                {
    
                    OpCode::Close =>
                    {

                        //println!("ReadWebSocketActor: obligated_send_frame_holder.pop() - OpCode::Close");

                        //Server initiated disconnection.

                        //Send close frame
    
                        of.clear_payload();

                        //From this message the WebSocketActor must complete the disconnection process.

                        let message = ReadWebSocketActorOutputMessage::ReadFrame(of);
    
                        let ided_message = ConnectionStateMessage::new(self.connection_state_id, message);

                        if let Err(_err) = self.io_server.output_sender_ref().send(ided_message).await
                        {
        
                            return false;
        
                        }

                        //Only set is_disconnecting in response to a writer actor message.

                        //is_disconnecting = true;

                        //Notify the actor-client that a close frame is being sent. - Not here

                        /*
                        let message = WebSocketActorOutputMessage::SendingCloseFrame;
    
                        let ided_message = ConnectionStateMessage::new(self.connection_state_id, message);

                        if let Err(_err) = self.write_frame_processor_actor_io_sender.send(ided_message).await
                        {
        
                            return false;
        
                        }
                        */
                        
                        /*
                        if let Err(_err) = self.io_server.output_sender_ref().send(WebSocketActorInputMessage::WriteFrame(of)).await
                        {
        
                            return false;
        
                        }
                        */
                        
                    }
                    OpCode::Ping =>
                    {

                        //println!("ReadWebSocketActor: obligated_send_frame_holder.pop() - OpCode::Ping");

                        if is_disconnecting
                        {

                            //Don't send a pong frame.

                            continue;

                        }

                        //Now send a pong frame.

                        of.pong_setup();

                        //Clear the payload.

                        of.clear_payload();
                        
                        let message = ReadWebSocketActorOutputMessage::ReadFrame(of);
    
                        let ided_message = ConnectionStateMessage::new(self.connection_state_id, message);

                        if let Err(_err) = self.io_server.output_sender_ref().send(ided_message).await
                        {
        
                            return false;
        
                        }

                        //Notify the actor-client that a pong frame is being sent. - Not here

                        /*
                        let message = WebSocketActorOutputMessage::SendingPongFrame;
    
                        let ided_message = ConnectionStateMessage::new(self.connection_state_id, message);

                        if let Err(_err) = self.write_frame_processor_actor_io_sender.send(ided_message).await
                        {
        
                            return false;
        
                        }
                        */

                        /*
                        if let Err(_err) = self.io_server.output_sender_ref().send(WebSocketActorInputMessage::WriteFrame(of)).await
                        {
        
                            return false;
        
                        }
                        */

                    }
                    OpCode::Continuation | OpCode::Text | OpCode::Binary | OpCode::Pong =>
                    {

                        //println!("ReadWebSocketActor: obligated_send_frame_holder.pop() - OpCode - Other");

                        let opcode = of.opcode;

                        //If for some reason you get another type of frame...

                        //Send on behalf of the WebSocketActor.
                        
                        let message = WebSocketActorOutputMessage::ReadFrame(of);
    
                        let ided_message = self.connection_state_id.connection_state_message(message); //self.new_connection_state_message(message); //ConnectionStateMessage::new(self.connection_state_id, message);
    
                        if let Err(_err) = self.write_frame_processor_actor_io_sender.send(ided_message).await
                        {

                            return false;

                        }

                        //Notify the actor-client that an error has occured.

                        let message = WebSocketActorOutputMessage::ConnectionError(SendableText::String(format!("Error: This is not an obligated frame: {:?}", opcode)));
    
                        let ided_message = ConnectionStateMessage::new(self.connection_state_id, message);

                        if let Err(_err) = self.write_frame_processor_actor_io_sender.send(ided_message).await
                        {
        
                            return false;
        
                        }

                        return true;

                        /*
                        let message = ReadFrameProcessorActorInputMessage::Frame(of);
    
                        let counted_message = self.pipeline_message_counter.increment_with_message_mut(message);
    
                        if let Err(_err) = self.read_frame_processor_actor_io_sender.send(counted_message).await
                        {

                            return false;

                        }
                        */
    
                    }
    
                }

            }

            //Send a read frame to the ReadFrameProcessorActor, change connection or disconnect.

            match select_result
            {

                SelectResult::ReadFrame(frame_res) =>
                {

                    match frame_res
                    {

                        Ok(frame) =>
                        {

                            //println!("ReadWebSocketActor: select_result - SelectResult::ReadFrame - Ok");

                            //Get from cache...

                            let mut of = OwnedFrame::new();

                            of.copy_all_from_read_frame(&frame);

                            println!("WebSocketReaderActor: SelectResult::ReadFrame");

                            println!("{:?}", of);

                            let message = WebSocketActorOutputMessage::ReadFrame(of);
    
                            let ided_message = self.connection_state_id.connection_state_message(message); //self.new_connection_state_message(message);  //ConnectionStateMessage::new(self.connection_state_id, message);
        
                            if let Err(_err) = self.write_frame_processor_actor_io_sender.send(ided_message).await
                            {

                                return false;

                            }

                            /*
                            let message = ReadFrameProcessorActorInputMessage::Frame(of);
    
                            let counted_message = self.pipeline_message_counter.increment_with_message_mut(message);
        
                            if let Err(_err) = self.read_frame_processor_actor_io_sender.send(counted_message).await
                            {
    
                                return false;
    
                            }
                            */

                        }
                        Err(_err) =>
                        {

                            //println!("ReadWebSocketActor: select_result - SelectResult::ReadFrame - Err");

                            let message = ReadWebSocketActorOutputMessage::Disconnected;
    
                            let ided_message = self.connection_state_id.connection_state_message(message);  //self.new_connection_state_message(message);

                            if let Err(_err) = self.io_server.output_sender_ref().send(ided_message).await
                            {
    
                                return false;
    
                            }

                            //The writer actor should only report arrors to its actor-client.

                            /*
                            let message = WebSocketActorOutputMessage::ConnectionError(SendableText::String(err.to_string()));
    
                            let ided_message = self.connection_state_id.connection_state_message(message); //self.new_connection_state_message(message);
        
                            if let Err(_err) = self.write_frame_processor_actor_io_sender.send(ided_message).await
                            {
    
                                return false;
    
                            }
                            */

                            return true;

                            /*
                            if let Err(_err) = self.io_server.output_sender_ref().send(WebSocketActorInputMessage::Disconnect).await
                            {
    
                                return false;
    
                            }

                            let message = ReadFrameProcessorActorInputMessage::ClientMessage(WebSocketActorOutputClientMessage::ConnectionError(SendableText::String(err.to_string())));
    
                            let counted_message = self.pipeline_message_counter.increment_with_message_mut(message);
        
                            if let Err(_err) = self.read_frame_processor_actor_io_sender.send(counted_message).await
                            {
    
                                return false;
    
                            }
                            */

                        }

                    }

                }
                SelectResult::Connection(input_result) =>
                {

                    if let Ok(message) = input_result
                    {

                        //println!("ReadWebSocketActor: select_result - SelectResult::Connection - Ok");

                        //self.connection_state_id = message.id();

                        match message.set_id_take_message(&mut self.connection_state_id) //message.take_message()
                        {

                            ReadWebSocketActorInputMessage::NewConnection(new_reader) =>
                            {

                                //Send the new connection notification.

                                let message = WebSocketActorOutputMessage::NewConnection;

                                let ided_message = self.connection_state_id.connection_state_message(message);
            
                                if let Err(_err) = self.write_frame_processor_actor_io_sender.send(ided_message).await
                                {

                                    return false;

                                }

                                //Set the new reader

                                reader = new_reader;

                                //Make sure this is false too.

                                is_disconnecting = false;

                            }
                            ReadWebSocketActorInputMessage::Disconnecting =>
                            {

                                //println!("ReadWebSocketActorInputMessage::Disconnecting Received!");

                                is_disconnecting = true;

                            }
                            ReadWebSocketActorInputMessage::Disconnect =>
                            {

                               //println!("ReadWebSocketActorInputMessage::Disconnect Received!");

                                return true;

                            }

                        }
       
                        /*
                        match input
                        {

                            ReadWebSocketActorInputMessage::Connected(new_reader) =>
                            {

                                reader = new_reader;

                            }
                            ReadWebSocketActorInputMessage::Disconnect =>
                            {

                                return true;

                            }

                        }
                        */
                        
                    }
                    else
                    {

                        //println!("ReadWebSocketActor: SelectResult::Connection(input_result) => ReceiveError!");

                        return false;

                    }

                }

            }

            //println!("ReadWebSocketActor: connected_loop bottom reached");

        }

    }

    /*
    fn new_connection_state_message<T>(&self, message: T) -> ConnectionStateMessage<T>
    {

        ConnectionStateMessage::new(self.connection_state_id, message)

    }
    */

}

impl_mac_task_actor!(WebSocketReaderActor);

//impl_mac_task_actor!(ReadWebSocketClientActor);