//! This crate provides the core ZMTP data types, as well as
//! Readers and Writers for those data types.

extern crate byteorder;

use self::byteorder::{ByteOrder, BigEndian};
use std::borrow::{Borrow, Cow};
use std::collections::HashMap;
use std::cmp;
use std::fmt::{Debug, Display, Formatter, Result};
use std::io;
use std::io::{ErrorKind, Read, Write};
use std::slice::{Iter, IterMut};
use std::{u8, u64, usize};

/// The number of bytes in a u64
pub const BYTES_PER_U64: usize = 8;

/// The largest number that fits inside a u8
pub const U8_MAX: usize = u8::MAX as usize;

pub const NULL_MECHANISM: &'static [u8; 20] =    // "NULL" as bytes, then
    b"NULL\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0"; // 16 bytes of 0-padding

/*----------------------------------------------------------------------------*/
/*                 Useful traits                                              */
/*----------------------------------------------------------------------------*/
/// This trait allows peeking (i.e. non-destructive reading) in [`Read`]ers.
/// Non-destructive here refers to the fact that when calling `peek(slice)`
/// on the same `slice` multiple times in a row, every call should return
/// the same data.
///
/// [`Read`]: https://doc.rust-lang.org/std/io/trait.Read.html
pub trait Peek : Read + Debug {
    /// Peek `buf.len()` bytes and write them to `buf`.
    fn peek(&mut self, mut buf: &mut [u8])  -> io::Result<usize>;
}

impl<'a> Peek for &'a [u8] {
    fn peek(&mut self, mut buf: &mut [u8])  -> io::Result<usize> {
        if self.len() == 0 {
            return Err(io::Error::new(ErrorKind::InvalidData,
                                      "Peeking from empty buf"));
        }
        let end = cmp::min(buf.len(), self.len());
        for i in 0..end { buf[i] = self[i] }
        Ok(end)
    }
}

/// This trait is used by [`Peer`] to actually read and write data.
///
/// [`Peer`]: ./struct.Peer.html
pub trait PeerSource: Read + Write + Debug + Send {}
impl<S: Read + Write + Debug + Send> PeerSource for S {}

/*----------------------------------------------------------------------------*/
/*                 Primitive ZMQ socket type identifiers                      */
/*----------------------------------------------------------------------------*/
#[derive(PartialEq, Eq, Debug, Copy, Clone)]
pub enum SocketType {
    PAIR    = 0x00,
    PUB     = 0x01,
    SUB     = 0x02,
    REQ     = 0x03,
    REP     = 0x04,
    DEALER  = 0x05,
    ROUTER  = 0x06,
    PULL    = 0x07,
    PUSH    = 0x08,
    XPUB    = 0x09,
    XSUB    = 0x0A,

    // TODO: The STREAM socket variant is defined in ZeroMQ, but it remains
    //     to be seen what it can do, since the RFC doesn't mention it. See
    //  https://github.com/zeromq/libzmq/blob/2b6200c49e1265f87019be8a99352529c9d2378c/include/zmq.h#L259
    // STREAM  = 0x0B,
}

impl SocketType {
    pub fn as_bytes(&self) -> Vec<u8> { self.to_string().as_bytes().to_vec() }

    fn from_slice(bytes: &[u8]) -> SocketType {
        match String::from_utf8_lossy(bytes).borrow() {
            "PAIR" => SocketType::PAIR,
            "PUB" => SocketType::PUB,
            "SUB" => SocketType::SUB,
            "REQ" => SocketType::REQ,
            "REP" => SocketType::REP,
            "DEALER" => SocketType::DEALER,
            "ROUTER" => SocketType::ROUTER,
            "PULL" => SocketType::PULL,
            "PUSH" => SocketType::PUSH,
            "XPUB" => SocketType::XPUB,
            "XSUB" => SocketType::XSUB,
            t => panic!("Unknown socket type: {:?}", t),
        }
    }

    pub fn can_connect_to(&self, t: SocketType) -> bool {
        use SocketType::*;
        match *self {
            // Implements a check to see if a socket type is compatible
            // with another. See http://zmtp.org/page:read-the-docs#toc16
            REQ =>    t == REP || t == ROUTER,
            REP =>    t == REQ || t == DEALER,
            DEALER => t == REP || t == DEALER || t == ROUTER,
            ROUTER => t == REQ || t == DEALER || t == ROUTER,
            PUB =>    t == SUB || t == XSUB,
            XPUB =>   t == SUB || t == XSUB,
            SUB =>    t == PUB || t == XPUB,
            XSUB =>   t == PUB || t == XPUB,
            PUSH =>   t == PULL,
            PULL =>   t == PUSH,
            PAIR =>   t == PAIR,
        }
    }
}

