// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.
#![deny(missing_docs,
        trivial_casts,
        unstable_features,
        unused_import_braces)]

//! A fully-asynchronous Postgres client. Postgres has a notion of
//! "statements", which are named or unnamed SQL commands, and of
//! "portals", where a portal is an instance of a result. All queries
//! to the database follow the same sequence: parse, bind, execute,
//! close (portal and/or statement).
//!
//! Both statements and portals can be named or unnamed, although
//! prototype code might be easier to write using only the `query`
//! method of the `Connection` type, which does the correct sequence
//! using the unnamed statement and unnamed portal.
//!
//! Contrarily to synchronous clients, this client needs to store
//! requests in the event loop while doing asynchronous I/O. The
//! precise type for this storage needs to be provided by the user of
//! this crate, and needs to implement both the `Request` and
//! `HandleRow` traits.
//!
//! This crate provides two API levels:
//!
//! - A higher-level one based using `spawn_connection`, which uses an
//! existing event loop. This API requires an existing instance of
//! `Request+HandleRow`.
//!
//! - A lower-level one based on the `Connection` type, which might be
//! used to build more complex, or more direct pipelines, such as
//! tunneling through another protocol, or waiting for a query to be
//! executed by the server.
//!
//!
//! ```
//! extern crate tokio_core;
//! extern crate pleingres;
//! extern crate futures;
//! extern crate env_logger;
//! extern crate uuid;
//!
//! use uuid::Uuid;
//! use std::sync::Arc;
//! use std::net::ToSocketAddrs;
//! use futures::Future;
//!
//! // A `Request` is the type used for communication with the
//! // client event loop. It is used both for sending input and
//! // receiving output.
//!
//! struct Request {
//!     login: String,
//!     id: Option<Uuid>
//! }
//!
//! // Requests must implement the `pleingres::Request` trait,
//! // meaning they can be converted to commands to be sent to
//! // the server.
//! impl pleingres::Request for Request {
//!     fn request(&mut self, mut buf: pleingres::Buffer) {
//!         buf.bind("SELECT id FROM users WHERE login=$1", &[&self.login]).execute(0);
//!     }
//! }
//!
//! impl pleingres::HandleRow for Request {
//!     fn row(&mut self, mut row: pleingres::Row) -> bool {
//!         if let Some(id) = row.next() {
//!            self.id = Some(pleingres::FromSql::from_sql(id).unwrap())
//!         }
//!         true
//!     }
//! }
//!
//! fn main() {
//!     env_logger::init().unwrap();
//!     let p = Arc::new(pleingres::Parameters {
//!         addr: "::1:5432".to_socket_addrs().unwrap().next().unwrap(),
//!         user: "pe".to_string(),
//!         password: "password".to_string(),
//!         database: Some("pijul".to_string()),
//!         max_wait: std::time::Duration::from_millis(1000),
//!         idle_timeout: Some(std::time::Duration::from_millis(20_000)),
//!         tcp_keepalive: Some(10000),
//!     });
//!     let mut l = tokio_core::reactor::Core::new().unwrap();
//!
//!     let db:pleingres::Handle<Request> =
//!         pleingres::spawn_connection(&mut l, p.clone()).unwrap();
//!     l.run(db
//!           .send_request(
//!             Request { login: "me".to_string(), id: None }
//!           )
//!           .and_then(|dbreq| {
//!               if let Some(ref id) = dbreq.id {
//!                   println!("id: {:?}", id)
//!               }
//!               futures::finished(())
//!           })).unwrap()
//! }
//! ```
//!
//! Alternatively, the `pleingres::Request` can be implemented using
//! the sql plugin. On the above example, we would replace
//!
//! ``` ignore
//! struct Request {
//!     login: String,
//!     id: Option<Uuid>
//! }
//!
//! impl pleingres::Request for Request {
//!     fn request(&mut self, mut buf: pleingres::Buffer) {
//!         buf.bind("SELECT id FROM users WHERE login=$1", &[&self.login]).execute(0);
//!     }
//! }
//! ```
//!
//! With just
//!
//! ``` ignore
//! #[sql("SELECT id FROM users WHERE login = $login")]
//! struct Request {
//!     login: String,
//!     id: Option<Uuid>
//! }
//! ```


#[macro_use]
extern crate futures;
extern crate env_logger;
extern crate tokio_core;
extern crate tokio_io;
#[macro_use]
extern crate log;
extern crate byteorder;
extern crate md_5;
extern crate uuid;
extern crate chrono;
extern crate generic_array;
extern crate digest;

