use crossbeam_channel::{after, unbounded, Sender};
use failure::ResultExt;
use messages::{IsRequest, IsResponse, Message, MessageExt, Notification,
               Request, Response};
use serialport::SerialPort;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
use {serialport, std, Error, ErrorKind, RegisteredSenders, Result,
     RxHandle, SenderRegistration};

#[derive(Default, Clone)]
pub(crate) struct SequenceHolder {
    // Once AtomicU8 is out of experimental, it can be used here
    // See: https://github.com/rust-lang/rust/issues/32976
    seq: Arc<Mutex<u8>>,
}

impl SequenceHolder {
    pub fn request_sequence_number(&mut self) -> Result<u8> {
        let mut lock = self.seq.lock().unwrap();
        let seq: u8 = *lock;
        *lock = seq + 1;
        Ok(seq)
    }
}

#[derive(Clone, Debug)]
pub(crate) enum RequestItem {
    Request {
        request: Request,
        options: u8,
        sequence: u8,
    },
    Finish,
}

#[derive(Clone, Debug)]
pub struct ResponseItem {
    pub response: Response,
    pub sequence: u8,
}

#[derive(Clone, Debug)]
pub struct NotificationItem {
    pub notification: Notification,
    pub sequence: u8,
}

pub struct Connection {
    port: Box<SerialPort>,
    tx_request_sender: Sender<RequestItem>,

    register_response_sender: Sender<SenderRegistration<ResponseItem>>,
    register_notification_sender:
        Sender<SenderRegistration<NotificationItem>>,

    rx_shutdown_sender: Sender<()>,

    tx_handle: Option<thread::JoinHandle<()>>,
    rx_handle: Option<thread::JoinHandle<()>>,

    sequence: SequenceHolder,
}

impl Drop for Connection {
    fn drop(&mut self) {
        // make the threads finish their loop
        self.rx_shutdown_sender.send(());
        self.tx_request_sender.send(RequestItem::Finish);

        // join the threads
        let tx_result = if let Some(handle) = self.tx_handle.take() {
            handle.join()
        } else {
            Ok(())
        };
        let rx_result = if let Some(handle) = self.rx_handle.take() {
            handle.join()
        } else {
            Ok(())
        };

        if tx_result.is_err() {
            warn!("Failure joining xio tx thread");
        }
        if rx_result.is_err() {
            warn!("Failure joining xio rx thread");
        }
    }
}

impl Connection {
    /// Create a Connection, and give it an identifier.
    ///
    /// The identifier gets inserted in the thread name, helping to identify
    /// threads when debugging.
    ///
    /// The resulting thread name is something like "xio_tx:<identifier>" or
    /// "xio_rx:<identifier>".
    pub fn new_with_identifier(
        port: &str,
        identifier: &str,
    ) -> Result<Self> {
        let mut settings = serialport::SerialPortSettings::default();
        settings.timeout = Duration::from_millis(100);
        let port = serialport::open_with_settings(port, &settings)?;

        let mut tx_port = port.try_clone()?;
        let mut rx_port = port.try_clone()?;

        let (register_response_sender, register_response_receiver) =
            unbounded::<SenderRegistration<ResponseItem>>();
        let (register_notification_sender, register_notification_receiver) =
            unbounded::<SenderRegistration<NotificationItem>>();

        let (tx_request_sender, tx_request_receiver) =
            unbounded::<RequestItem>();
        let (rx_shutdown_sender, rx_shutdown_receiver) = unbounded::<()>();
        let tx_handle = thread::Builder::new()
            .name(format!("xio_tx:{}", identifier))
            .spawn(move || {
                for item in tx_request_receiver {
                    match item {
                        RequestItem::Request {
                            request,
                            options,
                            sequence,
                        } => {
                            debug!("Sending request {:?}", request);
                            if let Err(e) = request.write_to(
                                &mut tx_port,
                                options,
                                sequence,
                            ) {
                                warn!(
                                    "Error writing to xio tx port, \
                                     closing down: {:?}",
                                    e
                                );
                                return;
                            }
                        }
                        RequestItem::Finish => {
                            return;
                        }
                    }
                }
            })
            .context(ErrorKind::ThreadCreation)?;

        let rx_handle = {
            thread::Builder::new()
                .name(format!("xio_rx:{}", identifier))
                .spawn(move || {
                    let mut responses =
                        RegisteredSenders::new(register_response_receiver);
                    let mut notifications = RegisteredSenders::new(
                        register_notification_receiver,
                    );

                    loop {
                        match Message::read_from(&mut rx_port) {
                            Ok((sequence, msg)) => match msg {
                                Message::Request(r) => {
                                    warn!(
                                        "Received a request message from \
                                         a xio device which should never \
                                         happen: {:?}",
                                        r
                                    );
                                }
                                Message::Response(response) => {
                                    responses.process_pending();

                                    let response = ResponseItem {
                                        sequence,
                                        response,
                                    };
                                    responses.send(response);
                                }
                                Message::Notification(notification) => {
                                    notifications.process_pending();
                                    let notification = NotificationItem {
                                        sequence,
                                        notification,
                                    };
                                    notifications.send(notification);
                                }
                            },
                            Err(e) => {
                                if let Some(()) =
                                    rx_shutdown_receiver.try_recv()
                                {
                                    return;
                                }
                                if e.kind() != std::io::ErrorKind::TimedOut
                                {
                                    // might happen if a shutdown is in progress
                                    warn!("Error received: {:?}", e);
                                }
                            }
                        }
                    }
                })
                .context(ErrorKind::ThreadCreation)?
        };
        Ok(Connection {
            port,
            tx_request_sender,
            register_response_sender,
            register_notification_sender,
            rx_shutdown_sender,
            tx_handle: Some(tx_handle),
            rx_handle: Some(rx_handle),
            sequence: SequenceHolder::default(),
        })
    }