impl Display for SocketType {
    fn fmt(&self, f: &mut Formatter) -> Result { write!(f, "{:?}", self) }
}

/*----------------------------------------------------------------------------*/
/*                 Core data types                                            */
/*----------------------------------------------------------------------------*/
#[derive(PartialEq, Eq, Debug)]
pub enum Size { Short(u8), Long(u64) }

impl Size {
    pub fn as_usize(&self) -> usize {
        match *self {
            Size::Short(size) => size as usize,
            Size::Long(size) => {
                if size >= usize::MAX as u64 {
                    panic!("Size is larger than this system can handle: {}",
                           size)
                }
                size as usize
            },
        }
    }
}


#[derive(PartialEq, Eq, Debug)]
pub enum FrameKind { Cmd, Msg }


/// A frame, commonly used to create [`Message`]s and [`Command`]s.
///
/// [`Message`]: struct.Message.html
/// [`Command`]: struct.Command.html
#[derive(PartialEq, Eq, Debug)]
pub struct Frame {
    kind: FrameKind,
    size: Size,
    is_last: bool,
    body: Vec<u8>,
}

impl Frame {
    pub fn new() -> Self {
        Frame {
            kind: FrameKind::Msg,
            size: Size::Short(0),
            is_last: false,
            body: vec![],
        }
    }

    pub fn from_str(s: &str) -> Frame { Self::from_bytes(s.as_bytes()) }

    pub fn from_bytes(bytes: &[u8]) -> Frame { Self::new().extend_body(bytes) }

    pub fn get_size(&self) -> &Size { &self.size }

    pub fn get_kind(&self) -> &FrameKind { &self.kind }

    pub fn get_is_last(&self) -> bool { self.is_last }

    pub fn body_len(&self) -> usize { self.view_body().len() }

    pub fn view_body<'l>(&'l self) -> &'l [u8] { &self.body }

    pub fn edit_body<'l>(&'l mut self) -> &'l mut Vec<u8> { &mut self.body }

    pub fn get_body(self) -> Vec<u8> { self.body }

    pub fn kind(mut self, kind: FrameKind) -> Self {
        self.kind = kind;
        self
    }

    pub fn is_last(mut self, is_last: bool) -> Self {
        self.is_last = is_last;
        self
    }

    pub fn set_size(mut self, size: Size) -> Self {
        self.size = size;
        self
    }

    pub fn set_body(&mut self, bytes: Vec<u8>) -> &mut Self {
        self.body.clear();
        self.body.extend(bytes);
        self
    }

    pub fn extend_body(mut self, bytes: &[u8]) -> Self {
        self.body.extend_from_slice(bytes);
        let new_size = self.size.as_usize() + bytes.len();
        self.size(new_size)
    }

    fn size(mut self, size: usize) -> Self {
        self.size = match size {
            0...U8_MAX =>  Size::Short(size as u8),
            _ =>           Size::Long(size as u64),
        };
        self
    }
}


#[derive(PartialEq, Eq, Debug)]
pub struct Command {
    name: Vec<u8>,
    data: Vec<u8>,
    metadata: HashMap<String, Vec<u8>>,
}

impl Command {
    pub fn new() -> Self {
        Command {
            name: vec![],
            data: vec![],
            metadata: HashMap::new(),
        }
    }

    pub fn named<S: Into<String>>(name: S) -> Self {
        Command::new().str_name(name)
    }

    pub fn metadata(mut self, name: &str, value: Vec<u8>) -> Self {
        self.metadata.insert(name.to_string(), value);
        self
    }

    pub fn add_metadata<S>(&mut self, name: S, value: &[u8])
                           -> &mut Self where S: Into<String> {
        self.metadata.insert(name.into(), value.to_owned());
        self
    }

    pub fn get_socket_type(&self) -> Option<SocketType> {
        self.metadata.get("Socket-Type")
            .map(|bytes| SocketType::from_slice(bytes))
    }

    pub fn get_metadata(&self) -> &HashMap<String, Vec<u8>> { &self.metadata }