use byteorder::{WriteBytesExt, BigEndian, ByteOrder};
use std::sync::Arc;
use futures::{Future, Poll, Async, Sink, AsyncSink};
use std::io::Write;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_io::io::{read_exact, ReadExact, WriteHalf, ReadHalf};
use futures::sync::*;
use std::net::SocketAddr;
use tokio_core::reactor::Timeout;
use std::time::Duration;
use std::collections::HashMap;

const MAJOR_VERSION: u16 = 3;
const MINOR_VERSION: u16 = 0;

mod error;
pub use error::*;

mod statement;
pub use statement::*;

mod rows;
pub use rows::*;

#[allow(missing_docs)]
mod msg;
use msg::*;

mod buffer;
pub use buffer::*;

mod spawned;
pub use spawned::*;

trait WritePostgresExt {
    fn write_string(&mut self, s: &[u8]) -> Result<(), std::io::Error>;
}

impl<W: Write> WritePostgresExt for W {
    fn write_string(&mut self, s: &[u8]) -> Result<(), std::io::Error> {
        assert!(s.iter().all(|&c| c != 0));
        try!(self.write(s));
        try!(self.write(&[0]));
        Ok(())
    }
}

/// Connection configuration: network address, user, database and
/// options. The connection is unencrypted.
pub struct Parameters {
    /// Network address of the server
    pub addr: SocketAddr,
    /// User name on the server
    pub user: String,
    /// Password for that user
    pub password: String,
    /// Database to connect to
    pub database: Option<String>,
    /// Maximum wait duration on the connection.
    pub max_wait: Duration,
    /// TCP keep alive, in milliseconds.
    pub tcp_keepalive: Option<u32>,
    /// Duration before reconnecting an idle connection.
    pub idle_timeout: Option<Duration>,
}

enum ReadState<W: AsyncRead> {
    ReadLength(ReadExact<ReadHalf<W>, Vec<u8>>),
    Read {
        msg: u8,
        body: ReadExact<ReadHalf<W>, Vec<u8>>,
    },
}
enum WriteState<W: AsyncWrite> {
    Idle { w: WriteHalf<W>, buf: Vec<u8> },
    Write { w: tokio_io::io::WriteAll<WriteHalf<W>, Vec<u8>> },
    Flush { flush: tokio_io::io::Flush<WriteHalf<W>>, buf: Vec<u8> },
}

use std::borrow::Cow;

/// An active connection to a database.
pub struct Connection<W: AsyncWrite + AsyncRead> {
    parameters: Arc<Parameters>,
    handle: tokio_core::reactor::Handle,
    // columns: Vec<Column>,
    write: Option<WriteState<W>>,
    read: Option<ReadState<W>>,
    // state: Option<State<W>>,
    process_id: Option<i32>,
    secret_key: Option<i32>,
    parsed_queries: HashMap<Cow<'static, str>, usize>
}

/// A future ultimately resolving into an established connection.
struct Connecting<W: AsyncWrite + AsyncRead>(Option<Connection<W>>);

impl<W: AsyncWrite + AsyncRead> Future for Connecting<W> {
    type Item = Connection<W>;
    type Error = Error;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        loop {
            debug!("connecting poll");
            if let Some(ref mut c) = self.0 {
                try_ready!(c.poll_connecting());
            }
            return Ok(Async::Ready(self.0.take().unwrap()))
        }
    }
}


const AUTH_MD5_PASSWORD: u32 = 5;
const AUTH_OK: u32 = 0;

/// A readiness signal sent by the server.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
#[repr(u8)]
pub enum Wait {
    /// The last statement was executed.
    CommandComplete = b'C',

    /// The last parse command is done.
    Parse = b'1',

    /// The last bind command is done.
    Bind = b'2',

    /// The server is ready for the next query.
    Ready = b'Z',

    /// The last close command was completed.
    CloseComplete = b'3',
}


fn read_length<A: AsyncRead>(stream: ReadHalf<A>, mut buf: Vec<u8>) -> ReadState<A> {
    buf.resize(5, 0);
    ReadState::ReadLength(read_exact(stream, buf))
}

