use futures::{Future, Async, Poll};
use {ReadState, WriteState, read_length, read, Error, Connection, error, Parameters};
use std::str::from_utf8;
use byteorder::{BigEndian, ByteOrder};
use std::fmt;
use msg::*;
use tokio_io::io::flush;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_timer::Delay;


/// A result row, implementing an iterator on fields. Each field is a
/// slice of bytes.
#[derive(Clone)]
pub struct Row<'a> {
    len: u16,
    buf: &'a [u8],
    n: usize,
    pos: usize,
}

impl<'a> fmt::Debug for Row<'a> {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        let v: Vec<_> = self.clone().map(|x| from_utf8(x)).collect();
        v.fmt(f)
    }
}


impl<'a> Row<'a> {
    /// Number of fields in this row.
    pub fn len(&self) -> usize {
        self.len as usize
    }
}

impl<'a> Iterator for Row<'a> {
    type Item = &'a [u8];
    fn next(&mut self) -> Option<Self::Item> {
        if self.pos >= self.buf.len() {
            None
        } else {
            let pos = self.pos;
            let len = BigEndian::read_i32(&self.buf[pos..]);
            if len >= 0 {
                let len = len as usize;
                let result = &self.buf[pos + 4..pos + 4 + len];
                self.pos += 4 + len;
                self.n += 1;
                Some(result)
            } else {
                self.pos += 4;
                // self.n += 1;
                Some(&self.buf[pos..pos])
            }
        }
    }
}


/// Result rows, returned by the `query` method of
/// `Connection`.
pub struct Rows<W: AsyncRead + AsyncWrite, F: HandleRow, P: AsRef<Parameters>> {
    #[doc(hidden)]
    pub connection: Option<(Connection<W, P>, F)>,
    #[doc(hidden)]
    pub timeout: Option<Delay>,
    #[doc(hidden)]
    pub last_row_was_halted: bool,
}

/// Result from a request.
#[derive(Debug)]
pub enum RowResult<'a> {
    /// An actual row.
    Row(Row<'a>),

    /// A completion signal, with the number of rows modified.
    Complete(u32),

    /// A termination signal, sent when there are no more rows.
    NoMoreRows,

    /// An error.
    Err(&'a str),
}


/// Requests that can handle a resulting row.
#[allow(unused_variables)]
pub trait HandleRow {
    /// Called on each row.
    fn row(&mut self, row: Row) -> bool { true }
    /// Called at the end of the request, with the number of modified rows.
    fn complete<'a>(&mut self, modified: u32) {
        debug!("complete, modified {:?}", modified);
    }
    /// Called when the server is ready for the next request.
    fn ready_for_query<'a>(&mut self) {
        debug!("ready_for_query");
    }
    /// This function gets called when the server returns an error.
    fn err(&mut self, err: &str) {
        debug!("err: {:?}", err);
    }
}

impl<F: Fn(Row) -> bool> HandleRow for F {
    fn row(&mut self, row: Row) -> bool {
        (self)(row)
    }
}

#[derive(PartialEq, Eq)]
enum RowPoll {
    // The last row was halted.
    Halted,
    // All rows have been read (and possibly discarded after a halt).
    Ready,
    // There are still rows to be read.
    NotReady,
}

impl<W: AsyncWrite + AsyncRead, F: HandleRow, P: AsRef<Parameters>> Future for Rows<W, F, P> {
    type Item = ((Connection<W, P>, F), Option<Error>);
    type Error = Error;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        loop {
            debug!("rows poll");
            let mut was_err = None;
            let mut async_not_ready = false;
            let ready = if let Some((ref mut c, ref mut f)) = self.connection {
                match c.poll_rows(self.last_row_was_halted, f) {
                    Ok(Async::Ready(RowPoll::Halted)) => {
                        self.last_row_was_halted = true;
                        false
                    }
                    Ok(Async::Ready(RowPoll::Ready)) => true,
                    Ok(Async::Ready(RowPoll::NotReady)) => false,
                    Ok(Async::NotReady) => { async_not_ready = true; false },
                    Err(e) => {
                        error!("{:?}", e);
                        was_err = Some(e);
                        true
                    }
                }
            } else {
                unreachable!()
            };
            debug!("rows: {:?} {:?} {:?}", ready, was_err, async_not_ready);
            if ready {
                return Ok(Async::Ready((self.connection.take().unwrap(), was_err)));
            }

            if async_not_ready {
                if let Some(ref mut timeout) = self.timeout {
                    if let Ok(Async::Ready(())) = timeout.poll() {
                        debug!("Rows timeout fired (file {}, line {})", file!(), line!());
                        return Ok(Async::Ready((self.connection.take().unwrap(), Some(Error::Timeout))))
                    }
                }
                return Ok(Async::NotReady)
            } else {
                // We've received something, the timeout can be cancelled.
                self.timeout = None
            }

        }
    }
}


