//! Async version of connection.
//!
//! This module requires the `futures` feature to be enabled.
//!
//! Current status:
//!  * Basic client functionality is up and running, i e, you can make method calls and
//!    receive incoming messages (e g signals).
//!  * As for server side code, you can use the `tree` module with this connection, but it does not
//!    support async method handlers.
//!
//! You're probably going to need a companion crate - dbus-tokio - for this connection to make sense.
//! (Although you can also just call read_write and process_all at regular intervals, and possibly
//! set a timeout handler.)


use crate::{Error, Message};
use crate::channel::{MatchingReceiver, Channel, Sender, Token};
use crate::strings::{BusName, Path, Interface, Member};
use crate::arg::{AppendAll, ReadAll, IterAppend};
use crate::message::{MatchRule, MessageType};

use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::{task, pin, mem};
use std::cell::RefCell;
use std::time::Duration;
use crate::filters::Filters;
use std::future::Future;
use std::time::Instant;
use std::collections::HashMap;


#[allow(missing_docs)]
mod generated_org_freedesktop_standard_interfaces;
#[allow(dead_code)]
mod generated_org_freedesktop_dbus;


/// This module contains some standard interfaces and an easy way to call them.
///
/// See the [D-Bus specification](https://dbus.freedesktop.org/doc/dbus-specification.html#standard-interfaces) for more information about these standard interfaces.
///
/// The code was created by dbus-codegen.
pub mod stdintf {
    #[allow(missing_docs)]
    pub mod org_freedesktop_dbus {
        pub use super::super::generated_org_freedesktop_standard_interfaces::*;
        #[allow(unused_imports)]
        pub(crate) use super::super::generated_org_freedesktop_dbus::*;

        #[derive(Debug, PartialEq, Eq, Copy, Clone)]
        pub enum RequestNameReply {
            PrimaryOwner = 1,
            InQueue = 2,
            Exists = 3,
            AlreadyOwner = 4,
        }

        #[derive(Debug, PartialEq, Eq, Copy, Clone)]
        pub enum ReleaseNameReply {
            Released = 1,
            NonExistent = 2,
            NotOwner = 3,
        }

    }
}


type Replies<F> = HashMap<Token, F>;

/// A connection to D-Bus, thread local + async version
pub struct LocalConnection {
    channel: Channel,
    filters: RefCell<Filters<LocalFilterCb>>,
    replies: RefCell<Replies<LocalRepliesCb>>,
    timeout_maker: Option<TimeoutMakerCb>,
    waker: Option<WakerCb>,
    all_signal_matches: AtomicBool,
}

/// A connection to D-Bus, async version, which is Send but not Sync.
pub struct Connection {
    channel: Channel,
    filters: RefCell<Filters<FilterCb>>,
    replies: RefCell<Replies<RepliesCb>>,
    timeout_maker: Option<TimeoutMakerCb>,
    waker: Option<WakerCb>,
    all_signal_matches: AtomicBool,
}

/// A connection to D-Bus, Send + Sync + async version
pub struct SyncConnection {
    channel: Channel,
    filters: Mutex<Filters<SyncFilterCb>>,
    replies: Mutex<Replies<SyncRepliesCb>>,
    timeout_maker: Option<TimeoutMakerCb>,
    waker: Option<WakerCb>,
    all_signal_matches: AtomicBool,
}

use stdintf::org_freedesktop_dbus::DBus;

