#![warn(missing_docs)]
//! Netty-rs allows exposes a simple-to-use API used to create stateful application level network
//! protocols as both a client or server.
//!
//! Netty-rs allows requires consumers specify how to handle messages in different
//! circumstances. Whenever specifying this the same API is used by using [Connection](Connection).
//! This very simple API allows consumers to specify restful protocols of varying complexity.
//! Each message-and-reply chain is in their own channel which does not and is not impacted by
//! messages sent or received in other message-and-reply chains.
//!
//! The situations where how to handle messages need to be specified are:
//! 1. When a consumer sends a message it can choose to wait for a reply and handle it in a
//! custom way.
//! 2. When acting as a server consumers need to specify how to handshake with new connections,
//! which allows custom authentication of clients among any other handshake related action.
//! 3. When acting as a server consumers need to specify how to handle non-reply messages from
//! connections that have already been authenticated.
//!
//! The main API is accessed through the [Networker](Networker) struct.
//!
//! Netty-rs uses the [DirectoryService](DirectoryService) trait in order to allow consumers to either implement
//! their own directory service for example a DNS or use the `SimpleDirectoryService` struct that
//! implements this trait.
//!
//! Example
//! ```rust
//!# use futures::FutureExt;
//!# use serde::{Serialize, Deserialize};
//!# use rand::{thread_rng, Rng};
//!# use tokio::time::Duration;
//!# use std::sync::Arc;
//!# use netty_rs::{Networker, SimpleDirectoryService, NetworkMessage, Connection, Action};
//!# // Custom error struct
//!# #[derive(Debug, Clone)]
//!# struct MyError;
//!# fn main() -> Result<(), MyError> {
//!#   tokio_test::block_on(async {
//!fn generate_challenge() -> Vec<u8> {
//!    // Generate the challenge ...
//!#            let mut arr = [0u8; 32];
//!#            thread_rng().fill(&mut arr[..]);
//!#            arr.to_vec()
//!}
//!
//!fn verify_challenge(a: &Vec<u8>) -> bool {
//!    // Verify challenge answer ...
//!#            let mut arr = [0u8; 32];
//!#            thread_rng().fill(&mut arr[..]);
//!#            arr.to_vec();
//!#            true
//!}
//!
//!fn sign(c: &Vec<u8>) -> Vec<u8> {
//!    // Sign the challenge
//!#            let mut arr = [0u8; 32];
//!#            thread_rng().fill(&mut arr[..]);
//!#            arr.to_vec()
//!}
//!
//!// Enum for the different types of messages we want to send
//!#[derive(Clone, Serialize, Deserialize, Debug, Eq, PartialEq)]
//!enum Content {
//!    Init,
//!    Challenge(Vec<u8>),
//!    Answer(Vec<u8>),
//!    Accept,
//!    Deny,
//!    Request,
//!    Response(i32),
//!    ProtocolError,
//!}
//!
//!let ds = SimpleDirectoryService::new();
//!let networker = Networker::new("127.0.0.1:8080".parse().unwrap(), ds,
//!    |handshake_msg: NetworkMessage<Content>, mut con: Connection<Content, MyError>| async move {
//!       // Perhaps you authenticate by producing a challenge and then
//!       // waiting for a response
//!       let challenge = generate_challenge();
//!       let message = handshake_msg.reply(Content::Challenge(challenge));
//!       let timeout = Duration::from_secs(2);
//!       // On timeout or other errors we just abort this whole process
//!       let response = con.send_message_await_reply(message, Some(timeout)).await?;
//!       if let Content::Answer(a) = &response.content {
//!           if verify_challenge(a) {
//!               let accept_msg = response.reply(Content::Accept);
//!               con.send_message(accept_msg).await?;
//!           } else {
//!               let deny_msg = response.reply(Content::Deny);
//!               con.send_message(deny_msg).await?;
//!           }
//!       } else {
//!           let deny_msg = response.reply(Content::Deny);
//!           con.send_message(deny_msg).await?;
//!       }
//!       // Return the id of this client
//!       let inner = Arc::try_unwrap(handshake_msg.from).unwrap_or_else(|e| (*e).clone());
//!        Ok(inner)
//!    },
//!    |message: NetworkMessage<Content>, mut con: Connection<Content, MyError>| async move {
//!        if let Content::Request = message.content {
//!            // Respond with the magical number for the meaning of life
//!            let response = message.reply(Content::Response(42));
//!            con.send_message(response).await?;
//!        } else {
//!            let response = message.reply(Content::ProtocolError);
//!            con.send_message(response).await?;
//!        }
//!        Ok(())
//!    }).await.map_err(|_| MyError)?;
//!networker.listen(true).await.map_err(|_| MyError)?;
//!// Send a message to ourselves
//!let first_message = NetworkMessage::new(
//!    Arc::new("127.0.0.1:8080".to_string()),
//!    Arc::new("127.0.0.1:8080".to_string()),
//!    Content::Init,
//!);
//!let timeout = Duration::from_secs(2);
//!let action = Action::new(
//!    |msg: NetworkMessage<Content>, mut con: Connection<Content, MyError>| {
//!        async move {
//!            if let Content::Challenge(c) = &msg.content {
//!                let answer = sign(c);
//!                let resp = msg.reply(Content::Answer(answer));
//!                let timeout = Duration::from_secs(2);
//!                let accept = con.send_message_await_reply(resp, Some(timeout)).await?;
//!                if let Content::Accept = accept.content {
//!                    Ok(())
//!                } else {
//!                    Err(MyError.into())
//!                }
//!            } else {
//!                Err(MyError.into())
//!            }
//!        }
//!        .boxed()
//!    },
//!);
//!networker
//!    .send_message(first_message, Some(timeout), Some(action))
//!    .await.map_err(|_| MyError)?;
//!Result::<(), MyError>::Ok(())
//!#    })?;
//!#    Ok(())
//!# }
//!```