impl<W: AsyncRead + AsyncWrite, P: AsRef<Parameters>> Connection<W, P> {
    fn poll_rows<R: HandleRow>(&mut self,
                               last_row_was_halted: bool,
                               f: &mut R)
                               -> Poll<RowPoll, Error> {
        loop {
            match self.write.take() {
                Some(WriteState::Write { mut w }) => {
                    match try!(w.poll()) {
                        Async::Ready((stream, buf)) => {
                            self.write = Some(WriteState::Flush {
                                flush: flush(stream),
                                buf
                            });
                        },
                        Async::NotReady => {
                            self.write = Some(WriteState::Write { w });
                            return Ok(Async::NotReady);
                        }
                    }
                }
                Some(WriteState::Flush { mut flush, buf }) => {
                    match try!(flush.poll()) {
                        Async::Ready(w) => {
                            self.write = Some(WriteState::Idle { w, buf });
                            break
                        },
                        Async::NotReady => {
                            self.write = Some(WriteState::Flush { flush, buf });
                            return Ok(Async::NotReady);
                        }
                    }
                }
                Some(state) => {
                    self.write = Some(state);
                    break
                },
                None => {}
            }
        }
        // If we're done writing, start reading.

        loop {
            match self.read.take() {
                Some(ReadState::ReadLength(mut r)) => {
                    match try!(r.poll()) {
                        Async::Ready((w, buf)) => {
                            self.read = Some(read(w, buf))
                        }
                        Async::NotReady => {
                            self.read = Some(ReadState::ReadLength(r));
                            return Ok(Async::NotReady)
                        }
                    }
                }
                Some(ReadState::Read { msg, mut body }) => {

                    let (stream, buf) = match try!(body.poll()) {
                        Async::Ready((w, buf)) => (w, buf),
                        Async::NotReady => {
                            self.read = Some(ReadState::Read {
                                msg: msg,
                                body: body,
                            });
                            return Ok(Async::NotReady);
                        }
                    };
                    match msg {
                        /*MSG_ROW_DESCRIPTION => {
                        // This message type only happens when we send a "quick query".
                        {
                        let mut reader = reader(&buf[..]);
                        let n_fields = try!(reader.read_i16());
                        debug!("n_fields: {:?}", n_fields);
                        self.columns.clear();
                        for _ in 0..n_fields {
                        let name = try!(reader.read_string());
                        let col = Column {
                        table_id: try!(reader.read_i32()),
                        column_num: try!(reader.read_i16()),
                        data_type: try!(reader.read_i32()),
                        data_size: try!(reader.read_i16()),
                        type_modifier: try!(reader.read_i32()),
                        format: try!(reader.read_i16())
                    };
                        debug!("name {:?} {:?}", name, col);
                        self.columns.push(col);
                    }
                    }
                        self.state = Some(read_length(stream, buf));
                        Ok(Async::Ready(false))
                    }*/
                        MSG_DATA_ROW if !last_row_was_halted => {
                            debug!("data row");
                            let query_finished = {
                                let len = BigEndian::read_i16(&buf[..]);
                                let row = Row {
                                    len: len as u16,
                                    buf: &buf[..],
                                    n: 0,
                                    pos: 2,
                                };
                                if f.row(row) {
                                    RowPoll::NotReady
                                } else {
                                    RowPoll::Halted
                                }
                            };
                            self.read = Some(read_length(stream, buf));
                            return Ok(Async::Ready(query_finished))
                        }
                        MSG_COMMAND_COMPLETE => {
                            debug!("command_complete, buf = {:?}", buf);
                            // self.columns.clear();
                            // f.handle_row(RowResult::NoMoreRows);
                            {
                                if let Some(i) = (&buf[..]).iter().position(|&x| x == 0) {
                                    let buf = &buf[..i];
                                    let mut it = buf.split(|x| *x == b' ');
                                    match (it.next(), it.next(), it.next()) {

                                        (Some(b"TABLE"), _, _) => {
                                            f.complete(0);
                                        }
                                        (Some(b"INSERT"), _, Some(rows)) |
                                        (Some(_), Some(rows), _) => {
                                            debug!("rows: {:?}", from_utf8(rows));
                                            let rows:Result<u32, Error> =
                                                from_utf8(rows).map_err(From::from)
                                                .and_then(|x| x.parse::<u32>().map_err(From::from));
                                            if let Ok(rows) = rows {
                                                // debug!("{:?} {:?}", rows, parsed);
                                                f.complete(rows);
                                            }
                                        }
                                        _ => {
                                            debug!("command complete: {:?}", buf);
                                        }
                                    }
                                } else {
                                    return Err(Error::Protocol);
                                }
                            }
                            self.read = Some(read_length(stream, buf));
                            return Ok(Async::Ready(RowPoll::NotReady))
                        }
                        MSG_READY_FOR_QUERY => {
                            // self.columns.clear();
                            f.ready_for_query();
                            self.read = Some(read_length(stream, buf));
                            return Ok(Async::Ready(RowPoll::Ready))
                        }
                        MSG_ERROR => {
                            let mut into = String::new();
                            try!(error(&buf, &mut into));
                            self.read = Some(read_length(stream, buf));
                            return Err(Error::Postgres { message: into })
                        }
                        msg => {
                            debug!("unknown message {:?}", msg as char);
                            self.read = Some(read_length(stream, buf));
                            return Ok(Async::Ready(RowPoll::NotReady))
                        }
                    }
                }
                None => unreachable!()
            }
        }
    }
}