macro_rules! connimpl {
     ($c: ident, $cb: ident, $rcb: ident $(, $ss:tt)*) =>  {

type
    $cb = Box<dyn FnMut(Message, &$c) -> bool $(+ $ss)* + 'static>;
type
    $rcb = Box<dyn FnOnce(Message, &$c) $(+ $ss)* + 'static>;

impl From<Channel> for $c {
    fn from(x: Channel) -> Self {
        $c {
            channel: x,
            replies: Default::default(),
            filters: Default::default(),
            timeout_maker: None,
            waker: None,
            all_signal_matches: AtomicBool::new(false),
        }
    }
}

impl AsRef<Channel> for $c {
    fn as_ref(&self) -> &Channel { &self.channel }
}

impl Sender for $c {
    fn send(&self, msg: Message) -> Result<u32, ()> {
        let token = self.channel.send(msg);
        if self.channel.has_messages_to_send() {
            // Non-blocking send failed
            // Wake up task that will send the message
            if self.waker.as_ref().map(|wake| wake().is_err() ).unwrap_or(false) {
                return Err(());
            }
        }
        token
    }
}

impl MatchingReceiver for $c {
    type F = $cb;
    fn start_receive(&self, m: MatchRule<'static>, f: Self::F) -> Token {
        self.filters_mut().add(m, f)
    }
    fn stop_receive(&self, id: Token) -> Option<(MatchRule<'static>, Self::F)> {
        self.filters_mut().remove(id)
    }
}

impl NonblockReply for $c {
    type F = $rcb;
    fn send_with_reply(&self, msg: Message, f: Self::F) -> Result<Token, ()> {
        let token = {
            // We must hold the mutex from moment we send the message
            // To moment we set a handler for the reply
            // So reply can't arrive before we set handler
            let mut replies = self.replies_mut();
            self.channel.send(msg).map(|x| {
                let t = Token(x as usize);
                replies.insert(t, f);
                t
            })
        };
        if self.channel.has_messages_to_send() {
            // Non-blocking send failed
            // Wake up task that will send the message
            if self.waker.as_ref().map(|wake| wake().is_err() ).unwrap_or(false) {
                return Err(());
            }
        }
        token
    }
    fn cancel_reply(&self, id: Token) -> Option<Self::F> { self.replies_mut().remove(&id) }
    fn make_f<G: FnOnce(Message, &Self) + Send + 'static>(g: G) -> Self::F { Box::new(g) }
    fn timeout_maker(&self) -> Option<TimeoutMakerCb> { self.timeout_maker }
    fn set_timeout_maker(&mut self, f: Option<TimeoutMakerCb>) -> Option<TimeoutMakerCb> {
        mem::replace(&mut self.timeout_maker, f)
    }
    fn set_waker(&mut self, f: Option<WakerCb>) -> Option<WakerCb> {
        mem::replace(&mut self.waker, f)
    }
}


impl Process for $c {
    fn process_one(&self, msg: Message) {
        if let Some(serial) = msg.get_reply_serial() {
            if let Some(f) = self.replies_mut().remove(&Token(serial as usize)) {
                f(msg, self);
                return;
            }
        }
        if self.all_signal_matches.load(Ordering::Acquire) && msg.msg_type() == MessageType::Signal {
            // If it's a signal and the mode is enabled, send a copy of the message to all
            // matching filters.
            let matching_filters = self.filters_mut().remove_all_matching(&msg);
            // `matching_filters` needs to be a separate variable and not inlined here, because if
            // it's inline then the `MutexGuard` will live too long and we'll get a deadlock on the
            // next call to `filters_mut()` below.
            for mut ff in matching_filters {
                if let Ok(copy) = msg.duplicate() {
                    if ff.2(copy, self) {
                        self.filters_mut().insert(ff);
                    }
                } else {
                    // Silently drop the message, but add the filter back.
                    self.filters_mut().insert(ff);
                }
            }
        } else {
            // Otherwise, send the original message to only the first matching filter.
            let ff = self.filters_mut().remove_first_matching(&msg);
            if let Some(mut ff) = ff {
                if ff.2(msg, self) {
                    self.filters_mut().insert(ff);
                }
            } else if let Some(reply) = crate::channel::default_reply(&msg) {
                let _ = self.channel.send(reply);
            }
        }
    }
}

impl $c {
    fn dbus_proxy(&self) -> Proxy<'_, &Self> {
        Proxy::new("org.freedesktop.DBus", "/org/freedesktop/DBus", Duration::from_secs(10), self)
    }

    /// Get the connection's unique name.
    ///
    /// It's usually something like ":1.54"
    pub fn unique_name(&self) -> BusName<'_> { self.channel.unique_name().unwrap().into() }

    /// Request a name on the D-Bus.
    ///
    /// For detailed information on the flags and return values, see the libdbus documentation.
    pub async fn request_name<'a, N: Into<BusName<'a>>>(&self, name: N, allow_replacement: bool, replace_existing: bool, do_not_queue: bool)
    -> Result<stdintf::org_freedesktop_dbus::RequestNameReply, Error> {
        let flags: u32 =
            if allow_replacement { 1 } else { 0 } +
            if replace_existing { 2 } else { 0 } +
            if do_not_queue { 4 } else { 0 };
        let r = self.dbus_proxy().request_name(&name.into(), flags).await?;
        use stdintf::org_freedesktop_dbus::RequestNameReply::*;
        let all = [PrimaryOwner, InQueue, Exists, AlreadyOwner];
        all.iter().find(|x| **x as u32 == r).copied().ok_or_else(||
            crate::Error::new_failed("Invalid reply from DBus server")
        )
    }

    /// Release a previously requested name on the D-Bus.
    pub async fn release_name<'a, N: Into<BusName<'a>>>(&self, name: N) -> Result<stdintf::org_freedesktop_dbus::ReleaseNameReply, Error> {
        let r = self.dbus_proxy().release_name(&name.into()).await?;
        use stdintf::org_freedesktop_dbus::ReleaseNameReply::*;
        let all = [Released, NonExistent, NotOwner];
        all.iter().find(|x| **x as u32 == r).copied().ok_or_else(||
            crate::Error::new_failed("Invalid reply from DBus server")
        )
    }

    /// Adds a new match to the connection, and sets up a callback when this message arrives.
    ///
    /// If multiple [`MatchRule`]s match the same message, then by default only the first will get
    /// the callback. This behaviour can be changed for signal messages by calling
    /// [`set_signal_match_mode`](Self::set_signal_match_mode).
    ///
    /// The returned value can be used to remove the match.
    pub async fn add_match(&self, match_rule: MatchRule<'static>) -> Result<MsgMatch, Error> {
        let m = match_rule.match_str();
        self.add_match_no_cb(&m).await?;
        let mi = Arc::new(MatchInner {
            token: Default::default(),
            cb: Default::default(),
        });
        let mi_weak = Arc::downgrade(&mi);
        let token = self.start_receive(match_rule, Box::new(move |msg, _| {
            mi_weak.upgrade().map(|mi| mi.incoming(msg)).unwrap_or(false)
        }));
        mi.token.store(token.0, Ordering::SeqCst);
        Ok(MsgMatch(mi))
    }


    /// Adds a new match to the connection, without setting up a callback when this message arrives.
    pub async fn add_match_no_cb(&self, match_str: &str) -> Result<(), Error> {
        self.dbus_proxy().add_match(match_str).await
    }

    /// Removes a match from the connection, without removing any callbacks.
    pub async fn remove_match_no_cb(&self, match_str: &str) -> Result<(), Error> {
        self.dbus_proxy().remove_match(match_str).await
    }

    /// Removes a previously added match and callback from the connection.
    pub async fn remove_match(&self, id: Token) -> Result<(), Error> {
        let (mr, _) = self.stop_receive(id).ok_or_else(|| Error::new_failed("No match with that id found"))?;
        self.remove_match_no_cb(&mr.match_str()).await
    }

    /// If true, configures the connection to send signal messages to all matching [`MatchRule`]
    /// filters added with [`add_match`](Self::add_match) rather than just the first one. This comes
    /// with the following gotchas:
    ///
    ///  * The messages might be duplicated, so the message serial might be lost (this is
    ///    generally not a problem for signals).
    ///  * Panicking inside a match callback might mess with other callbacks, causing them
    ///    to be permanently dropped.
    ///  * Removing other matches from inside a match callback is not supported.
    ///
    /// This is false by default, for a newly-created connection.
    pub fn set_signal_match_mode(&self, match_all: bool) {
        self.all_signal_matches.store(match_all, Ordering::Release);
    }
}


    }
}