fn read<A: AsyncRead>(stream: ReadHalf<A>, mut buf: Vec<u8>) -> ReadState<A> {
    let msg = buf[0];
    let n = BigEndian::read_u32(&buf[1..]) as usize;
    debug!("n = {:?}", n);
    buf.resize(n - 4, 0);
    ReadState::Read {
        msg: msg,
        body: read_exact(stream, buf),
    }
}


/// Future resolving to the connection once a command has been sent to the server.
pub struct WriteFuture<W: AsyncWrite + AsyncRead>(Option<Connection<W>>);

impl<W: AsyncWrite + AsyncRead> Connection<W> {
    /// Starts a connection on the given stream (usually a socket, but
    /// other instances of `Write`, such as a tunnel, work too). This
    /// function returns a future resolving into an established
    /// connection.
    fn new(handle: tokio_core::reactor::Handle, stream: W, parameters: Arc<Parameters>, parsed_queries: HashMap<Cow<'static, str>, usize>) -> Connecting<W> {
        let mut v = Vec::new();
        v.extend(&[0, 0, 0, 0]); // Length
        v.write_u16::<BigEndian>(MAJOR_VERSION).unwrap();
        v.write_u16::<BigEndian>(MINOR_VERSION).unwrap();

        v.write_string(b"user").unwrap();
        v.write_string(parameters.user.as_bytes()).unwrap();
        if let Some(ref database) = parameters.database {
            v.write_string(b"database").unwrap();
            v.write_string(database.as_bytes()).unwrap();
        }
        // if let Some(ref options) = parameters.options {
        // v.write_string(b"options").unwrap();
        // v.write_string(options.as_bytes()).unwrap();
        // }
        v.write(&[0]).unwrap();
        let len = v.len() as u32;
        BigEndian::write_u32(&mut v, len);
        let (read_half, write_half) = stream.split();
        Connecting(Some(Connection {
            write: Some(WriteState::Write { w: tokio_io::io::write_all(write_half, v) }),
            read: Some(read_length(read_half, Vec::new())),
            parameters: parameters,
            handle: handle,
            // columns: Vec::new(),
            process_id: None,
            secret_key: None,
            parsed_queries: parsed_queries
        }))
    }


    fn rows<F: HandleRow>(self, f: F, timeout: Option<Timeout>) -> Result<Rows<W, F>, Error> {
        // let timeout = Timeout::new(self.parameters.max_wait, &self.handle)?;
        Ok(Rows {
            connection: Some((self, f)),
            timeout: timeout,
            last_row_was_halted: false,
        })
    }

    fn timeout(&self) -> Result<Timeout, Error> {
        Ok(Timeout::new(self.parameters.max_wait, &self.handle)?)
    }

    /// Perform a complete query, i.e. parse, bind and execute the statement.
    pub fn query<F: HandleRow>(mut self, q: &'static str, args: &[&ToSql], f: F) -> Result<Rows<W, F>, Error> {
        {
            let mut buf = self.buffer();
            buf.bind(q, args).execute(0);
        }
        match self.write.take() {
            Some(WriteState::Idle { w, buf }) => {
                self.write = Some(WriteState::Write { w: tokio_io::io::write_all(w, buf) });
            }
            _ => unreachable!()
        }
        let timeout = self.timeout()?;
        Ok(self.rows(f, Some(timeout))?)
    }

    /// Write the current buffer to this connection. The resulting
    /// future resolves into `self` once the message has been written.
    pub fn write(mut self) -> WriteFuture<W> {
        match self.write.take() {
            Some(WriteState::Idle { w, buf }) => {
                self.write = Some(WriteState::Write { w: tokio_io::io::write_all(w, buf) });
                WriteFuture(Some(self))
            }
            _ => unreachable!()
        }
    }