    pub fn new(port: &str) -> Result<Self> {
        Self::new_with_identifier(port, port)
    }

    pub fn add_response_rx(&mut self) -> RxHandle<ResponseItem> {
        RxHandle::new(&self.register_response_sender)
    }

    pub fn add_notification_rx(&mut self) -> RxHandle<NotificationItem> {
        RxHandle::new(&self.register_notification_sender)
    }

    pub fn get_handle(&mut self) -> Result<Handle> {
        let seq = self.sequence.clone();
        let rx = Arc::new(Mutex::new(self.add_response_rx()));
        let tx = self.tx_request_sender.clone();
        Ok(Handle { seq, tx, rx })
    }
}

pub trait SendAndReceive<Q, S>
where
    Q: IsRequest<Response = S>,
    S: IsResponse<Request = Q>,
{
    type Error;
    fn send_and_receive(
        &mut self,
        request: Q,
    ) -> std::result::Result<S, Self::Error>;
}

impl SendAndReceive<Vec<Request>, Vec<Response>> for Handle {
    type Error = Error;
    fn send_and_receive(
        &mut self,
        request: Vec<Request>,
    ) -> Result<Vec<Response>> {
        let expected_sequences = request
            .into_iter()
            .map(|request| self.send(0u8, request))
            .collect::<Result<Vec<_>>>()?;

        let responses = expected_sequences
            .into_iter()
            .map(|sequence| self.receive_next(sequence))
            .collect::<Result<Vec<_>>>()?;

        Ok(responses)
    }
}

impl SendAndReceive<Request, Response> for Handle {
    type Error = Error;
    fn send_and_receive(&mut self, request: Request) -> Result<Response> {
        let mut responses = self.send_and_receive(vec![request])?;
        let response = responses.drain(..).next();
        response.ok_or_else(|| ErrorKind::ProgrammerError.into())
    }
}

impl<Q, S> SendAndReceive<Q, S> for Handle
where
    Q: Into<Request> + IsRequest<Response = S>,
    S: super::messages::TryFromResponse + IsResponse<Request = Q>,
{
    type Error = Error;
    fn send_and_receive(&mut self, request: Q) -> Result<S> {
        let response = self.send_and_receive(request.into())?;
        Ok(S::try_from_response(response)?)
    }
}

#[derive(Clone)]
pub struct Handle {
    seq: SequenceHolder,
    tx: Sender<RequestItem>,
    rx: Arc<Mutex<RxHandle<ResponseItem>>>,
}

impl Handle {
    pub fn send(&mut self, options: u8, request: Request) -> Result<u8> {
        let sequence = self.seq.request_sequence_number()?;
        self.tx.send(RequestItem::Request {
            request,
            options,
            sequence,
        });
        Ok(sequence)
    }

    fn receive_next(&self, expected_sequence: u8) -> Result<Response> {
        let timeout = Duration::from_millis(100);

        let mut i = 0usize;

        while i < 20usize {
            let rx_lock = self.rx.lock().unwrap();
            let rx = rx_lock.receiver();
            select! {
                recv(rx, item) => {
                    if let Some(ResponseItem{sequence,response}) = item {
                        if expected_sequence == sequence {
                            return Ok(response);
                        }
                    } else {
                        return Err(ErrorKind::ChannelReceiverDisconnected.into());
                    }
                }
                recv(after(timeout)) => {
                    i = i + 1usize;
                }
            }
        }
        Err(ErrorKind::Timeout.into())
    }
}