connimpl!(Connection, FilterCb, RepliesCb, Send);
connimpl!(LocalConnection, LocalFilterCb, LocalRepliesCb);
connimpl!(SyncConnection, SyncFilterCb, SyncRepliesCb, Send);

impl Connection {
    fn filters_mut(&self) -> std::cell::RefMut<'_, Filters<FilterCb>> { self.filters.borrow_mut() }
    fn replies_mut(&self) -> std::cell::RefMut<'_, Replies<RepliesCb>> { self.replies.borrow_mut() }
}

impl LocalConnection {
    fn filters_mut(&self) -> std::cell::RefMut<'_, Filters<LocalFilterCb>> { self.filters.borrow_mut() }
    fn replies_mut(&self) -> std::cell::RefMut<'_, Replies<LocalRepliesCb>> { self.replies.borrow_mut() }
}

impl SyncConnection {
    fn filters_mut(&self) -> std::sync::MutexGuard<'_, Filters<SyncFilterCb>> { self.filters.lock().unwrap() }
    fn replies_mut(&self) -> std::sync::MutexGuard<'_, Replies<SyncRepliesCb>> { self.replies.lock().unwrap() }
}

/// Internal callback for the executor when a timeout needs to be made.
pub type TimeoutMakerCb = fn(timeout: Instant) -> pin::Pin<Box<dyn Future<Output=()> + Send + Sync + 'static>>;