    fn poll_connecting(&mut self) -> Poll<(), Error> {

        // First write anything that needs to be written.
        loop {
            match self.write.take() {
                Some(WriteState::Write { mut w }) => {
                    debug!("write {}: {}", file!(), line!());
                    let (stream, buf) = match try!(w.poll()) {
                        Async::Ready(s) => s,
                        Async::NotReady => {
                            self.write = Some(WriteState::Write { w });
                            return Ok(Async::NotReady);
                        }
                    };
                    self.write = Some(WriteState::Flush { flush: tokio_io::io::flush(stream), buf });
                    continue
                }
                Some(WriteState::Flush { mut flush, buf }) => {
                    debug!("flush {}: {}", file!(), line!());
                    let w = match try!(flush.poll()) {
                        Async::Ready(s) => s,
                        Async::NotReady => {
                            self.write = Some(WriteState::Flush { flush, buf });
                            return Ok(Async::NotReady);
                        }
                    };
                    self.write = Some(WriteState::Idle { w, buf });
                }
                Some(idle) => self.write = Some(idle),
                None => unreachable!(),
            }

            match self.read.take() {
                Some(ReadState::ReadLength(mut r)) => {
                    debug!("readlength: {} {}", file!(), line!());
                    match try!(r.poll()) {
                        Async::Ready((connection, buffer)) => {
                            self.read = Some(read(connection, buffer));
                        },
                        Async::NotReady => {
                            self.read = Some(ReadState::ReadLength(r));
                            return Ok(Async::NotReady);
                        }
                    }
                }
                Some(ReadState::Read { msg, mut body }) => {
                    debug!("read {} {}", line!(), msg as char);
                    let (stream, buf) = match try!(body.poll()) {
                        Async::Ready(s) => s,
                        Async::NotReady => {
                            self.read = Some(ReadState::Read {
                                msg: msg,
                                body: body,
                            });
                            return Ok(Async::NotReady);
                        }
                    };

                    debug!("poll connecting, read, {:?}", buf);
                    match msg {
                        MSG_AUTH => {
                            self.auth_request(&buf);
                            self.read = Some(read_length(stream, buf));
                        }
                        MSG_PARAMETER_STATUS => {
                            try!(parameter_status(&buf));
                            self.read = Some(read_length(stream, buf));
                        }
                        MSG_BACKEND_KEY_DATA => {
                            self.backend_key_data(&buf);
                            self.read = Some(read_length(stream, buf));
                        }
                        MSG_READY_FOR_QUERY => {
                            debug!("ready for query");
                            self.read = Some(read_length(stream, buf));
                            return Ok(Async::Ready(()))
                        }
                        MSG_ERROR => {
                            let mut into = String::new();
                            try!(error(&buf, &mut into));
                            // self.state = Some(read_length(stream, buf));
                            // Ok(Async::Ready(true))
                            return Err(Error::Postgres { message: into })
                        }
                        msg => {
                            debug!("unknown message {:?}", msg);
                            self.read = Some(read_length(stream, buf));
                        }
                    }
                },
                None => unreachable!()
            }
        }
    }

    fn auth_request(&mut self, read_buf: &[u8]) {
        // Authentication request.
        match self.write.take() {
            Some(WriteState::Idle { w, mut buf }) => {
                debug!("auth request: buf: {:?}", buf);
                let auth_type = BigEndian::read_u32(&read_buf[..]);
                match auth_type {
                    AUTH_OK => {
                        debug!("auth ok");
                        self.write = Some(WriteState::Idle { w, buf })
                    }
                    AUTH_MD5_PASSWORD => {
                        debug!("md5 password");

                        let md5 = md5(&read_buf[4..],
                                      self.parameters.user.as_bytes(),
                                      self.parameters.password.as_bytes());
                        buf.clear();
                        buf.push(MSG_PASSWORD);
                        buf.extend(&[0, 0, 0, 0]);
                        buf.extend(b"md5");
                        for i in md5.iter() {
                            let i = *i as usize;
                            buf.push(HEX[i >> 4]);
                            buf.push(HEX[i & 0xf]);
                        }
                        buf.push(0);
                        let len = buf.len() - 1;
                        BigEndian::write_u32(&mut buf[1..], len as u32);
                        self.write = Some(WriteState::Write { w: tokio_io::io::write_all(w, buf) })
                    }
                    t => panic!("auth {:?} not implemented", t),
                }
            },
            _ => {}
        }
    }

    fn backend_key_data(&mut self, buf: &[u8]) {
        self.process_id = Some(BigEndian::read_i32(&buf[..]));
        self.secret_key = Some(BigEndian::read_i32(&buf[4..]));
    }
}

fn parameter_status(buf: &[u8]) -> Result<(), Error> {
    if let Some(i) = (&buf[..]).iter().position(|&x| x == 0) {

        if let Some(j) = (&buf[i + 1..]).iter().position(|&x| x == 0) {
            debug!("parameter status: {:?}", std::str::from_utf8(&buf[..i]));
            debug!("parameter status: {:?}",
                   std::str::from_utf8(&buf[i + 1..i + 1 + j]));
            return Ok(());
        }
    }
    Err(Error::Protocol)
}