    pub fn get_name_str<'l>(&'l self) -> Cow<'l, str> {
        String::from_utf8_lossy(self.get_name())
    }

    pub fn get_name(&self) -> &[u8] { &self.name }

    pub fn get_data(&self) -> &[u8] { &self.data }

    pub fn name(mut self, name: &[u8]) -> Self {
        assert!(name.len() <= 0xFF);
        self.name.clear();
        self.name.extend_from_slice(name);
        self
    }

    pub fn str_name<S: Into<String>>(mut self, name: S) -> Self {
        let string = name.into();
        assert!(string.len() <= 0xFF);
        self.name = string.into_bytes();
        self
    }

    pub fn data(mut self, bytes: &[u8]) -> Self {
        self.data.extend_from_slice(bytes);
        self
    }

    pub fn add_data(&mut self, bytes: &[u8]) -> &mut Self {
        self.data.extend_from_slice(bytes);
        self
    }

    fn metadata_size(&self) -> usize {
        let mut f = Frame::new();
        for (name, val) in self.get_metadata().iter() {
            let name_bytes = name.as_bytes();
            let name_len = name_bytes.len();
            assert!(1 <= name_len && name_len <= 255);
            let val_len = val.len() as u32;
            let mut val_len_bytes = [0 as u8; 4];
            BigEndian::write_u32(&mut val_len_bytes, val_len);
            f.body.extend_from_slice(&[name_len as u8]);
            f.body.extend_from_slice(name_bytes);
            f.body.extend_from_slice(&val_len_bytes);
            f.body.extend_from_slice(val);
        }
        f.body.len()
    }

    pub fn size(&self) -> Size {
        const NAME_SIZE_BYTE: usize = 1;
        match NAME_SIZE_BYTE + self.name.len() + self.data.len()
            + self.metadata_size() {
            size@0...U8_MAX => Size::Short(size as u8),
            size => Size::Long(size as u64),
        }
    }
}


#[derive(PartialEq, Eq, Debug)]
pub struct Message(Vec<Frame>);

impl Message {
    pub fn new(mut frame: Frame) -> Self {
        frame.kind = FrameKind::Msg;
        frame.is_last = true;
        Message(vec![frame])
    }

    pub fn empty_frame() -> Self { Self::new(Frame::new()) }

    pub fn from_str(string: &str) -> Self {
        Message::new(Frame::new().extend_body(string.as_bytes()))
    }

    pub fn as_string(self) -> String {
        let mut string_bytes = vec![];
        for frame in self.0.into_iter() {
            let mut bytes = frame.get_body();
            string_bytes.append(&mut bytes);
        }
        String::from_utf8(string_bytes).unwrap()
    }

    pub fn add_frame(&mut self, mut frame: Frame) -> &mut Self {
        assert!(self.frame_count() >= 1);
        self.last_frame().is_last = false;
        frame.is_last = true;
        frame.kind = FrameKind::Msg;
        self.0.push(frame);
        self
    }

    pub fn last_frame(&mut self) -> &mut Frame {
        self.0.last_mut().expect("Expected the last msg frame")
    }

    pub fn frame_count(&self) -> usize { self.frames().len() }

    pub fn frames(&self) -> &[Frame] { &self.0 }

    pub fn into_frame_iter(self) -> <Vec<Frame> as IntoIterator>::IntoIter {
        self.0.into_iter()
    }

    pub fn frame_iter(&self) -> Iter<Frame> { self.frames().iter() }
}


#[derive(PartialEq, Eq, Debug, Clone, Copy)]
pub struct Greeting {
    pub signature: [u8; 10],
    pub version:   (u8, u8),
    pub mechanism: [u8; 20],
    pub as_server: bool,
    pub filler:    [u8; 31],
}

impl Greeting {
    pub fn new() -> Self {
        Greeting {
            signature: [0xFF, 0x0,0x0,0x0,0x0, 0x0,0x0,0x0,0x1, 0x7F],
            version:   (0x3, 0x1),
            mechanism: *NULL_MECHANISM,
            as_server: false,
            filler:    [0x0; 31],
        }
    }

    pub fn signature(&mut self, padding: &[u8]) -> &mut Self {
        assert!(padding.len() == 8);
        for (idx, byte) in padding.iter().enumerate() {
            self.signature[idx + 1] = *byte;
        }
        self
    }

    pub fn version(&mut self, major: u8, minor: u8) -> &mut Self {
        self.version = (major, minor);
        self
    }

    pub fn mechanism(&mut self, mechanism: &[u8]) -> &mut Self {
        assert!(mechanism.len() == 20);
        for (idx, byte) in mechanism.iter().enumerate() {
            self.mechanism[idx] = *byte;
        }
        self
    }

    pub fn as_server(&mut self, as_server: bool) -> &mut Self {
        self.as_server = as_server;
        self
    }

    pub fn filler(&mut self, filler: &[u8]) -> &mut Self {
        assert!(filler.len() == 31);
        for (idx, byte) in filler.iter().enumerate() {
            self.filler[idx] = *byte;
        }
        self
    }
}


#[derive(PartialEq, Eq, Debug)]
pub struct Handshake(Vec<Command>);

impl Handshake {
    pub fn new(command: Command) -> Self { Handshake(vec![command]) }

    pub fn num_commands(&self) -> usize { self.0.len() }

