use crate::error::{SqlError, SqlResult};
use crate::handle::SqlHandle;
use crate::stmt::*;
use futures::{task, Async, Future, Map, Poll};
use futures_state_stream::*;

pub(crate) trait AsyncExt<T> {
    fn unwrap_async(self) -> SqlResult<T>;
}

impl<T> AsyncExt<T> for SqlPoll<T> {
    fn unwrap_async(self) -> SqlResult<T> {
        match self? {
            Async::Ready(x) => Ok(x),
            Async::NotReady => Err(SqlError::StillExecuting),
        }
    }
}

pub type SqlPoll<T = ()> = Poll<T, SqlError>;
pub type SqlHandlePoll<T, V> = Poll<(T, V), (SqlError, T)>;

pub struct SqlValueFuture<T, V>
where
    T: SqlHandle,
{
    handle: Option<T>,
    poll_fn: Box<dyn FnMut(&mut T) -> SqlPoll<V> + Send>,
    not_ready: bool,
}

pub type SqlFuture<T> = Map<SqlValueFuture<T, ()>, fn((T, ())) -> T>;

pub(crate) fn flatten_value_future<T: SqlHandle>(f: SqlValueFuture<T, ()>) -> SqlFuture<T> {
    f.map(|(a, _b)| a)
}

impl<T: SqlHandle, V> SqlValueFuture<T, V> {
    pub(crate) fn new<F>(handle: T, poll_fn: F) -> SqlValueFuture<T, V>
    where
        F: FnMut(&mut T) -> SqlPoll<V> + Send + 'static,
    {
        SqlValueFuture {
            handle: Some(handle),
            poll_fn: Box::new(poll_fn),
            not_ready: false,
        }
    }
}

impl<T: SqlHandle, V> Future for SqlValueFuture<T, V> {
    type Item = (T, V);
    type Error = (SqlError, T);

    fn poll(&mut self) -> SqlHandlePoll<T, V> {
        let poll = {
            let handle = self.handle.as_mut().unwrap();
            (self.poll_fn)(handle)
        };

        match poll {
            Ok(x) => match x {
                Async::Ready(value) => {
                    let handle = self.handle.take().unwrap();
                    let value: (T, V) = (handle, value);

                    Ok(Async::Ready(value))
                }
                Async::NotReady => {
                    self.not_ready = true;
                    task::current().notify();
                    Ok(Async::NotReady)
                }
            },
            Err(e) => {
                let handle = self.handle.take().unwrap();
                Err((e, handle))
            }
        }
    }
}

#[cfg(feature = "cancellation")]
impl<T: SqlHandle, V> Drop for SqlValueFuture<T, V> {
    fn drop(&mut self) {
        use odbc_sys::*;

        if !self.not_ready {
            return;
        }

        if let Some(mut handle) = self.handle.take() {
            let ret = unsafe { SQLCancelHandle(T::TYPE, handle.handle()) };
            match ret {
                SQL_SUCCESS | SQL_SUCCESS_WITH_INFO => loop {
                    match (self.poll_fn)(&mut handle) {
                        Ok(x) => {
                            if x.is_ready() {
                                break;
                            }
                        }
                        Err(e) => {
                            if e.is_async_canceled_error() {
                                break;
                            } else {
                                panic!("{}", e)
                            }
                        }
                    }
                },
                SQL_ERROR => {
                    let e = handle.get_detailed_error(ret);
                    panic!("{}", e)
                }
                _ => panic!("Unexpected SQLCancelHandle return code {:?}", ret),
            }
        }
    }
}

pub struct SqlStatementStream<T> {
    statement: Option<SqlStatement>,
    poll_fn: Box<dyn FnMut(&mut SqlStatement) -> SqlPoll<Option<T>> + Send>,
    unbind: bool,
    not_ready: bool,
}

impl<T> SqlStatementStream<T> {
    pub fn new<F>(statement: SqlStatement, unbind: bool, poll_fn: F) -> SqlStatementStream<T>
    where
        F: FnMut(&mut SqlStatement) -> SqlPoll<Option<T>> + Send + 'static,
    {
        SqlStatementStream {
            statement: Some(statement),
            poll_fn: Box::new(poll_fn),
            unbind,
            not_ready: false,
        }
    }

    fn take_statement(&mut self) -> SqlStatement {
        let mut statement = self.statement.take().unwrap();
        if self.unbind {
            statement.unbind().unwrap();
        }
        statement
    }
}

impl<T> StateStream for SqlStatementStream<T> {
    type Item = T;
    type State = SqlStatement;
    type Error = SqlError;

    fn poll(&mut self) -> Poll<StreamEvent<T, SqlStatement>, (SqlError, SqlStatement)> {
        let poll = {
            let statement = self.statement.as_mut().unwrap();
            (self.poll_fn)(statement)
        };

        match poll {
            Ok(x) => match x {
                Async::Ready(value) => {
                    if let Some(item) = value {
                        let event = StreamEvent::Next(item);
                        Ok(Async::Ready(event))
                    } else {
                        let state = self.take_statement();
                        let event = StreamEvent::Done(state);
                        Ok(Async::Ready(event))
                    }
                }
                Async::NotReady => {
                    self.not_ready = true;
                    task::current().notify();
                    Ok(Async::NotReady)
                }
            },
            Err(e) => {
                let statement = self.take_statement();
                Err((e, statement))
            }
        }
    }
}

#[cfg(feature = "cancellation")]
impl<T> Drop for SqlStatementStream<T> {
    fn drop(&mut self) {
        use odbc_sys::*;

        if !self.not_ready {
            return;
        }

        if let Some(mut statement) = self.statement.take() {
            let ret = unsafe { SQLCancelHandle(SqlStatement::TYPE, statement.handle()) };
            match ret {
                SQL_SUCCESS | SQL_SUCCESS_WITH_INFO => loop {
                    match (self.poll_fn)(&mut statement) {
                        Ok(x) => {
                            if x.is_ready() {
                                break;
                            }
                        }
                        Err(e) => {
                            if e.is_async_canceled_error() {
                                break;
                            } else {
                                panic!("{}", e)
                            }
                        }
                    }
                },
                SQL_ERROR => {
                    let e = statement.get_detailed_error(ret);
                    panic!("{}", e)
                }
                _ => panic!("Unexpected SQLCancelHandle return code {:?}", ret),
            }
        }
    }
}