// TODO: Add a retry-loop feature
// TODO: Consider making handshakes have a different content than regular messages
// TODO: If a channel dies then it will need to handshake again, this MIGHT be a problem
// TODO: Create function for handshaking on sending message
// TODO: We should verify the fields of the network message so that to and from are correct
// TODO: Make the handshake closure at least into a FnOnce, perhaps also the message closure. They
// are already cloned so they don't need to be FnMut to be called more than once
use futures::future::BoxFuture;
use futures::Future;
use log::debug;
use rand::distributions::Alphanumeric;
use rand::{thread_rng, Rng};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fmt::Debug;
use std::net::SocketAddr;
use std::net::ToSocketAddrs;
use std::sync::Arc;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::sync::oneshot::{channel as os_channel, Sender as OsSender};
use tokio::time::sleep;
use tokio::time::Duration;

/// Marker trait for errors that are returned by the handlers
pub trait HandlerError: Send + Sync + Debug + 'static + Clone {}

impl<T> HandlerError for T where T: Send + Sync + Debug + 'static + Clone {}

/// Errorkind in the error of netty-rs
#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
#[doc(hidden)]
pub enum ErrorKind<E: HandlerError> {
    Timeout,
    ProtocolBreak,
    Unspecified,
    SerializationError,
    NotFound,
    DirectoryService,
    HandlerError(E),
}

/// Error returned by netty-rs
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub struct Error<E: HandlerError> {
    /// The kind of error
    pub kind: ErrorKind<E>,
    /// The message provided with the error
    pub msg: String,
}

impl<T: HandlerError> From<T> for Error<T> {
    fn from(e: T) -> Self {
        Self::handler_error(e)
    }
}

impl<E: HandlerError> Error<E> {
    fn handler_error(e: E) -> Self {
        Self {
            kind: ErrorKind::HandlerError(e),
            msg: "Handler returned an error".to_string(),
        }
    }

    fn timeout<S: ToString>(msg: S) -> Self {
        Self {
            kind: ErrorKind::Timeout,
            msg: msg.to_string(),
        }
    }

    fn custom<S: ToString>(msg: S) -> Self {
        Self {
            kind: ErrorKind::Unspecified,
            msg: msg.to_string(),
        }
    }

    fn serialization_error<S: ToString>(msg: S) -> Self {
        Self {
            kind: ErrorKind::SerializationError,
            msg: msg.to_string(),
        }
    }

    fn directory_service_error<S: ToString>(msg: S) -> Self {
        Self {
            kind: ErrorKind::DirectoryService,
            msg: msg.to_string(),
        }
    }