/// Internal callback for the executor when we need wakeup a task
pub type WakerCb = Box<dyn Fn() -> Result<(), ()> + Send + Sync +'static>;

/// Internal helper trait for async method replies.
pub trait NonblockReply {
    /// Callback type
    type F;
    /// Sends a message and calls the callback when a reply is received.
    fn send_with_reply(&self, msg: Message, f: Self::F) -> Result<Token, ()>;
    /// Cancels a pending reply.
    fn cancel_reply(&self, id: Token) -> Option<Self::F>;
    /// Internal helper function that creates a callback.
    fn make_f<G: FnOnce(Message, &Self) + Send + 'static>(g: G) -> Self::F where Self: Sized;
    /// Set the internal timeout maker
    fn set_timeout_maker(&mut self, f: Option<TimeoutMakerCb>) -> Option<TimeoutMakerCb>;
    /// Get the internal timeout maker
    fn timeout_maker(&self) -> Option<TimeoutMakerCb>;
    /// Set the wakeup call
    fn set_waker(&mut self, f: Option<WakerCb>) -> Option<WakerCb>;
}


/// Internal helper trait, implemented for connections that process incoming messages.
pub trait Process: Sender + AsRef<Channel> {
    /// Dispatches all pending messages, without blocking.
    ///
    /// This is usually called from the reactor only, after read_write.
    /// Despite this taking &self and not "&mut self", it is a logic error to call this
    /// recursively or from more than one thread at a time.
    fn process_all(&self) {
        let c: &Channel = self.as_ref();
        while let Some(msg) = c.pop_message() {
            self.process_one(msg);
        }
    }

    /// Dispatches a message.
    fn process_one(&self, msg: Message);
}

/// A struct used to handle incoming matches
///
/// Note: Due to the lack of async destructors, please call Connection.remove_match()
/// in order to properly stop matching (instead of just dropping this struct).
pub struct MsgMatch(Arc<MatchInner>);

struct MatchInner {
    token: AtomicUsize,
    cb: Mutex<Option<Box<dyn FnMut(Message) -> bool + Send>>>,
}

impl MatchInner {
    fn incoming(&self, msg: Message) -> bool {
        if let Some(ref mut cb) = self.cb.lock().unwrap().as_mut() {
            cb(msg)
        }
        else { true }
    }
}