    pub fn commands(&self) -> Iter<Command> { self.0.iter() }

    pub fn add_command(&mut self, command: Command) -> &mut Self {
        self.0.push(command);
        self
    }
}



#[derive(PartialEq, Eq, Debug)]
pub enum TrafficItem { Cmd(Command), Msg(Message) }


#[derive(PartialEq, Eq, Debug)]
pub struct Traffic(Vec<TrafficItem>);

impl Traffic {
    pub fn new() -> Self { Traffic(vec![]) }

    pub fn from_items(items: Vec<TrafficItem>) -> Self { Traffic(items) }

    pub fn items(&self) -> Iter<TrafficItem> { self.0.iter() }

    pub fn items_mut(&mut self) -> IterMut<TrafficItem> { self.0.iter_mut() }

    pub fn item_count(&self) -> usize { self.0.len() }

    pub fn add(&mut self, item: TrafficItem) { self.0.push(item); }
}

/*----------------------------------------------------------------------------*/
/*                 Validators                                                 */
/*----------------------------------------------------------------------------*/

/// A validation result.
#[derive(PartialEq, Eq, Debug, Clone)]
pub enum ValidateResult {
    /// The bytes were correctly validated.
    Ok,

    /// The bytes have a \0 terminator in them.
    NullError(u8, char, Vec<u8>),

    /// The bytes have a digit in them.
    DigitError(u8, char, Vec<u8>),

    /// The bytes slice is too short.
    TooShortError(u64),

    /// The bytes slice is too long.
    TooLongError(u64),

    /// The bytes slice contains one of `-` `_` `.` or `+`.
    SpecialError(u8, char, Vec<u8>),

    /// The bytes slice contains a character that is not in
    /// any other category and thus by default disallowed.
    InvalidError(u8, char, Vec<u8>)
}

impl ValidateResult {
    pub fn expect<S: Into<String>>(&self, msg: S) {
        match *self {
            ValidateResult::Ok => { },
            _ => panic!("{:?}", msg.into()),
        }
    }
}

pub struct CommandValidator(ValidateResult);

impl CommandValidator {
    pub fn validate(cmd: &Command) -> ValidateResult {
        CommandValidator(ValidateResult::Ok)
            .validate_name(cmd.get_name())
            .validate_data(cmd.get_data())
            .result()
    }

    pub fn result(self) -> ValidateResult { self.0 }

    pub fn is_upper(b: u8) -> bool { 'A' as u8 <= b && b <= 'Z' as u8 }

    pub fn is_lower(b: u8) -> bool { 'a' as u8 <= b && b <= 'z' as u8 }

    pub fn is_alpha(b: u8) -> bool { Self::is_upper(b) || Self::is_lower(b) }

    pub fn is_digit(b: u8) -> bool { '0' <= b as char && b as char <= '9' }

    pub fn is_vchar(b: u8) -> bool { 0x21 <= b && b <= 0x7E }

    pub fn is_nonnull_char(b: u8) -> bool { 0x01 <= b && b <= 0x7F }

    pub fn is_name_char(b: u8) -> bool {
        Self::is_alpha(b) || Self::is_special(b)
    }

    pub fn is_null(b: u8) -> bool { b == 0 }

    pub fn is_special(b: u8) -> bool {
        ['-', '_', '.', '+'].contains(&(b as char))
    }

    fn validate_name(mut self, bytes: &[u8]) -> Self {
        assert_eq!(self.0, ValidateResult::Ok);
        let len = bytes.len();
        if bytes.len() < 1 {
            self.0 = ValidateResult::TooShortError(len as u64);
            return self
        }
        if bytes.len() > u8::MAX as usize {
            self.0 = ValidateResult::TooLongError(len as u64);
            return self
        }
        for byte in bytes {
            if Self::is_alpha(*byte) { continue /* This byte is Ok */ }
            return self.find_cause(*byte, bytes);
        }
        self
    }

    fn find_cause(mut self, byte: u8, bytes: &[u8]) -> Self {
        if self.0 != ValidateResult::Ok { return self }
        let (c, bvec) = (byte as char, bytes.to_vec());
        self.0 = if Self::is_digit(byte) {
            ValidateResult::DigitError(byte, c, bvec)
        } else if Self::is_null(byte) {
            ValidateResult::NullError(byte, c, bvec)
        } else if Self::is_special(byte) {
            ValidateResult::SpecialError(byte, c, bvec)
        } else { ValidateResult::InvalidError(byte, c, bvec) };
        self
    }

    #[inline]
    #[allow(unused)]
    fn validate_data(mut self, data: &[u8]) -> Self {
        // Don't validate the data in any way.
        self
    }
}

//  LocalWords:  ZMTP ers TcpStream impls PeerSource buf plen