    fn network_error<S: ToString>(msg: S) -> Self {
        Self {
            kind: ErrorKind::DirectoryService,
            msg: msg.to_string(),
        }
    }
}

/// Typedef for the result in netty-rs
type Result<T, E> = std::result::Result<T, Error<E>>;

/// Marker trait for the content contained in the [NetworkMessage](NetworkMessage) struct
pub trait NetworkContent:
    Serialize + DeserializeOwned + Send + Sync + Eq + PartialEq + 'static + Debug
{
}

impl<T> NetworkContent for T where
    T: Serialize + DeserializeOwned + Send + Sync + Eq + PartialEq + 'static + Debug
{
}

const MAX_SOCKET_BUF_SIZE: usize = 1500;
const DEFAULT_MESSAGE_TIMEOUT_MILLIS: u64 = 5000;
const CHANNEL_SIZE: usize = 100;

/// Netty-rs uses a pluggable directory service to translate from an id to an IP address and
/// port number. This is provided via this trait. A [SimpleDirectoryService](SimpleDirectoryService)
/// struct is provided which translates from any type that implements [ToSocketAddrs](ToSocketAddrs).
/// This includes strings which are in the format of for example "127.0.0.1:8080", which will allow consumers to
/// easily use strings or IP addresses as identifiers. If a more complicated lookup is required
/// then implementing this trait is always avaliable.
pub trait DirectoryService<N: Send + Sync, E: HandlerError>: Send + Sync {
    /// Translates names to Socket addresses
    fn translate(&self, name: &N) -> Result<SocketAddr, E>;
}

/// Directory service that translates Strings and socket addresses to socket addresses
/// Strings have to be in the format of "127.0.0.1:8080"
pub struct SimpleDirectoryService<
    S: ToSocketAddrs<Iter = std::vec::IntoIter<SocketAddr>> + Send + Sync,
> {
    _pd: std::marker::PhantomData<S>,
}

impl<S: ToSocketAddrs<Iter = std::vec::IntoIter<SocketAddr>> + Send + Sync>
    SimpleDirectoryService<S>
{
    /// Create a new [SimpleDirectoryService](SimpleDirectoryService)
    pub fn new() -> Self {
        Self {
            _pd: std::marker::PhantomData::<S>,
        }
    }
}

impl<S: ToSocketAddrs<Iter = std::vec::IntoIter<SocketAddr>> + Send + Sync, E: HandlerError>
    DirectoryService<S, E> for SimpleDirectoryService<S>
{
    /// Translates a `ToSocketAddrs` to the first `SocketAddr` it yields
    fn translate(&self, name: &S) -> Result<SocketAddr, E> {
        let mut sockets = name.to_socket_addrs().map_err(|_| {
            Error::directory_service_error("Could not get socket address from directory service")
        })?;
        let socket = sockets.next().ok_or_else(|| {
            Error::directory_service_error("Could not get socket address from directory service")
        })?;
        Ok(socket)
    }
}

type NetPack<T, E> = (
    NetworkMessage<T>,
    Option<Action<T, E>>,
    Option<Duration>,
    OsSender<Result<(), E>>,
);
type ConnectionPackage<T, E> = (
    NetworkMessage<T>,
    Option<Duration>,
    bool,
    OsSender<Result<Option<NetworkMessage<T>>, E>>,
);

/// This struct is the main API for using netty. It allows the creation of a server and the ability
/// to send messages to clients.
#[derive(Debug, Clone)]
pub struct Networker<T: NetworkContent + 'static, E: HandlerError> {
    tx: Sender<NetPack<T, E>>,
    // NOTE: command_tx currently sends a bool that represents if the networker should act as a
    // server or not.
    command_tx: Sender<(bool, OsSender<Result<(), E>>)>,
}

/// This struct represents the network messages sent and received, it can be created either from
/// the new `new` constructor if a fresh message is desired.
/// If a reply is desired then the `reply` method should be used.
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Serialize, Deserialize)]
pub struct NetworkMessage<T: NetworkContent> {
    /// The receiver id, this is currently a string but this may change to a generic representing
    /// an id in the future
    pub to: Arc<String>,
    /// The sender id, this is currently a string but this may change to a generic representing
    /// an id in the future
    pub from: Arc<String>,
    /// The message id, used to reply to this message
    pub id: Arc<String>,
    /// If this is set to `Some` then this message is a reply to the contained ID, else it is a
    /// fresh message
    pub reply: Option<Arc<String>>,
    /// The content of the message
    #[serde(bound(deserialize = "T: DeserializeOwned"))]
    pub content: T,
}