impl MsgMatch {
    /// Configures the match to receive a synchronous callback with only a message parameter.
    pub fn msg_cb<F: FnMut(Message) -> bool + Send + 'static>(self, f: F) -> Self {
        {
            let mut cb = self.0.cb.lock().unwrap();
            *cb = Some(Box::new(f));
        }
        self
    }

    /// Configures the match to receive a synchronous callback with a message parameter and typed
    /// message arguments.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let mr = MatchRule::new_signal("com.example.dbustest", "HelloHappened");
    /// let incoming_signal = connection.add_match(mr).await?.cb(|_, (source,): (String,)| {
    ///    println!("Hello from {} happened on the bus!", source);
    ///    true
    /// });
    /// ```
    pub fn cb<R: ReadAll, F: FnMut(Message, R) -> bool + Send + 'static>(self, mut f: F) -> Self {
        self.msg_cb(move |msg| {
            if let Ok(r) = R::read(&mut msg.iter_init()) {
                f(msg, r)
            } else { true }
        })
    }

    /// Configures the match to receive a stream of messages.
    ///
    /// Note: If the receiving end is disconnected and a message is received,
    /// the message matching will end but not in a clean fashion. Call remove_match() to
    /// stop matching cleanly.
    pub fn msg_stream(self) -> (Self, futures_channel::mpsc::UnboundedReceiver<Message>) {
        let (sender, receiver) = futures_channel::mpsc::unbounded();
        (self.msg_cb(move |msg| {
            sender.unbounded_send(msg).is_ok()
        }), receiver)
    }

    /// Configures the match to receive a stream of messages, parsed and ready.
    ///
    /// Note: If the receiving end is disconnected and a message is received,
    /// the message matching will end but not in a clean fashion. Call remove_match() to
    /// stop matching cleanly.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let mr = MatchRule::new_signal("com.example.dbustest", "HelloHappened");
    /// let (incoming_signal, stream) = conn.add_match(mr).await?.stream();
    /// let stream = stream.for_each(|(_, (source,)): (_, (String,))| {
    ///    println!("Hello from {} happened on the bus!", source);
    ///    async {}
    /// });
    /// ```
    pub fn stream<R: ReadAll + Send + 'static>(self) -> (Self, futures_channel::mpsc::UnboundedReceiver<(Message, R)>) {
        let (sender, receiver) = futures_channel::mpsc::unbounded();
        (self.cb(move |msg, r| {
            sender.unbounded_send((msg, r)).is_ok()
        }), receiver)
    }

    /// The token retrieved can be used in a call to remove_match to stop matching on the data.
    pub fn token(&self) -> Token { Token(self.0.token.load(Ordering::SeqCst)) }
}

/// A struct that wraps a connection, destination and path.
///
/// A D-Bus "Proxy" is a client-side object that corresponds to a remote object on the server side.
/// Calling methods on the proxy object calls methods on the remote object.
/// Read more in the [D-Bus tutorial](https://dbus.freedesktop.org/doc/dbus-tutorial.html#proxies)
#[derive(Clone, Debug)]
pub struct Proxy<'a, C> {
    /// Destination, i e what D-Bus service you're communicating with
    pub destination: BusName<'a>,
    /// Object path on the destination
    pub path: Path<'a>,
    /// Some way to send and/or receive messages, non-blocking.
    pub connection: C,
    /// Timeout for method calls
    pub timeout: Duration,
}

impl<'a, C> Proxy<'a, C> {
    /// Creates a new proxy struct.
    pub fn new<D: Into<BusName<'a>>, P: Into<Path<'a>>>(dest: D, path: P, timeout: Duration, connection: C) -> Self {
        Proxy { destination: dest.into(), path: path.into(), timeout, connection }
    }
}

struct MRAwait {
    mrouter: MROuter,
    token: Result<Token, ()>,
    timeout: Instant,
    timeoutfn: Option<TimeoutMakerCb>
}