fn error(buf: &[u8], into: &mut String) -> Result<(), Error> {
    let mut i = 0;
    while i < buf.len() && buf[i] != 0 {
        if let Some(j) = (&buf[i + 1..]).iter().position(|&x| x == 0) {
            let s = std::str::from_utf8(&buf[i + 1..i + 1 + j]);
            error!("Error: {:?}", s);
            into.push_str(s.unwrap());
            into.push('\n');
            i += j + 2
        } else {
            return Err(Error::Protocol);
        }
    }
    Ok(())
}

// #[derive(Debug)]
// pub struct Column {
// pub table_id: i32,
// pub column_num: i16,
// pub data_type: i32,
// pub data_size: i16,
// pub type_modifier: i32,
// pub format: i16
// }
//


impl<W: AsyncWrite + AsyncRead> Future for WriteFuture<W> {
    type Item = Connection<W>;
    type Error = Error;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        loop {
            debug!("writefuture poll");
            let is_flushed = if let Some(ref mut c) = self.0 {
                match c.write.take() {
                    Some(WriteState::Write { mut w }) => {
                        match try!(w.poll()) {
                            Async::Ready((stream, buf)) => {
                                debug!("written buf = {:?}", buf);
                                c.write = Some(WriteState::Flush {
                                    flush: tokio_io::io::flush(stream),
                                    buf
                                });
                                false
                            }
                            Async::NotReady => {
                                c.write = Some(WriteState::Write { w });
                                return Ok(Async::NotReady);
                            }
                        }
                    }
                    Some(WriteState::Flush { mut flush, buf }) => {
                        match try!(flush.poll()) {
                            Async::Ready(w) => {
                                c.write = Some(WriteState::Idle { w, buf });
                                true
                            },
                            Async::NotReady => {
                                c.write = Some(WriteState::Flush { flush, buf });
                                return Ok(Async::NotReady);
                            }
                        }
                    }
                    _ => unreachable!(),
                }
            } else {
                false
            };

            if is_flushed {
                return Ok(Async::Ready(self.0.take().unwrap()));
            }
        }
    }
}

/// A future driving the connected connection. Never yields anything.
impl<W: AsyncWrite + AsyncRead> Future for Connection<W> {
    type Item = ();
    type Error = Error;
    fn poll(&mut self) -> Poll<(), 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: tokio_io::io::flush(stream),
                                buf
                            });
                            continue
                        },
                        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 })
                        },
                        Async::NotReady => {
                            self.write = Some(WriteState::Flush { flush, buf });
                            return Ok(Async::NotReady);
                        }
                    }
                }
                Some(state) => self.write = Some(state),
                None => unreachable!(),
            }

            debug!("connection poll");
            match self.read.take() {
                Some(ReadState::ReadLength(mut r)) => {
                    match r.poll()? {
                        Async::Ready((connection, buffer)) => {
                            self.read = Some(read(connection, buffer));
                        },
                        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(s) => s,
                        Async::NotReady => {
                            self.read = Some(ReadState::Read {
                                msg: msg,
                                body: body,
                            });
                            return Ok(Async::NotReady);
                        }
                    };

                    match msg {
                        MSG_ERROR => {
                            let mut into = String::new();
                            try!(error(&buf, &mut into));
                            // self.state = Some(read_length(stream, buf));
                            // Ok(Async::Ready(true))
                            return Err(Error::Postgres { message: into });
                        }
                        msg => {
                            debug!("unknown message {:?}", msg as char);
                            self.read = Some(read_length(stream, buf));
                        }
                    }
                }
                None => unreachable!(),
            }
        }
    }
}


const HEX: &'static [u8] = b"0123456789abcdef";
fn md5(salt: &[u8], user: &[u8], password: &[u8]) ->
    generic_array::GenericArray<u8, generic_array::typenum::U16> {

        use digest::{FixedOutput, Input};
        let mut md5 = md_5::Md5::default();
        md5.digest(password);
        md5.digest(user);

        let output = md5.fixed_result();

        let mut md5 = md_5::Md5::default();
        for i in output.iter() {
            let i = *i as usize;
            md5.digest(&[HEX[i >> 4], HEX[i & 0xf]])
        }
        md5.digest(salt);
        md5.fixed_result()
    }