fn new_id() -> String {
    let rand_string: String = thread_rng()
        .sample_iter(&Alphanumeric)
        .take(30)
        .map(char::from)
        .collect();
    rand_string
}

impl<T: NetworkContent> NetworkMessage<T> {
    /// Create a new fresh message
    pub fn new(to: Arc<String>, from: Arc<String>, content: T) -> Self {
        Self {
            to,
            from,
            id: Arc::new(new_id()),
            reply: None,
            content,
        }
    }

    /// Used to construct replies to the provided message.
    pub fn reply(&self, content: T) -> Self {
        Self {
            to: self.from.clone(),
            from: self.to.clone(),
            id: Arc::new(new_id()),
            reply: Some(self.id.clone()),
            content,
        }
    }
}

async fn create_socket_task<T: NetworkContent, M, F, E: HandlerError>(
    mut socket: TcpStream,
    handle_message: M,
) -> Result<Sender<NetPack<T, E>>, E>
where
    F: Future<Output = HandlerResult<(), E>> + Send,
    M: FnMut(NetworkMessage<T>, Connection<T, E>) -> F + Send + Sync + Clone + 'static,
{
    let (tx, mut rx): (Sender<NetPack<T, E>>, Receiver<NetPack<T, E>>) = channel(CHANNEL_SIZE);
    tokio::spawn(async move {
        debug!("Starting up a new socket task");
        let (tx, mut reaction_rx): (
            Sender<ConnectionPackage<T, E>>,
            Receiver<ConnectionPackage<T, E>>,
        ) = channel(CHANNEL_SIZE);
        let mut awaiting_reply: HashMap<Arc<String>, OsSender<NetworkMessage<T>>> = HashMap::new();
        loop {
            //let mut buf = [0; MAX_SOCKET_BUF_SIZE];
            // This task is responsible for:
            // 1. Listening to the channel from the main manager thread for messages to send on
            //      this socket, sending the message out on the socket and recording which (if
            //      any) reaction was interested in responses to that message as well as
            //      recording when a timeout happens and then returning an error on the channel
            // 2. Listening to the reaction_rx channel for NetworkMessages to send and
            //    recording which message (if any) message the reaction is interested in
            //    listening to.
            // 3. Listening to the socket and parsing the data to NetworkMessages, figuring out
            //    if the parsed message is a response to any channels and in that case routing
            //    it to that channel, else calling a new handle_message instance with the
            //    message
            tokio::select! {
                Some((msg, timeout, want_reply, os_tx)) = reaction_rx.recv() => {
                    // 2) Send the message and then create a timeout that sends a timeout error
                    //    on expiry. Also create a one-shot channel chain where we record which
                    //    message we are waiting for and give it the transmission end of the
                    //    chain
                    debug!("Socket task - Received a request to send a message on reaction thread");
                    let msg_s = match serde_json::to_string(&msg) {
                        Ok(r) => r,
                        Err(_) => {
                            let e = Error::serialization_error(format!("Could not serialize: {:?}", msg));
                            if let Err(e) = os_tx.send(Err(e)) {
                                debug!("Oneshot return channel did not stay open: {:?}", e);
                            }
                            continue;
                        },
                    };
                    match socket.write_all(msg_s.as_bytes()).await {
                        Ok(()) => (),
                        Err(_) => {
                            let e = Error::network_error(format!("Could not write on socket"));
                            if let Err(e) = os_tx.send(Err(e)) {
                                debug!("Oneshot return channel did not stay open: {:?}", e);
                            }
                            continue;
                        },
                    }
                    if want_reply {
                        let timeout_time = match timeout {
                            Some(t) => t,
                            // If no timeout is provided then we use a default
                            None => Duration::from_millis(DEFAULT_MESSAGE_TIMEOUT_MILLIS),
                        };
                        let (hm_tx, hm_rx) = os_channel();
                        awaiting_reply.insert(msg.id, hm_tx);
                        tokio::spawn(async move {
                            let timeout = tokio::time::sleep(timeout_time);
                            tokio::pin!(timeout);
                            tokio::select! {
                                Ok(msg) = hm_rx => {
                                    if let Err(e) = os_tx.send(Ok(Some(msg))) {
                                        debug!("Oneshot return channel did not stay open: {:?}", e);
                                    }
                                },
                                _ = timeout => {
                                    if let Err(e) = os_tx.send(Err(Error::timeout("Did not recieve a response in time!"))) {
                                        debug!("Oneshot return channel did not stay open: {:?}", e);
                                    }
                                }
                            }
                        });
                    } else {
                        if let Err(e) = os_tx.send(Ok(None)) {
                            debug!("Oneshot return channel did not stay open: {:?}", e);
                        }
                    }
                },
                Result::<NetworkMessage<T>, E>::Ok(msg) = read_message(&mut socket) => {
                    // 3) Listens to the socket and parses the data to NetworkMessages then
                    //    looks in the awaiting_reply hashmap to see if there is any connection
                    //    that are waiting for reply to a message that the parsed message is
                    //    replying to, in that case send the message on that channel if it
                    //    remains open, if the channel is closed discard the message.
                    //    If there is no connection that awaits reply from this message then
                    //    start a new connection for this message
                    match &msg.reply {
                        Some(id) => {
                            match awaiting_reply.remove(id) {
                                Some(tx) => {
                                    debug!("Sending message to waiter {:?}", msg);
                                    if let Err(e) = tx.send(msg) {
                                        //Discard the channel and message
                                        debug!("Discarded the message due to: {:?}", e);
                                    }
                                }
                                None => {
                                    //Discard the channel and message
                                    debug!("Could not find channel to pass message on");
                                }
                            };
                        },
                        None => {
                            debug!("Did not find any waiters for {:?}", msg);
                            let con = Connection {
                                sender: tx.clone(),
                            };
                            let mut handle_message = handle_message.clone();
                            tokio::spawn(async move {
                                if let Err(e) = handle_message(msg, con).await {
                                    debug!("Handle message returned an error {:?}", e);
                                }
                            });
                        },
                    };
                },
                Some((msg, react, timeout, return_tx)) = rx.recv() => {
                    debug!("On socket task - Received a message to send {:?}", msg);
                    // 1. Listens to the main-thread communication channel and sends the
                    //    message on the internal reaction channel to schedule it. If a
                    //    reaction is passed then listens to the oneshot channel for a reply
                    //    and provides this to the reaction call
                    let (os_tx, os_rx) = os_channel();
                    let want_reply = react.is_some();
                    tx.send((msg, timeout, want_reply, os_tx)).await.expect("Networker internal error due to reaction channel being closed");
                    let tx = tx.clone();
                    tokio::spawn(async move {
                        let r = os_rx.await.expect("Networker internal error, awaiting os_rx channel but transmitter closed");
                        match r {
                            Ok(r) => {
                                if want_reply {
                                    let react = react.expect("Networker unreachable state");
                                    let msg = r.expect("Network unreachable state - received no message while expecting reply");
                                    let con = Connection {
                                        sender: tx.clone(),
                                    };
                                    let result = react.0(msg, con).await;
                                    return_tx.send(result).expect("Networker internal error - networker did not listen to return channel");
                                } else {
                                    return_tx.send(Ok(())).expect("Networker internal error - networker did not listen to return channel");
                                }
                            },
                            Err(e) => {
                                return_tx.send(Err(e)).expect("Networker internal error - networker did not listen to return channel");
                            }
                        };
                    });
                }
            }
        }
    });
    Ok(tx)
}