async fn method_call_await(mra: MRAwait) -> Result<Message, Error> {
    use futures_util::future;
    let MRAwait { mrouter, token, timeout, timeoutfn } = mra;
    if token.is_err() { return Err(Error::new_failed("Failed to send message")) };
    let timeout = if let Some(tfn) = timeoutfn { tfn(timeout) } else { Box::pin(future::pending()) };
    match future::select(mrouter, timeout).await {
        future::Either::Left((r, _)) => r,
        future::Either::Right(_) => Err(Error::new_custom("org.freedesktop.DBus.Error.Timeout", "Timeout waiting for reply")),
    }
}

impl<'a, T, C> Proxy<'a, C>
where
    T: NonblockReply,
    C: std::ops::Deref<Target=T>
{

    fn method_call_setup(&self, msg: Message) -> MRAwait {
        let mr = Arc::new(Mutex::new(MRInner::Neither));
        let mrouter = MROuter(mr.clone());
        let f = T::make_f(move |msg: Message, _: &T| {
            let mut inner = mr.lock().unwrap();
            let old = mem::replace(&mut *inner, MRInner::Ready(Ok(msg)));
            if let MRInner::Pending(waker) = old { waker.wake() }
        });

        let timeout = Instant::now() + self.timeout;
        let token = self.connection.send_with_reply(msg, f);
        let timeoutfn = self.connection.timeout_maker();
        MRAwait { mrouter, token, timeout, timeoutfn }
    }

    /// Make a method call using typed input argument, returns a future that resolves to the typed output arguments.
    pub fn method_call<'i, 'm, R: ReadAll + 'static, A: AppendAll, I: Into<Interface<'i>>, M: Into<Member<'m>>>(&self, i: I, m: M, args: A)
    -> MethodReply<R> {
        let mut msg = Message::method_call(&self.destination, &self.path, &i.into(), &m.into());
        args.append(&mut IterAppend::new(&mut msg));
        let mra = self.method_call_setup(msg);
        let r = method_call_await(mra);
        let r = futures_util::FutureExt::map(r, |r| -> Result<R, Error> { r.and_then(|rmsg| rmsg.read_all()) } );
        MethodReply::new(r)
    }
}

enum MRInner {
    Ready(Result<Message, Error>),
    Pending(task::Waker),
    Neither,
}

struct MROuter(Arc<Mutex<MRInner>>);

impl Future for MROuter {
    type Output = Result<Message, Error>;
    fn poll(self: pin::Pin<&mut Self>, ctx: &mut task::Context) -> task::Poll<Self::Output> {
        let mut inner = self.0.lock().unwrap();
        let r = mem::replace(&mut *inner, MRInner::Neither);
        if let MRInner::Ready(r) = r { task::Poll::Ready(r) }
        else {
            *inner = MRInner::Pending(ctx.waker().clone());
            return task::Poll::Pending
        }
    }
}

/// Future method reply, used while waiting for a method call reply from the server.
pub struct MethodReply<T>(pin::Pin<Box<dyn Future<Output=Result<T, Error>> + Send + 'static>>);

impl<T> MethodReply<T> {
    /// Creates a new method reply from a future.
    fn new<Fut: Future<Output=Result<T, Error>> + Send + 'static>(fut: Fut) -> Self {
        MethodReply(Box::pin(fut))
    }
}

impl<T> Future for MethodReply<T> {
    type Output = Result<T, Error>;
    fn poll(mut self: pin::Pin<&mut Self>, ctx: &mut task::Context) -> task::Poll<Result<T, Error>> {
        self.0.as_mut().poll(ctx)
    }
}

impl<T: 'static> MethodReply<T> {
    /// Convenience combinator in case you want to post-process the result after reading it
    pub fn and_then<T2>(self, f: impl FnOnce(T) -> Result<T2, Error> + Send + Sync + 'static) -> MethodReply<T2> {
        MethodReply(Box::pin(async move {
            let x = self.0.await?;
            f(x)
        }))
    }
}

#[test]
fn test_conn_send_sync() {
    fn is_send<T: Send>() {}
    fn is_sync<T: Sync>() {}
    is_send::<Connection>();
    is_send::<SyncConnection>();
    is_sync::<SyncConnection>();
    is_send::<MsgMatch>();
}