#[test]
fn test_md5() {
    let salt = [0xc4, 0x4e, 0x74, 0x12];
    let md5 = md5(&salt[..], b"pe", b"password");
    let mut v = Vec::new();
    for i in md5.iter() {
        let i = *i as usize;
        v.push(HEX[i >> 4]);
        v.push(HEX[i & 0xf]);
    }
    assert_eq!(&v[..], b"66008dc01e22d8e920e0458c43584d9e")
}

/// Types that can be converted to postgres commands.
pub trait Request {
    /// Translate this request into a sequence of Postgres commands.
    fn request(&mut self, buf: Buffer);
}

enum SendState<R> {
    Init(R, oneshot::Sender<Result<R, Error>>),
    Sending,
    Receiving,
}

/// A future ultimately resolving to a processed request.
pub struct SendRequest<R> {
    sender: mpsc::UnboundedSender<(R, oneshot::Sender<Result<R, Error>>)>,
    receiver: oneshot::Receiver<Result<R, Error>>,
    state: Option<SendState<R>>,
}




impl<R> Future for SendRequest<R> {
    type Item = R;
    type Error = Error;
    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        loop {
            debug!("send request poll");
            match self.state.take() {
                Some(SendState::Init(r, send)) => {
                    debug!("state init");
                    match try!(self.sender.start_send((r, send))) {
                        AsyncSink::NotReady((r, send)) => {
                            self.state = Some(SendState::Init(r, send));
                            return Ok(Async::NotReady);
                        }
                        AsyncSink::Ready => self.state = Some(SendState::Sending),
                    }
                }
                Some(SendState::Sending) => {
                    debug!("state sending");
                    match try!(self.sender.poll_complete()) {
                        Async::NotReady => {
                            self.state = Some(SendState::Sending);
                            return Ok(Async::NotReady);
                        }
                        Async::Ready(()) => {
                            self.state = Some(SendState::Receiving);
                        }
                    }
                }
                Some(SendState::Receiving) => {
                    debug!("state receiving");
                    match try!(self.receiver.poll()) {
                        Async::NotReady => {
                            debug!("not ready");
                            self.state = Some(SendState::Receiving);
                            return Ok(Async::NotReady);
                        }
                        Async::Ready(r) => {
                            debug!("ready: {:?}", r.is_ok());
                            return Ok(Async::Ready(try!(r)))
                        },
                    }
                }
                None => panic!("sendstate polled after completion"),
            }
        }
    }
}


#[cfg(test)]
mod test {
    use tokio_core;
    use super::*;
    use futures;
    use env_logger;
    use uuid::Uuid;
    #[test]
    fn basic_connection() {

        use std::sync::Arc;
        use std::net::ToSocketAddrs;
        use futures::Future;

        // A `Request` is the type used for communication with the
        // client event loop. It is used both for sending input and
        // receiving output.

        struct Request {
            login: String,
            id: Option<Uuid>
        }

        // Requests must implement the `pleingres::Request` trait,
        // meaning they can be converted to commands to be sent to
        // the server.
        impl super::Request for Request {
            fn request(&mut self, mut buf: Buffer) {
                buf.bind("SELECT id FROM users WHERE login=$1", &[&self.login])
                   .execute(0);
            }
        }

        impl HandleRow for Request {
            fn row(&mut self, mut row: Row) -> bool {
                if let Some(id_) = row.next() {
                    debug!("{:?}", std::str::from_utf8(id_));
                    let id_ = std::str::from_utf8(id_)
                        .unwrap().parse().unwrap();
                    self.id = Some(id_)
                }
                true
            }
        }
        env_logger::init().unwrap();

        let p = Arc::new(Parameters {
            addr: "::1:5432".to_socket_addrs().unwrap().next().unwrap(),
            user: "pe".to_string(),
            password: "password".to_string(),
            database: Some("pijul".to_string()),
            max_wait: std::time::Duration::from_millis(1000),
            idle_timeout: Some(std::time::Duration::from_millis(20_000)),
            tcp_keepalive: Some(10000),
        });
        let mut l = tokio_core::reactor::Core::new().unwrap();

        let db:Handle<Request> =
            spawn_connection(&mut l, p.clone()).unwrap();
        println!("spawned");
        l.run(db
              .send_request(
                  Request { login: "me".to_string(), id: None }
              )
              .and_then(|dbreq| {
                  if let Some(ref id) = dbreq.id {
                      println!("id: {:?}", id)
                  }
                  futures::finished(())
              })).unwrap()
    }
}