async fn read_message<T: NetworkContent, E: HandlerError>(
    socket: &mut TcpStream,
) -> Result<NetworkMessage<T>, E> {
    let mut buf = [0; MAX_SOCKET_BUF_SIZE];
    match socket.read(&mut buf).await {
        Ok(n) => match String::from_utf8(buf[..n].to_vec()) {
            Ok(s) => match serde_json::from_str(&s) {
                Ok(s) => return Ok(s),
                Err(_) => {
                    return Err(Error::serialization_error(
                        "Could not deserialize recieved message",
                    ));
                }
            },
            Err(e) => {
                return Err(Error::serialization_error(format!(
                    "Could not convert to utf-8 - {:?}",
                    e
                )));
            }
        },

        Err(e) => {
            return Err(Error::network_error(format!(
                "Could not read from socket - {:?}",
                e
            )));
        }
    }
}

async fn process_socket<T: NetworkContent, H, M, FH, FM, E: HandlerError>(
    mut socket: TcpStream,
    mut handle_handshake: H,
    handle_message: M,
) -> Result<(String, Sender<NetPack<T, E>>), E>
where
    FH: Future<Output = HandlerResult<String, E>> + Send,
    FM: Future<Output = HandlerResult<(), E>> + Send,
    H: FnMut(NetworkMessage<T>, Connection<T, E>) -> FH + Send + Sync + 'static,
    M: FnMut(NetworkMessage<T>, Connection<T, E>) -> FM + Send + Sync + Clone + 'static,
{
    let msg = read_message(&mut socket).await?;
    let (handshake_tx, mut handshake_rx) = channel(CHANNEL_SIZE);
    let con = Connection {
        sender: handshake_tx.clone(),
    };
    let listen_for_messages = async {
        //This loop never returns Ok, instead it sends successes on the channel to the
        //Connection thread in order to drive that thread forward
        loop {
            if let Some((msg, timeout, want_reply, os_tx)) = handshake_rx.recv().await {
                let msg = match serde_json::to_string(&msg) {
                    Ok(s) => s,
                    Err(_) => {
                        let e =
                            Error::serialization_error(format!("Could not serialize {:?}", msg));
                        if let Err(e) = os_tx.send(Err(e)) {
                            debug!("Oneshot return channel did not stay open: {:?}", e);
                        }
                        continue;
                    }
                };
                match socket.write_all(msg.as_bytes()).await {
                    Ok(()) => (),
                    Err(_) => {
                        let e = Error::network_error("Could not send over network socket");
                        if let Err(e) = os_tx.send(Err(e)) {
                            debug!("Oneshot return channel did not stay open: {:?}", e);
                        }
                        continue;
                    }
                };
                if !want_reply {
                    if let Err(e) = os_tx.send(Ok(None)) {
                        debug!("Oneshot return channel did not stay open: {:?}", e);
                    }
                    continue;
                } else {
                    let timeout = sleep(
                        timeout.unwrap_or(Duration::from_millis(DEFAULT_MESSAGE_TIMEOUT_MILLIS)),
                    );
                    tokio::pin!(timeout);
                    tokio::select! {
                        s = read_message(&mut socket) => {
                            match s {
                                Ok(s) => {
                                    if let Err(e) = os_tx.send(Ok(Some(s))) {
                                        debug!("Oneshot return channel did not stay open: {:?}", e);
                                    }
                                },
                                Err(e) => {
                                    if let Err(e) = os_tx.send(Err(e)) {
                                        debug!("Oneshot return channel did not stay open: {:?}", e);
                                    }
                                }
                            }
                        },
                        _ = timeout => {
                            if let Err(e) = os_tx.send(Err(Error::timeout("Did not recieve a response in time"))) {
                                debug!("Oneshot return channel did not stay open: {:?}", e);
                            }
                        }
                    }
                }
            }
        }
    };
    let r = tokio::select! {
        Result::<T, E>::Err(e) = listen_for_messages => return Err(e),
        r = handle_handshake(msg, con) => {
            match r {
                Ok(r) => r,
                Err(e) => {
                    return Err(e);
                }
            }
        }
    };
    let tx = create_socket_task(socket, handle_message).await?;
    Ok((r, tx))
}

type HandlerResult<T, E> = Result<T, E>;
impl<T: NetworkContent, E: HandlerError> Networker<T, E> {
    /// Creates a new networker using. `address` is the network socket address that the server should
    /// listen to. `directory_service` is the directory service to use in order to translate IDs to
    /// addresses. `handle_handshakes` is a closure that is called when a new connection is
    /// recieved which sends a `NetworkMessage`. This closure should authenticate if appropriate
    /// and do other handshake and setup related things. `handle_messages` is a closure that is
    /// called for all messages received from a connection that has already been handshaked.
    pub async fn new<H, M, FH, FM>(
        address: SocketAddr,
        directory_service: impl DirectoryService<String, E> + 'static,
        handle_handshakes: H,
        handle_messages: M,
    ) -> Result<Networker<T, E>, E>
    where
        FM: Future<Output = HandlerResult<(), E>> + Send,
        FH: Future<Output = HandlerResult<String, E>> + Send,
        M: FnMut(NetworkMessage<T>, Connection<T, E>) -> FM + Send + Sync + Clone + 'static,
        H: FnMut(NetworkMessage<T>, Connection<T, E>) -> FH + Send + Sync + Clone + 'static,
    {
        let (net_tx, mut thread_rx): (Sender<NetPack<T, E>>, Receiver<NetPack<T, E>>) =
            channel(CHANNEL_SIZE);
        let (command_tx, mut command_rx): (
            Sender<(bool, OsSender<Result<(), E>>)>,
            Receiver<(bool, OsSender<Result<(), E>>)>,
        ) = channel(CHANNEL_SIZE);
        let mut name_channel_hm = HashMap::new();
        let mut listener: Option<TcpListener> = None;
        tokio::spawn(async move {
            loop {
                tokio::select! {
                    Some((should_be_server, os_tx)) = command_rx.recv() => {
                        if should_be_server {
                            listener = match TcpListener::bind(address).await {
                                Ok(r) => Some(r),
                                Err(e) => {
                                    match os_tx.send(Err(Error::network_error(format!(
                                                    "Could not listen to address: {} due to: {:?}",
                                                    address, e
                                                    )))) {
                                        Ok(()) => (),
                                        Err(_) => {
                                            debug!("Internal networker error - could not return result from turning on/off server");
                                        },
                                    };
                                    continue;
                                }
                            };
                        } else {
                            listener = None;
                        }
                        match os_tx.send(Ok(())) {
                            Ok(()) => {
                            },
                            Err(_) => {
                                debug!("Internal networker error - could not send on return channel from request to start listening");
                            },
                        };
                    }
                    // NOTE: We only listen for new connections in the case where the networker
                    // should act like a server
                    Ok((socket, _)) = async {
                        if let Some(listener) = &listener {
                            listener.accept().await
                        } else {
                            // If the networker is not a server then we will wait forever, if this
                            // changes then this future will be cancelled
                            let forever = futures::future::pending();
                            let () = forever.await;
                            unreachable!("Networker unreachable state - tried to listen to a non-existent server");
                        }
                    } => {
                        debug!("Received a TCP connection");
                        let (name, tx) = match process_socket(socket, handle_handshakes.clone(), handle_messages.clone()).await {
                            Ok(r) => r,
                            Err(e) => {
                                debug!("Could not establish contact {:?}", e);
                                continue;
                            },
                        };
                        debug!("Handshake finished peer name is: {}", name);
                        name_channel_hm.insert(name, tx);
                    },
                    Some((message, react, timeout, os_tx)) = thread_rx.recv() => {
                        // Here we need to find the correct socket to send to. And we should
                        // maybe allow a "ALL" option to send towards
                        match name_channel_hm.get(&*message.to) {
                            // There is already a recipiant connected with that name
                            Some(tx) => {
                                if let Err(e) = tx.send((message, react, timeout, os_tx)).await {
                                    debug!("{}", e);
                                }
                            }
                            // No connection to provided recipiant was found
                            None => {
                                // TODO: We should allow consumers to provide a do_handshake
                                // function which is run automatically be ran on sending a new
                                // message to an uninitiated peer. This function should return
                                // a Result containing the name of the peer. If no such
                                // function is provided then we skip that part
                                match directory_service.translate(&message.to) {
                                    Ok(address) => {
                                        let socket = match TcpStream::connect(address).await
                                            .map_err(|_| Error::network_error(format!("Could not connect to address {:?}", address))) {
                                                Ok(s) => s,
                                                Err(e) => {
                                                    if let Err(_) = os_tx.send(Err(e)) {
                                                        debug!("Could not return error send on one-shot channel");
                                                    }
                                                    continue;
                                                }
                                            };
                                        // Spawn a new thread with a new tcp connection and
                                        // send the message
                                        let tx = match create_socket_task(socket, handle_messages.clone()).await {
                                            Ok(tx) => tx,
                                            Err(e) => {
                                                if let Err(_) = os_tx.send(Err(e)) {
                                                    debug!("Could not return error send on one-shot channel");
                                                }
                                                continue;
                                            }
                                        };
                                        let name = message.to.clone();
                                        if let Err(e) = tx.send((message, react, timeout, os_tx)).await {
                                            debug!("{}", e);
                                        }
                                        name_channel_hm.insert((*name).clone(), tx);
                                    },
                                    Err(e) => {
                                        if let Err(_) = os_tx.send(Err(e)) {
                                            debug!("Could not return error send on one-shot channel");
                                        }
                                        continue;
                                    }
                                }
                            }
                        }
                    },
                }
            }
        });

        Ok(Networker {
            tx: net_tx,
            command_tx,
        })
    }

    /// Sends a message and then reacts to the response with the action and then returns the
    /// last message returned
    pub async fn send_message(
        &self,
        message: NetworkMessage<T>,
        timeout: Option<Duration>,
        react: Option<Action<T, E>>,
    ) -> Result<(), E> {
        let (os_tx, os_rx) = os_channel();
        if let Err(_) = self.tx.send((message, react, timeout, os_tx)).await {
            debug!("Could not send to networker");
        }
        match os_rx.await.expect("Oneshot transmitter dropped in socket") {
            Ok(()) => Ok(()),
            Err(e) => Err(e),
        }
    }

    /// Starts the server
    pub async fn listen(&self, should_listen: bool) -> Result<(), E> {
        let (tx, rx) = os_channel();
        match self.command_tx.send((should_listen, tx)).await {
            Ok(()) => (),
            Err(_) => {
                return Err(Error::custom(format!(
                    "Internal error - Could change listening status due to channel being down"
                )))
            }
        };
        match rx.await {
            Ok(r) => r,
            Err(_) => return Err(Error::custom(format!("Internal error - Could not get response from listening call due to return channel closing prematurely"))),
        }
    }
}

/// Action contains a closure that handles the communication on a channel
pub struct Action<
    T: Send + Sync + Serialize + DeserializeOwned + Eq + PartialEq + Debug + 'static,
    E: HandlerError,
>(
    Box<
        dyn FnOnce(NetworkMessage<T>, Connection<T, E>) -> BoxFuture<'static, HandlerResult<(), E>>
            + Send
            + Sync,
    >,
);

impl<T: NetworkContent, E: HandlerError> Action<T, E> {
    /// Creates a new Action
    pub fn new<
        F: FnOnce(NetworkMessage<T>, Connection<T, E>) -> BoxFuture<'static, HandlerResult<(), E>>
            + 'static
            + Send
            + Sync,
    >(
        f: F,
    ) -> Self {
        Self(Box::new(f))
    }
}

/// A connection serves as the main API to specify how a network conversation should look. It
/// allows sending messages to the recipiant and awaiting their response using the two methods
/// `send_message` and `send_message_await_reply`.
pub struct Connection<T: NetworkContent, E: HandlerError> {
    sender: Sender<ConnectionPackage<T, E>>,
}

impl<T: NetworkContent, E: HandlerError> Connection<T, E> {
    /// Send a message over the connection without waiting for a reply, returning upon successfully
    /// sending the message out.
    pub async fn send_message(&mut self, msg: NetworkMessage<T>) -> Result<(), E> {
        let (tx, rx) = os_channel();
        match self.sender.send((msg, None, false, tx)).await {
            Ok(()) => (),
            Err(_) => {
                panic!("Internal error - could not send on internal channel",);
            }
        };
        let r = match rx.await {
            Ok(r) => r,
            Err(_) => {
                panic!("Internal error - internal return channel was closed before receiving a message");
            }
        };
        r.map(|r| {
            if r.is_some() {
                panic!("Unreachable state - expected None but was provided a network message")
            }
        })
    }

    /// Send a message over the connection while waiting for a reply, a `Result` is returned with
    /// the replying `NetworkMessage`.
    pub async fn send_message_await_reply(
        &mut self,
        msg: NetworkMessage<T>,
        timeout: Option<Duration>,
    ) -> Result<NetworkMessage<T>, E> {
        let (tx, rx) = os_channel();
        match self.sender.send((msg, timeout, true, tx)).await {
            Ok(()) => (),
            Err(_) => {
                return Err(Error::custom(
                    "Internal error - could not send on internal channel",
                ));
            }
        };
        let r = match rx.await {
            Ok(r) => r,
            Err(_) => {
                return Err(Error::custom("Internal error - internal return channel was closed before receiving a message"));
            }
        };
        r.map(|r| r.expect("Expecting a network message as response but None was provided"))
    }
}