use tokio_xmpp::{Client, Event, Packet};
use tokio::prelude::future::{self, Either};
use tokio::prelude::stream;
use tokio::prelude::{Future, Stream};
use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant};
use super::stanzas;
use super::element_processor;
use crate::config;
#[derive(Debug)]
pub enum XmppCommand {
    /// Send message to someone by jid
    Chat {
        xmpp_to: xmpp_parsers::Jid,
        message: String,
    },
    /// Send message to MUC
    Chatroom { muc_id: String, message: String },
    /// Send ping request to the server to test connection
    Ping,
    /// Check iq requests if some have expired timeouts
    TimeoutCleanup,
}
/// trait of processing iq
/// each function consumes handlers and
/// returns false if connection should be reset
trait IqHandler {
    /// process result
    fn result(
        self: Box<Self>,
        conn: &mut XmppConnection,
        opt_element: Option<xmpp_parsers::Element>,
    ) -> bool;
    /// process error
    fn error(
        self: Box<Self>,
        conn: &mut XmppConnection,
        error: xmpp_parsers::stanza_error::StanzaError,
    ) -> bool;
    /// process tmeout
    fn timeout(self: Box<Self>, conn: &mut XmppConnection) -> bool;
}
struct AddRosterIqHandler {
    jid: xmpp_parsers::Jid,
}
impl IqHandler for AddRosterIqHandler {
    fn result(
        self: Box<Self>,
        conn: &mut XmppConnection,
        opt_element: Option<xmpp_parsers::Element>,
    ) -> bool {
        match opt_element {
            Some(element) => {
                warn!(
                    "Wrong payload when adding {} to roster: {}",
                    self.jid,
                    String::from(&element)
                );
            }
            None => {
                if conn.state.data.roster.contains_key(&self.jid) {
                    info!("Jid {} updated to roster", self.jid);
                } else {
                    info!("Jid {} added in roster", self.jid);
                    conn.state.data.roster.insert(
                        self.jid.clone(),
                        (
                            xmpp_parsers::roster::Subscription::None,
                            xmpp_parsers::roster::Ask::None,
                        ),
                    );
                }
                conn.process_jid(&self.jid);
            }
        }
        true
    }
    fn error(
        self: Box<Self>,
        _conn: &mut XmppConnection,
        _error: xmpp_parsers::stanza_error::StanzaError,
    ) -> bool {
        true
    }
    fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
        true // ignore
    }
}
struct PingIqHandler {}
impl IqHandler for PingIqHandler {
    fn result(
        self: Box<Self>,
        _conn: &mut XmppConnection,
        _opt_element: Option<xmpp_parsers::Element>,
    ) -> bool {
        info!("ping successed");
        true
    }
    fn error(
        self: Box<Self>,
        _conn: &mut XmppConnection,
        _error: xmpp_parsers::stanza_error::StanzaError,
    ) -> bool {
        false
    }
    fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
        false
    }
}
struct InitRosterIqHandler {}
impl IqHandler for InitRosterIqHandler {
    fn result(
        self: Box<Self>,
        conn: &mut XmppConnection,
        opt_element: Option<xmpp_parsers::Element>,
    ) -> bool {
        if let Some(result) = opt_element {
            use std::convert::TryInto;
            match result.try_into() as Result<xmpp_parsers::roster::Roster, _> {
                Ok(roster) => {
                    conn.state.data.roster_init = true;
                    conn.state.data.roster.clear();
                    info!("Got first roster:");
                    for i in roster.items {
                        info!(" >>> {:?}", i);
                        conn.state
                            .data
                            .roster
                            .insert(i.jid, (i.subscription, i.ask));
                    }
                    true
                }
                Err(e) => {
                    error!("Cann't parse roster: {}", e);
                    false
                }
            }
        } else {
            error!("No roster responded");
            false
        }
    }
    fn error(
        self: Box<Self>,
        _conn: &mut XmppConnection,
        _error: xmpp_parsers::stanza_error::StanzaError,
    ) -> bool {
        false
    }
    fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
        false
    }
}
#[derive(Default)]
struct XmppData {
    /// known roster data
    roster: HashMap<
        xmpp_parsers::Jid,
        (
            xmpp_parsers::roster::Subscription,
            xmpp_parsers::roster::Ask,
        ),
    >,
    /// if roster was initialized
    /// ToDo: remove it as it is used only for initialization
    roster_init: bool,
    /// if self-presence accepted
    /// ToDo: remove it as it is used only for initialization
    self_presence: bool,
    /// ids counter
    counter: usize,
    /// stanzas to send
    send_queue: VecDeque<minidom::Element>,
    /// outgoing mailbox
    outgoing_mailbox: HashMap<xmpp_parsers::Jid, Vec<String>>,
    /// muc id to muc jid
    mucs: HashMap<String, xmpp_parsers::Jid>,
    /// map from iq's id to handler of this type of iqs
    pending_ids: HashMap<String, (Instant, Box<dyn IqHandler>)>,
}
struct XmppState {
    client: Client,
    data: XmppData,
}
pub struct XmppConnection {
    account: std::rc::Rc<config::Account>,
    state: XmppState,
}
lazy_static! {
    static ref INCOMING: element_processor::Processor<XmppConnection, bool, xmpp_parsers::Element> = {
        let mut incoming = element_processor::Processor::new(&|_, e| {
            warn!("Unknown stanza {}", String::from(&e));
            true
        });
        incoming.register(&XmppConnection::incoming_iq_processing);
        incoming.register(&XmppConnection::incoming_presence_processing);
        incoming.register(&XmppConnection::incoming_message_processing);
        incoming
    };
}
pub struct MaybeXmppConnection {
    account: std::rc::Rc<config::Account>,
    state: Option<XmppState>,
}
impl From<XmppConnection> for MaybeXmppConnection {
    fn from(from: XmppConnection) -> MaybeXmppConnection {
        MaybeXmppConnection {
            account: from.account,
            state: Some(from.state),
        }
    }
}
impl From<config::Account> for MaybeXmppConnection {
    fn from(from: config::Account) -> MaybeXmppConnection {
        MaybeXmppConnection {
            account: std::rc::Rc::new(from),
            state: None,
        }
    }
}
impl From<std::rc::Rc<config::Account>> for MaybeXmppConnection {
    fn from(from: std::rc::Rc<config::Account>) -> MaybeXmppConnection {
        MaybeXmppConnection {
            account: from,
            state: None,
        }
    }
}
impl MaybeXmppConnection {
    /// connects if nothing connected
    /// don't connect only if stop_future resolved
    pub fn connect<F>(
        self,
        stop_future: F,
    ) -> impl Future<Item = XmppConnection, Error = failure::Error>
    where
        F: future::Future + Clone + 'static,
        <F as hyper::rt::Future>::Error: Into<failure::Error> + Send,
    {
        info!("xmpp connection...");
        let MaybeXmppConnection { account, state } = self;
        if let Some(state) = state {
            Box::new(future::ok(XmppConnection { account, state }))
                as Box<dyn Future<Item = _, Error = _>>
        } else {
            Box::new(
                stop_future
                    .clone()
                    .select2(
                        future::loop_fn(account, move |account| {
                            info!("xmpp initialization...");
                            let client =
                                Client::new_with_jid(account.jid.clone(), &account.password);
                            info!("xmpp initialized");
                            let stop_future2 = stop_future.clone();
                            let stop_future3 = stop_future.clone();
                            let stop_future4 = stop_future.clone();
                            // future to wait for online
                            Box::new(
                                XmppConnection {
                                    state: XmppState {
                                        client,
                                        data: std::default::Default::default(),
                                    },
                                    account,
                                }
                                .processing(XmppConnection::online, stop_future.clone())
                                .map_err(|(acc, _)| acc)
                                .and_then(|(conn, r)| match r {
                                    Ok(Either::A(_)) => future::ok(conn),
                                    Ok(Either::B(_)) => future::err(conn.account),
                                    Err(_e) => future::err(conn.account),
                                })
                                .and_then(|conn| conn.initial_roster(stop_future2))
                                .and_then(|conn| conn.self_presence(stop_future3))
                                .and_then(|conn| conn.enter_mucs(stop_future4))
                                .then(|r| match r {
                                    Ok(conn) => future::ok(future::Loop::Break(conn)),
                                    Err(acc) => future::ok(future::Loop::Continue(acc)),
                                }),
                            )
                        })
                        .map_err(|_: ()| ()),
                    )
                    .then(|r| match r {
                        Ok(Either::A((_x, _b))) => future::err(format_err!("Stop XMMP connection")),
                        Ok(Either::B((x, _a))) => future::ok(x),
                        Err(Either::A((e, _b))) => future::err(e.into()),
                        Err(Either::B((_, _a))) => {
                            future::err(format_err!("Cann't initiate XMPP connection"))
                        }
                    }),
            )
        }
    }
}
impl XmppConnection {
    /// base XMPP processing
    /// Returns false on error to disconnect
    fn xmpp_processing(&mut self, event: &Event) -> bool {
        match event {
            Event::Stanza(stanza) => INCOMING.process(self, stanza.clone()),
            Event::Online => true,
            e => {
                warn!("Unexpected event {:?}", e);
                false
            }
        }
    }
    /// Enforce to answer to IQ "set"
    fn incoming_iq_processing_set(
        &mut self,
        id: String,
        from: Option<xmpp_parsers::Jid>,
        element: minidom::Element,
    ) -> xmpp_parsers::iq::Iq {
        use std::convert::TryInto;
        if let Some(roster) =
            element.clone().try_into().ok() as Option<xmpp_parsers::roster::Roster>
        {
            // RFC 6212 2.1.6. Roster Push
            info!("Got roster push {} from {:?}", id, from);
            for i in roster.items {
                if let Some(ref mut rdata) = self.state.data.roster.get_mut(&i.jid) {
                    info!("Update {} in roster", i.jid);
                    rdata.0 = i.subscription;
                    rdata.1 = i.ask;
                } else {
                    info!("Add {} to roster", i.jid);
                    self.state
                        .data
                        .roster
                        .insert(i.jid.clone(), (i.subscription, i.ask));
                }
                self.process_jid(&i.jid);
            }
            return stanzas::make_roster_push_answer(id, self.state.client.jid.clone(), from);
        }
        warn!(
            "Unsupported IQ set request from {:?}: {}",
            from,
            String::from(&element)
        );
        stanzas::make_iq_unsupported_error(id, self.state.client.jid.clone(), from)
    }
    /// Enforce to answer to IQ "get"
    fn incoming_iq_processing_get(
        &mut self,
        id: String,
        from: Option<xmpp_parsers::Jid>,
        element: minidom::Element,
    ) -> xmpp_parsers::iq::Iq {
        use std::convert::TryInto;
        if let Some(_ping) = element.clone().try_into().ok() as Option<xmpp_parsers::ping::Ping> {
            info!("Got ping {} from {:?}", id, from);
            return stanzas::make_pong(&id, self.state.client.jid.clone(), from);
        }
        warn!(
            "Unsupported IQ get request from {:?}: {}",
            from,
            String::from(&element)
        );
        stanzas::make_iq_unsupported_error(id, self.state.client.jid.clone(), from)
    }
    fn incoming_iq_processing(&mut self, iq: xmpp_parsers::iq::Iq) -> bool {
        match iq.payload {
            xmpp_parsers::iq::IqType::Set(element) => {
                let iq_answer = self.incoming_iq_processing_set(iq.id, iq.from, element);
                self.state.data.send_queue.push_back(iq_answer.into());
            }
            xmpp_parsers::iq::IqType::Error(e) => {
                if let Some((_, handler)) = self.state.data.pending_ids.remove_entry(&iq.id) {
                    return handler.1.error(self, e);
                }
                error!("iq error: {:?}", e);
                return false;
            }
            xmpp_parsers::iq::IqType::Get(element) => {
                let iq_answer = self.incoming_iq_processing_get(iq.id, iq.from, element);
                self.state.data.send_queue.push_back(iq_answer.into());
            }
            xmpp_parsers::iq::IqType::Result(opt_element) => {
                if let Some((_, handler)) = self.state.data.pending_ids.remove_entry(&iq.id) {
                    return handler.1.result(self, opt_element);
                }
                warn!(
                    "Unwanted iq result id {} from {:?}: {:?}",
                    iq.id,
                    iq.from,
                    opt_element.map(|e| String::from(&e))
                );
            }
        }
        true
    }
    fn incoming_presence_processing(&mut self, presence: xmpp_parsers::presence::Presence) -> bool {
        if presence.from.as_ref() == Some(&self.state.client.jid) {
            info!("Self-presence accepted");
            self.state.data.self_presence = true;
        } else {
            warn!("Incoming presence stanza: {:?}", presence);
        }
        true
    }
    fn incoming_message_processing(&mut self, message: xmpp_parsers::message::Message) -> bool {
        for payload in message.payloads.iter() {
            use std::convert::TryInto;
            if let Some(_delay) =
                payload.clone().try_into().ok() as Option<xmpp_parsers::delay::Delay>
            {
                return true; // ignore delayed messages
            }
        }
        warn!("Incoming message stanza: {:?}", message);
        true
    }
    /// process event from xmpp stream
    /// returns from future when condition met
    /// or stop future was resolved.
    /// Return item if connection was preserved or error otherwise.
    /// Second part is a state of stop_future
    pub fn processing<S, F, T, E>(
        self,
        stop_condition: S,
        stop_future: F,
    ) -> impl Future<
        Item = (Self, Result<Either<F, T>, E>),
        Error = (std::rc::Rc<config::Account>, Result<Either<F, T>, E>),
    >
    where
        F: Future<Item = T, Error = E> + 'static,
        S: FnMut(&mut Self, Event) -> Result<bool, ()> + 'static,
        T: 'static,
        E: 'static,
    {
        future::loop_fn(
            (self, stop_future, stop_condition),
            |(xmpp, stop_future, mut stop_condition)| {
                // ToDo: check timeouts if iqs
                let XmppConnection {
                    state: XmppState { client, mut data },
                    account,
                } = xmpp;
                if let Some(send_element) = data.send_queue.pop_front() {
                    use tokio::prelude::Sink;
                    info!("Sending {}", String::from(&send_element));
                    Box::new(
                        client
                            .send(Packet::Stanza(send_element))
                            .select2(stop_future)
                            .then(move |r| match r {
                                Ok(Either::A((client, b))) => {
                                    Box::new(future::ok(future::Loop::Continue((
                                        XmppConnection {
                                            state: XmppState { client, data },
                                            account,
                                        },
                                        b,
                                        stop_condition,
                                    ))))
                                        as Box<dyn Future<Item = _, Error = _>>
                                }
                                Ok(Either::B((t, a))) => Box::new(a.then(|r| match r {
                                    Ok(client) => future::ok(future::Loop::Break((
                                        XmppConnection {
                                            state: XmppState { client, data },
                                            account,
                                        },
                                        Ok(Either::B(t)),
                                    ))),
                                    Err(se) => {
                                        warn!("XMPP sending error: {}", se);
                                        future::err((account, Ok(Either::B(t))))
                                    }
                                })),
                                Err(Either::A((e, b))) => {
                                    warn!("XMPP sending error: {}", e);
                                    Box::new(future::err((account, Ok(Either::A(b)))))
                                }
                                Err(Either::B((e, a))) => Box::new(a.then(|r| match r {
                                    Ok(client) => future::ok(future::Loop::Break((
                                        XmppConnection {
                                            state: XmppState { client, data },
                                            account,
                                        },
                                        Err(e),
                                    ))),
                                    Err(se) => {
                                        warn!("XMPP sending error: {}", se);
                                        future::err((account, Err(e)))
                                    }
                                })),
                            }),
                    ) as Box<dyn Future<Item = _, Error = _>>
                } else {
                    Box::new(
                        client
                            .into_future()
                            .select2(stop_future)
                            .then(move |r| match r {
                                Ok(Either::A(((event, client), b))) => {
                                    if let Some(event) = event {
                                        let mut xmpp = XmppConnection {
                                            state: XmppState { client, data },
                                            account,
                                        };
                                        if xmpp.xmpp_processing(&event) {
                                            match stop_condition(&mut xmpp, event) {
                                                Ok(true) => future::ok(future::Loop::Break((
                                                    xmpp,
                                                    Ok(Either::A(b)),
                                                ))),
                                                Ok(false) => future::ok(future::Loop::Continue((
                                                    xmpp,
                                                    b,
                                                    stop_condition,
                                                ))),
                                                Err(_e) => {
                                                    future::err((xmpp.account, Ok(Either::A(b))))
                                                }
                                            }
                                        } else {
                                            future::err((xmpp.account, Ok(Either::A(b))))
                                        }
                                    } else {
                                        future::err((account, Ok(Either::A(b))))
                                    }
                                }
                                Ok(Either::B((t, a))) => {
                                    if let Some(client) = a.into_inner() {
                                        future::ok(future::Loop::Break((
                                            XmppConnection {
                                                state: XmppState { client, data },
                                                account,
                                            },
                                            Ok(Either::B(t)),
                                        )))
                                    } else {
                                        future::err((account, Ok(Either::B(t))))
                                    }
                                }
                                Err(Either::A((e, b))) => {
                                    warn!("XMPP error: {}", e.0);
                                    future::err((account, Ok(Either::A(b))))
                                }
                                Err(Either::B((e, a))) => {
                                    if let Some(client) = a.into_inner() {
                                        future::ok(future::Loop::Break((
                                            XmppConnection {
                                                state: XmppState { client, data },
                                                account,
                                            },
                                            Err(e),
                                        )))
                                    } else {
                                        future::err((account, Err(e)))
                                    }
                                }
                            }),
                    )
                }
            },
        )
    }
    /// get connection and wait for online status and set presence
    /// returns error if something went wrong and xmpp connection is broken
    fn online(&mut self, event: Event) -> Result<bool, ()> {
        match event {
            Event::Online => {
                info!("Online!");
                Ok(true)
            }
            Event::Stanza(s) => {
                warn!("Stanza before online: {}", String::from(&s));
                Ok(false)
            }
            _ => {
                error!("Disconnected while online");
                Err(())
            }
        }
    }
    fn initial_roster<F, E>(
        self,
        stop_future: F,
    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
    where
        F: Future<Error = E> + 'static,
        E: 'static,
    {
        let XmppConnection {
            account,
            state: XmppState { client, mut data },
        } = self;
        use tokio::prelude::Sink;
        data.counter += 1;
        let id_init_roster = format!("id_init_roster{}", data.counter);
        let get_roster = stanzas::make_get_roster(&id_init_roster);
        let account2 = account.clone();
        info!("Quering roster... {}", String::from(&get_roster));
        data.pending_ids.insert(
            id_init_roster.clone(),
            (
                Instant::now() + Duration::from_secs(60),
                Box::new(InitRosterIqHandler {}),
            ),
        );
        client
            .send(Packet::Stanza(get_roster))
            .map_err(move |e| {
                error!("Error on querying roster: {}", e);
                account2
            })
            .and_then(move |client| {
                XmppConnection {
                    state: XmppState { client, data },
                    account,
                }
                .processing(move |conn, _| Ok(conn.state.data.roster_init), stop_future)
                .map_err(|(account, _)| account)
                .and_then(|(conn, r)| match r {
                    Ok(Either::A(_)) => future::ok(conn),
                    Ok(Either::B(_)) => future::err(conn.account),
                    Err(_e) => future::err(conn.account),
                })
            })
    }
    fn self_presence<F, E>(
        self,
        stop_future: F,
    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
    where
        F: Future<Error = E> + 'static,
        E: Into<failure::Error> + 'static,
    {
        let XmppConnection {
            account,
            state: XmppState { client, data },
        } = self;
        use tokio::prelude::Sink;
        let presence = stanzas::make_presence(&account);
        let account2 = account.clone();
        info!("Sending presence... {}", String::from(&presence));
        client
            .send(Packet::Stanza(presence))
            .map_err(|e| {
                error!("Error on send self-presence: {}", e);
                account2
            })
            .and_then(move |client| {
                XmppConnection {
                    state: XmppState { client, data },
                    account,
                }
                .processing(
                    move |conn, _| Ok(conn.state.data.self_presence),
                    stop_future,
                )
                .map_err(|(account, _)| account)
                .and_then(|(conn, r)| match r {
                    Ok(Either::A(_)) => future::ok(conn),
                    Ok(Either::B(_)) => future::err(conn.account),
                    Err(_e) => future::err(conn.account),
                })
            })
    }
    fn process_jid(&mut self, xmpp_to: &xmpp_parsers::Jid) {
        if let Some(ref mut mailbox) = self.state.data.outgoing_mailbox.get_mut(xmpp_to) {
            if !mailbox.is_empty() {
                if let Some(ref mut rdata) = self.state.data.roster.get_mut(xmpp_to) {
                    info!("Jid {} in roster", xmpp_to);
                    let sub_to = match rdata.0 {
                        xmpp_parsers::roster::Subscription::To => true,
                        xmpp_parsers::roster::Subscription::Both => true,
                        _ => false,
                    };
                    if sub_to {
                        info!("Subscribed to {}", xmpp_to);
                        self.state.data.send_queue.extend(
                            mailbox.drain(..).map(|message| {
                                stanzas::make_chat_message(xmpp_to.clone(), message)
                            }),
                        );
                    } else if rdata.1 == xmpp_parsers::roster::Ask::None {
                        info!("Not subscribed to {}", xmpp_to);
                        self.state
                            .data
                            .send_queue
                            .push_back(stanzas::make_ask_subscribe(xmpp_to.clone()));
                    }
                    let sub_from = match rdata.0 {
                        xmpp_parsers::roster::Subscription::From => true,
                        xmpp_parsers::roster::Subscription::Both => true,
                        _ => false,
                    };
                    if !sub_from {
                        info!("Not subscription from {}", xmpp_to);
                        self.state
                            .data
                            .send_queue
                            .push_back(stanzas::make_allow_subscribe(xmpp_to.clone()));
                    }
                } else {
                    info!("Jid {} not in roster", xmpp_to);
                    self.state.data.counter += 1;
                    let id_add_roster = format!("id_add_roster{}", self.state.data.counter);
                    let add_roster = stanzas::make_add_roster(&id_add_roster, xmpp_to.clone());
                    info!("Adding jid {} to roster id {}", xmpp_to, id_add_roster);
                    self.state.data.pending_ids.insert(
                        id_add_roster,
                        (
                            Instant::now() + Duration::from_secs(60),
                            Box::new(AddRosterIqHandler {
                                jid: xmpp_to.clone(),
                            }),
                        ),
                    );
                    self.state.data.send_queue.push_back(add_roster);
                }
            }
        }
    }
    pub fn process_command(&mut self, cmd: XmppCommand) {
        info!("Got command");
        match cmd {
            XmppCommand::Chat { xmpp_to, message } => {
                self.state
                    .data
                    .outgoing_mailbox
                    .entry(xmpp_to.clone())
                    .or_default()
                    .push(message);
                self.process_jid(&xmpp_to);
            }
            XmppCommand::Chatroom { muc_id, message } => {
                if let Some(muc) = self.state.data.mucs.get(&muc_id) {
                    self.state
                        .data
                        .send_queue
                        .push_back(stanzas::make_muc_message(muc.clone(), message));
                } else {
                    error!("Not found MUC {}", muc_id);
                }
            }
            XmppCommand::Ping => {
                self.state.data.counter += 1;
                let id_ping = format!("id_ping{}", self.state.data.counter);
                let ping = stanzas::make_ping(&id_ping, self.state.client.jid.clone());
                self.state.data.send_queue.push_back(ping);
                self.state.data.pending_ids.insert(
                    id_ping,
                    (
                        Instant::now() + Duration::from_secs(30),
                        Box::new(PingIqHandler {}),
                    ),
                );
            }
            XmppCommand::TimeoutCleanup => {
                let now = Instant::now();
                let timeouted: Vec<String> = self
                    .state
                    .data
                    .pending_ids
                    .iter()
                    .filter_map(|(id, (timeout, _))| {
                        if now >= *timeout {
                            Some(id.to_string())
                        } else {
                            None
                        }
                    })
                    .collect();
                let mut correct = true;
                timeouted.into_iter().for_each(|id| {
                    if let Some((_, handler)) = self.state.data.pending_ids.remove(&id) {
                        correct &= handler.timeout(&mut self);
                    }
                })
            }
        }
    }
    pub fn shutdown(self) -> impl Future<Item = (), Error = failure::Error> {
        info!("Shutdown connection");
        let XmppConnection { account, state } = self;
        stream::iter_ok(
            state
                .data
                .mucs
                .values()
                .map(std::clone::Clone::clone)
                .collect::<Vec<_>>(),
        )
        .fold(state, move |XmppState { client, data }, muc_jid| {
            let muc_presence =
                stanzas::make_muc_presence_leave(account.jid.clone(), muc_jid.clone());
            info!(
                "Sending muc leave presence... {}",
                String::from(&muc_presence)
            );
            use tokio::prelude::Sink;
            client
                .send(Packet::Stanza(muc_presence))
                .map_err(|e| {
                    error!("Error on send muc presence: {}", e);
                    e
                })
                .and_then(|client| future::ok(XmppState { client, data }))
        })
        .map(|_| ())
    }
    fn enter_mucs<F, E>(
        self,
        _stop_future: F,
    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
    where
        F: Future<Error = E> + 'static,
        E: Into<failure::Error> + 'static,
    {
        let XmppConnection { account, state } = self;
        let account2 = account.clone();
        let account3 = account.clone();
        stream::iter_ok(account.chatrooms.clone())
            .fold(state, move |XmppState { client, mut data }, muc_jid| {
                data.counter += 1;
                let id_muc_presence = format!("id_muc_presence{}", data.counter);
                let muc_presence = stanzas::make_muc_presence(
                    &id_muc_presence,
                    account2.jid.clone(),
                    muc_jid.1.clone(),
                );
                info!("Sending muc presence... {}", String::from(&muc_presence));
                let account4 = account2.clone();
                use tokio::prelude::Sink;
                client
                    .send(Packet::Stanza(muc_presence))
                    .map_err(|e| {
                        error!("Error on send muc presence: {}", e);
                        account4
                    })
                    .and_then(|client| {
                        data.mucs.insert(muc_jid.0, muc_jid.1);
                        future::ok(XmppState { client, data })
                    })
            })
            .map(|state| XmppConnection {
                account: account3,
                state,
            })
    }
}

use tokio_xmpp::{Client, Event, Packet};
use tokio::prelude::future::{self, Either};
use tokio::prelude::stream;
use tokio::prelude::{Future, Stream};
use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant};
use super::stanzas;
use super::element_processor;
use crate::config;
#[derive(Debug)]
pub enum XmppCommand {
    /// Send message to someone by jid
    Chat {
        xmpp_to: xmpp_parsers::Jid,
        message: String,
    },
    /// Send message to MUC
    Chatroom { muc_id: String, message: String },
    /// Send ping request to the server to test connection
    Ping,
    /// Check iq requests if some have expired timeouts
    TimeoutCleanup,
}
/// trait of processing iq
/// each function consumes handlers and
/// returns false if connection should be reset
trait IqHandler {
    /// process result
    fn result(
        self: Box<Self>,
        conn: &mut XmppConnection,
        opt_element: Option<xmpp_parsers::Element>,
    ) -> bool;
    /// process error
    fn error(
        self: Box<Self>,
        conn: &mut XmppConnection,
        error: xmpp_parsers::stanza_error::StanzaError,
    ) -> bool;
    /// process tmeout
    fn timeout(self: Box<Self>, conn: &mut XmppConnection) -> bool;
}
struct AddRosterIqHandler {
    jid: xmpp_parsers::Jid,
}
impl IqHandler for AddRosterIqHandler {
    fn result(
        self: Box<Self>,
        conn: &mut XmppConnection,
        opt_element: Option<xmpp_parsers::Element>,
    ) -> bool {
        match opt_element {
            Some(element) => {
                warn!(
                    "Wrong payload when adding {} to roster: {}",
                    self.jid,
                    String::from(&element)
                );
            }
            None => {
                if conn.state.data.roster.contains_key(&self.jid) {
                    info!("Jid {} updated to roster", self.jid);
                } else {
                    info!("Jid {} added in roster", self.jid);
                    conn.state.data.roster.insert(
                        self.jid.clone(),
                        (
                            xmpp_parsers::roster::Subscription::None,
                            xmpp_parsers::roster::Ask::None,
                        ),
                    );
                }
                conn.process_jid(&self.jid);
            }
        }
        true
    }
    fn error(
        self: Box<Self>,
        _conn: &mut XmppConnection,
        _error: xmpp_parsers::stanza_error::StanzaError,
    ) -> bool {
        true
    }
    fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
        true // ignore
    }
}
struct PingIqHandler {}
impl IqHandler for PingIqHandler {
    fn result(
        self: Box<Self>,
        _conn: &mut XmppConnection,
        _opt_element: Option<xmpp_parsers::Element>,
    ) -> bool {
        info!("ping successed");
        true
    }
    fn error(
        self: Box<Self>,
        _conn: &mut XmppConnection,
        _error: xmpp_parsers::stanza_error::StanzaError,
    ) -> bool {
        false
    }
    fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
        false
    }
}
struct InitRosterIqHandler {}
impl IqHandler for InitRosterIqHandler {
    fn result(
        self: Box<Self>,
        conn: &mut XmppConnection,
        opt_element: Option<xmpp_parsers::Element>,
    ) -> bool {
        if let Some(result) = opt_element {
            use std::convert::TryInto;
            match result.try_into() as Result<xmpp_parsers::roster::Roster, _> {
                Ok(roster) => {
                    conn.state.data.roster_init = true;
                    conn.state.data.roster.clear();
                    info!("Got first roster:");
                    for i in roster.items {
                        info!(" >>> {:?}", i);
                        conn.state
                            .data
                            .roster
                            .insert(i.jid, (i.subscription, i.ask));
                    }
                    true
                }
                Err(e) => {
                    error!("Cann't parse roster: {}", e);
                    false
                }
            }
        } else {
            error!("No roster responded");
            false
        }
    }
    fn error(
        self: Box<Self>,
        _conn: &mut XmppConnection,
        _error: xmpp_parsers::stanza_error::StanzaError,
    ) -> bool {
        false
    }
    fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
        false
    }
}
#[derive(Default)]
struct XmppData {
    /// known roster data
    roster: HashMap<
        xmpp_parsers::Jid,
        (
            xmpp_parsers::roster::Subscription,
            xmpp_parsers::roster::Ask,
        ),
    >,
    /// if roster was initialized
    /// ToDo: remove it as it is used only for initialization
    roster_init: bool,
    /// if self-presence accepted
    /// ToDo: remove it as it is used only for initialization
    self_presence: bool,
    /// ids counter
    counter: usize,
    /// stanzas to send
    send_queue: VecDeque<minidom::Element>,
    /// outgoing mailbox
    outgoing_mailbox: HashMap<xmpp_parsers::Jid, Vec<String>>,
    /// muc id to muc jid
    mucs: HashMap<String, xmpp_parsers::Jid>,
    /// map from iq's id to handler of this type of iqs
    pending_ids: HashMap<String, (Instant, Box<dyn IqHandler>)>,
}
struct XmppState {
    client: Client,
    data: XmppData,
}
pub struct XmppConnection {
    account: std::rc::Rc<config::Account>,
    state: XmppState,
}
lazy_static! {
    static ref INCOMING: element_processor::Processor<XmppConnection, bool, xmpp_parsers::Element> = {
        let mut incoming = element_processor::Processor::new(&|_, e| {
            warn!("Unknown stanza {}", String::from(&e));
            true
        });
        incoming.register(&XmppConnection::incoming_iq_processing);
        incoming.register(&XmppConnection::incoming_presence_processing);
        incoming.register(&XmppConnection::incoming_message_processing);
        incoming
    };
}
pub struct MaybeXmppConnection {
    account: std::rc::Rc<config::Account>,
    state: Option<XmppState>,
}
impl From<XmppConnection> for MaybeXmppConnection {
    fn from(from: XmppConnection) -> MaybeXmppConnection {
        MaybeXmppConnection {
            account: from.account,
            state: Some(from.state),
        }
    }
}
impl From<config::Account> for MaybeXmppConnection {
    fn from(from: config::Account) -> MaybeXmppConnection {
        MaybeXmppConnection {
            account: std::rc::Rc::new(from),
            state: None,
        }
    }
}
impl From<std::rc::Rc<config::Account>> for MaybeXmppConnection {
    fn from(from: std::rc::Rc<config::Account>) -> MaybeXmppConnection {
        MaybeXmppConnection {
            account: from,
            state: None,
        }
    }
}
impl MaybeXmppConnection {
    /// connects if nothing connected
    /// don't connect only if stop_future resolved
    pub fn connect<F>(
        self,
        stop_future: F,
    ) -> impl Future<Item = XmppConnection, Error = failure::Error>
    where
        F: future::Future + Clone + 'static,
        <F as hyper::rt::Future>::Error: Into<failure::Error> + Send,
    {
        info!("xmpp connection...");
        let MaybeXmppConnection { account, state } = self;
        if let Some(state) = state {
            Box::new(future::ok(XmppConnection { account, state }))
                as Box<dyn Future<Item = _, Error = _>>
        } else {
            Box::new(
                stop_future
                    .clone()
                    .select2(
                        future::loop_fn(account, move |account| {
                            info!("xmpp initialization...");
                            let client =
                                Client::new_with_jid(account.jid.clone(), &account.password);
                            info!("xmpp initialized");
                            let stop_future2 = stop_future.clone();
                            let stop_future3 = stop_future.clone();
                            let stop_future4 = stop_future.clone();
                            // future to wait for online
                            Box::new(
                                XmppConnection {
                                    state: XmppState {
                                        client,
                                        data: std::default::Default::default(),
                                    },
                                    account,
                                }
                                .processing(XmppConnection::online, stop_future.clone())
                                .map_err(|(acc, _)| acc)
                                .and_then(|(conn, r)| match r {
                                    Ok(Either::A(_)) => future::ok(conn),
                                    Ok(Either::B(_)) => future::err(conn.account),
                                    Err(_e) => future::err(conn.account),
                                })
                                .and_then(|conn| conn.initial_roster(stop_future2))
                                .and_then(|conn| conn.self_presence(stop_future3))
                                .and_then(|conn| conn.enter_mucs(stop_future4))
                                .then(|r| match r {
                                    Ok(conn) => future::ok(future::Loop::Break(conn)),
                                    Err(acc) => future::ok(future::Loop::Continue(acc)),
                                }),
                            )
                        })
                        .map_err(|_: ()| ()),
                    )
                    .then(|r| match r {
                        Ok(Either::A((_x, _b))) => future::err(format_err!("Stop XMMP connection")),
                        Ok(Either::B((x, _a))) => future::ok(x),
                        Err(Either::A((e, _b))) => future::err(e.into()),
                        Err(Either::B((_, _a))) => {
                            future::err(format_err!("Cann't initiate XMPP connection"))
                        }
                    }),
            )
        }
    }
}
impl XmppConnection {
    /// base XMPP processing
    /// Returns false on error to disconnect
    fn xmpp_processing(&mut self, event: &Event) -> bool {
        match event {
            Event::Stanza(stanza) => INCOMING.process(self, stanza.clone()),
            Event::Online => true,
            e => {
                warn!("Unexpected event {:?}", e);
                false
            }
        }
    }
    /// Enforce to answer to IQ "set"
    fn incoming_iq_processing_set(
        &mut self,
        id: String,
        from: Option<xmpp_parsers::Jid>,
        element: minidom::Element,
    ) -> xmpp_parsers::iq::Iq {
        use std::convert::TryInto;
        if let Some(roster) =
            element.clone().try_into().ok() as Option<xmpp_parsers::roster::Roster>
        {
            // RFC 6212 2.1.6. Roster Push
            info!("Got roster push {} from {:?}", id, from);
            for i in roster.items {
                if let Some(ref mut rdata) = self.state.data.roster.get_mut(&i.jid) {
                    info!("Update {} in roster", i.jid);
                    rdata.0 = i.subscription;
                    rdata.1 = i.ask;
                } else {
                    info!("Add {} to roster", i.jid);
                    self.state
                        .data
                        .roster
                        .insert(i.jid.clone(), (i.subscription, i.ask));
                }
                self.process_jid(&i.jid);
            }
            return stanzas::make_roster_push_answer(id, self.state.client.jid.clone(), from);
        }
        warn!(
            "Unsupported IQ set request from {:?}: {}",
            from,
            String::from(&element)
        );
        stanzas::make_iq_unsupported_error(id, self.state.client.jid.clone(), from)
    }
    /// Enforce to answer to IQ "get"
    fn incoming_iq_processing_get(
        &mut self,
        id: String,
        from: Option<xmpp_parsers::Jid>,
        element: minidom::Element,
    ) -> xmpp_parsers::iq::Iq {
        use std::convert::TryInto;
        if let Some(_ping) = element.clone().try_into().ok() as Option<xmpp_parsers::ping::Ping> {
            info!("Got ping {} from {:?}", id, from);
            return stanzas::make_pong(&id, self.state.client.jid.clone(), from);
        }
        warn!(
            "Unsupported IQ get request from {:?}: {}",
            from,
            String::from(&element)
        );
        stanzas::make_iq_unsupported_error(id, self.state.client.jid.clone(), from)
    }
    fn incoming_iq_processing(&mut self, iq: xmpp_parsers::iq::Iq) -> bool {
        match iq.payload {
            xmpp_parsers::iq::IqType::Set(element) => {
                let iq_answer = self.incoming_iq_processing_set(iq.id, iq.from, element);
                self.state.data.send_queue.push_back(iq_answer.into());
            }
            xmpp_parsers::iq::IqType::Error(e) => {
                if let Some((_, handler)) = self.state.data.pending_ids.remove_entry(&iq.id) {
                    return handler.1.error(self, e);
                }
                error!("iq error: {:?}", e);
                return false;
            }
            xmpp_parsers::iq::IqType::Get(element) => {
                let iq_answer = self.incoming_iq_processing_get(iq.id, iq.from, element);
                self.state.data.send_queue.push_back(iq_answer.into());
            }
            xmpp_parsers::iq::IqType::Result(opt_element) => {
                if let Some((_, handler)) = self.state.data.pending_ids.remove_entry(&iq.id) {
                    return handler.1.result(self, opt_element);
                }
                warn!(
                    "Unwanted iq result id {} from {:?}: {:?}",
                    iq.id,
                    iq.from,
                    opt_element.map(|e| String::from(&e))
                );
            }
        }
        true
    }
    fn incoming_presence_processing(&mut self, presence: xmpp_parsers::presence::Presence) -> bool {
        if presence.from.as_ref() == Some(&self.state.client.jid) {
            info!("Self-presence accepted");
            self.state.data.self_presence = true;
        } else {
            warn!("Incoming presence stanza: {:?}", presence);
        }
        true
    }
    fn incoming_message_processing(&mut self, message: xmpp_parsers::message::Message) -> bool {
        for payload in message.payloads.iter() {
            use std::convert::TryInto;
            if let Some(_delay) =
                payload.clone().try_into().ok() as Option<xmpp_parsers::delay::Delay>
            {
                return true; // ignore delayed messages
            }
        }
        warn!("Incoming message stanza: {:?}", message);
        true
    }
    /// process event from xmpp stream
    /// returns from future when condition met
    /// or stop future was resolved.
    /// Return item if connection was preserved or error otherwise.
    /// Second part is a state of stop_future
    pub fn processing<S, F, T, E>(
        self,
        stop_condition: S,
        stop_future: F,
    ) -> impl Future<
        Item = (Self, Result<Either<F, T>, E>),
        Error = (std::rc::Rc<config::Account>, Result<Either<F, T>, E>),
    >
    where
        F: Future<Item = T, Error = E> + 'static,
        S: FnMut(&mut Self, Event) -> Result<bool, ()> + 'static,
        T: 'static,
        E: 'static,
    {
        future::loop_fn(
            (self, stop_future, stop_condition),
            |(xmpp, stop_future, mut stop_condition)| {
                // ToDo: check timeouts if iqs
                let XmppConnection {
                    state: XmppState { client, mut data },
                    account,
                } = xmpp;
                if let Some(send_element) = data.send_queue.pop_front() {
                    use tokio::prelude::Sink;
                    info!("Sending {}", String::from(&send_element));
                    Box::new(
                        client
                            .send(Packet::Stanza(send_element))
                            .select2(stop_future)
                            .then(move |r| match r {
                                Ok(Either::A((client, b))) => {
                                    Box::new(future::ok(future::Loop::Continue((
                                        XmppConnection {
                                            state: XmppState { client, data },
                                            account,
                                        },
                                        b,
                                        stop_condition,
                                    ))))
                                        as Box<dyn Future<Item = _, Error = _>>
                                }
                                Ok(Either::B((t, a))) => Box::new(a.then(|r| match r {
                                    Ok(client) => future::ok(future::Loop::Break((
                                        XmppConnection {
                                            state: XmppState { client, data },
                                            account,
                                        },
                                        Ok(Either::B(t)),
                                    ))),
                                    Err(se) => {
                                        warn!("XMPP sending error: {}", se);
                                        future::err((account, Ok(Either::B(t))))
                                    }
                                })),
                                Err(Either::A((e, b))) => {
                                    warn!("XMPP sending error: {}", e);
                                    Box::new(future::err((account, Ok(Either::A(b)))))
                                }
                                Err(Either::B((e, a))) => Box::new(a.then(|r| match r {
                                    Ok(client) => future::ok(future::Loop::Break((
                                        XmppConnection {
                                            state: XmppState { client, data },
                                            account,
                                        },
                                        Err(e),
                                    ))),
                                    Err(se) => {
                                        warn!("XMPP sending error: {}", se);
                                        future::err((account, Err(e)))
                                    }
                                })),
                            }),
                    ) as Box<dyn Future<Item = _, Error = _>>
                } else {
                    Box::new(
                        client
                            .into_future()
                            .select2(stop_future)
                            .then(move |r| match r {
                                Ok(Either::A(((event, client), b))) => {
                                    if let Some(event) = event {
                                        let mut xmpp = XmppConnection {
                                            state: XmppState { client, data },
                                            account,
                                        };
                                        if xmpp.xmpp_processing(&event) {
                                            match stop_condition(&mut xmpp, event) {
                                                Ok(true) => future::ok(future::Loop::Break((
                                                    xmpp,
                                                    Ok(Either::A(b)),
                                                ))),
                                                Ok(false) => future::ok(future::Loop::Continue((
                                                    xmpp,
                                                    b,
                                                    stop_condition,
                                                ))),
                                                Err(_e) => {
                                                    future::err((xmpp.account, Ok(Either::A(b))))
                                                }
                                            }
                                        } else {
                                            future::err((xmpp.account, Ok(Either::A(b))))
                                        }
                                    } else {
                                        future::err((account, Ok(Either::A(b))))
                                    }
                                }
                                Ok(Either::B((t, a))) => {
                                    if let Some(client) = a.into_inner() {
                                        future::ok(future::Loop::Break((
                                            XmppConnection {
                                                state: XmppState { client, data },
                                                account,
                                            },
                                            Ok(Either::B(t)),
                                        )))
                                    } else {
                                        future::err((account, Ok(Either::B(t))))
                                    }
                                }
                                Err(Either::A((e, b))) => {
                                    warn!("XMPP error: {}", e.0);
                                    future::err((account, Ok(Either::A(b))))
                                }
                                Err(Either::B((e, a))) => {
                                    if let Some(client) = a.into_inner() {
                                        future::ok(future::Loop::Break((
                                            XmppConnection {
                                                state: XmppState { client, data },
                                                account,
                                            },
                                            Err(e),
                                        )))
                                    } else {
                                        future::err((account, Err(e)))
                                    }
                                }
                            }),
                    )
                }
            },
        )
    }
    /// get connection and wait for online status and set presence
    /// returns error if something went wrong and xmpp connection is broken
    fn online(&mut self, event: Event) -> Result<bool, ()> {
        match event {
            Event::Online => {
                info!("Online!");
                Ok(true)
            }
            Event::Stanza(s) => {
                warn!("Stanza before online: {}", String::from(&s));
                Ok(false)
            }
            _ => {
                error!("Disconnected while online");
                Err(())
            }
        }
    }
    fn initial_roster<F, E>(
        self,
        stop_future: F,
    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
    where
        F: Future<Error = E> + 'static,
        E: 'static,
    {
        let XmppConnection {
            account,
            state: XmppState { client, mut data },
        } = self;
        use tokio::prelude::Sink;
        data.counter += 1;
        let id_init_roster = format!("id_init_roster{}", data.counter);
        let get_roster = stanzas::make_get_roster(&id_init_roster);
        let account2 = account.clone();
        info!("Quering roster... {}", String::from(&get_roster));
        data.pending_ids.insert(
            id_init_roster.clone(),
            (
                Instant::now() + Duration::from_secs(60),
                Box::new(InitRosterIqHandler {}),
            ),
        );
        client
            .send(Packet::Stanza(get_roster))
            .map_err(move |e| {
                error!("Error on querying roster: {}", e);
                account2
            })
            .and_then(move |client| {
                XmppConnection {
                    state: XmppState { client, data },
                    account,
                }
                .processing(move |conn, _| Ok(conn.state.data.roster_init), stop_future)
                .map_err(|(account, _)| account)
                .and_then(|(conn, r)| match r {
                    Ok(Either::A(_)) => future::ok(conn),
                    Ok(Either::B(_)) => future::err(conn.account),
                    Err(_e) => future::err(conn.account),
                })
            })
    }
    fn self_presence<F, E>(
        self,
        stop_future: F,
    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
    where
        F: Future<Error = E> + 'static,
        E: Into<failure::Error> + 'static,
    {
        let XmppConnection {
            account,
            state: XmppState { client, data },
        } = self;
        use tokio::prelude::Sink;
        let presence = stanzas::make_presence(&account);
        let account2 = account.clone();
        info!("Sending presence... {}", String::from(&presence));
        client
            .send(Packet::Stanza(presence))
            .map_err(|e| {
                error!("Error on send self-presence: {}", e);
                account2
            })
            .and_then(move |client| {
                XmppConnection {
                    state: XmppState { client, data },
                    account,
                }
                .processing(
                    move |conn, _| Ok(conn.state.data.self_presence),
                    stop_future,
                )
                .map_err(|(account, _)| account)
                .and_then(|(conn, r)| match r {
                    Ok(Either::A(_)) => future::ok(conn),
                    Ok(Either::B(_)) => future::err(conn.account),
                    Err(_e) => future::err(conn.account),
                })
            })
    }
    fn process_jid(&mut self, xmpp_to: &xmpp_parsers::Jid) {
        if let Some(ref mut mailbox) = self.state.data.outgoing_mailbox.get_mut(xmpp_to) {
            if !mailbox.is_empty() {
                if let Some(ref mut rdata) = self.state.data.roster.get_mut(xmpp_to) {
                    info!("Jid {} in roster", xmpp_to);
                    let sub_to = match rdata.0 {
                        xmpp_parsers::roster::Subscription::To => true,
                        xmpp_parsers::roster::Subscription::Both => true,
                        _ => false,
                    };
                    if sub_to {
                        info!("Subscribed to {}", xmpp_to);
                        self.state.data.send_queue.extend(
                            mailbox.drain(..).map(|message| {
                                stanzas::make_chat_message(xmpp_to.clone(), message)
                            }),
                        );
                    } else if rdata.1 == xmpp_parsers::roster::Ask::None {
                        info!("Not subscribed to {}", xmpp_to);
                        self.state
                            .data
                            .send_queue
                            .push_back(stanzas::make_ask_subscribe(xmpp_to.clone()));
                    }
                    let sub_from = match rdata.0 {
                        xmpp_parsers::roster::Subscription::From => true,
                        xmpp_parsers::roster::Subscription::Both => true,
                        _ => false,
                    };
                    if !sub_from {
                        info!("Not subscription from {}", xmpp_to);
                        self.state
                            .data
                            .send_queue
                            .push_back(stanzas::make_allow_subscribe(xmpp_to.clone()));
                    }
                } else {
                    info!("Jid {} not in roster", xmpp_to);
                    self.state.data.counter += 1;
                    let id_add_roster = format!("id_add_roster{}", self.state.data.counter);
                    let add_roster = stanzas::make_add_roster(&id_add_roster, xmpp_to.clone());
                    info!("Adding jid {} to roster id {}", xmpp_to, id_add_roster);
                    self.state.data.pending_ids.insert(
                        id_add_roster,
                        (
                            Instant::now() + Duration::from_secs(60),
                            Box::new(AddRosterIqHandler {
                                jid: xmpp_to.clone(),
                            }),
                        ),
                    );
                    self.state.data.send_queue.push_back(add_roster);
                }
            }
        }
    }
    pub fn process_command(&mut self, cmd: XmppCommand) {
        info!("Got command");
        match cmd {
            XmppCommand::Chat { xmpp_to, message } => {
                self.state
                    .data
                    .outgoing_mailbox
                    .entry(xmpp_to.clone())
                    .or_default()
                    .push(message);
                self.process_jid(&xmpp_to);
            }
            XmppCommand::Chatroom { muc_id, message } => {
                if let Some(muc) = self.state.data.mucs.get(&muc_id) {
                    self.state
                        .data
                        .send_queue
                        .push_back(stanzas::make_muc_message(muc.clone(), message));
                } else {
                    error!("Not found MUC {}", muc_id);
                }
            }
            XmppCommand::Ping => {
                self.state.data.counter += 1;
                let id_ping = format!("id_ping{}", self.state.data.counter);
                let ping = stanzas::make_ping(&id_ping, self.state.client.jid.clone());
                self.state.data.send_queue.push_back(ping);
                self.state.data.pending_ids.insert(
                    id_ping,
                    (
                        Instant::now() + Duration::from_secs(30),
                        Box::new(PingIqHandler {}),
                    ),
                );
            }
            XmppCommand::TimeoutCleanup => {
                let now = Instant::now();
                let timeouted: Vec<String> = self
                    .state
                    .data
                    .pending_ids
                    .iter()
                    .filter_map(|(id, (timeout, _))| {
                        if now >= *timeout {
                            Some(id.to_string())
                        } else {
                            None
                        }
                    })
                    .collect();
                let mut correct = true;
                timeouted.into_iter().for_each(|id| {
                    if let Some((_, handler)) = self.state.data.pending_ids.remove(&id) {
                        correct &= handler.timeout(&mut self);
                    }
                })
            }
        }
    }
    pub fn shutdown(self) -> impl Future<Item = (), Error = failure::Error> {
        info!("Shutdown connection");
        let XmppConnection { account, state } = self;
        stream::iter_ok(
            state
                .data
                .mucs
                .values()
                .map(std::clone::Clone::clone)
                .collect::<Vec<_>>(),
        )
        .fold(state, move |XmppState { client, data }, muc_jid| {
            let muc_presence =
                stanzas::make_muc_presence_leave(account.jid.clone(), muc_jid.clone());
            info!(
                "Sending muc leave presence... {}",
                String::from(&muc_presence)
            );
            use tokio::prelude::Sink;
            client
                .send(Packet::Stanza(muc_presence))
                .map_err(|e| {
                    error!("Error on send muc presence: {}", e);
                    e
                })
                .and_then(|client| future::ok(XmppState { client, data }))
        })
        .map(|_| ())
    }
    fn enter_mucs<F, E>(
        self,
        _stop_future: F,
    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
    where
        F: Future<Error = E> + 'static,
        E: Into<failure::Error> + 'static,
    {
        let XmppConnection { account, state } = self;
        let account2 = account.clone();
        let account3 = account.clone();
        stream::iter_ok(account.chatrooms.clone())
            .fold(state, move |XmppState { client, mut data }, muc_jid| {
                data.counter += 1;
                let id_muc_presence = format!("id_muc_presence{}", data.counter);
                let muc_presence = stanzas::make_muc_presence(
                    &id_muc_presence,
                    account2.jid.clone(),
                    muc_jid.1.clone(),
                );
                info!("Sending muc presence... {}", String::from(&muc_presence));
                let account4 = account2.clone();
                use tokio::prelude::Sink;
                client
                    .send(Packet::Stanza(muc_presence))
                    .map_err(|e| {
                        error!("Error on send muc presence: {}", e);
                        account4
                    })
                    .and_then(|client| {
                        data.mucs.insert(muc_jid.0, muc_jid.1);
                        future::ok(XmppState { client, data })
                    })
            })
            .map(|state| XmppConnection {
                account: account3,
                state,
            })
    }
}

use crate::config::Account;

use config::Account;

use xmpp_parsers::iq::Iq;
use xmpp_parsers::message::{Body, Message, MessageType};
use xmpp_parsers::ping::Ping;

use xmpp_parsers::roster::{Item, Roster};
use xmpp_parsers::stanza_error::{DefinedCondition, ErrorType, StanzaError};

    presence.into()
}
pub fn make_get_roster(id: &str) -> Element {
    Iq::from_get(
        id,
        Roster {
            items: vec![],
            ver: None,
        },
    )
    .into()
}
pub fn make_add_roster(id: &str, jid: xmpp_parsers::Jid) -> Element {
    Iq::from_set(
        id,
        Roster {
            items: vec![Item {
                jid,
                name: None,
                subscription: xmpp_parsers::roster::Subscription::None,
                ask: xmpp_parsers::roster::Ask::None,
                groups: vec![],
            }],
            ver: None,
        },
    )
    .into()
}
pub fn make_roster_push_answer(
    id: String,
    from: xmpp_parsers::Jid,
    to: Option<xmpp_parsers::Jid>,
) -> Iq {
    let mut answer = Iq::from_result(id, None as Option<Roster>);
    answer.from = Some(from);
    answer.to = to;
    answer
}
pub fn make_ask_subscribe(jid: xmpp_parsers::Jid) -> Element {
    let mut presence = Presence::new(PresenceType::Subscribe);
    presence.to = Some(jid);
    presence.into()
}
pub fn make_allow_subscribe(jid: xmpp_parsers::Jid) -> Element {
    let mut presence = Presence::new(PresenceType::Subscribed);
    presence.to = Some(jid);
    presence.into()
}
pub fn make_chat_message(jid: xmpp_parsers::Jid, text: String) -> Element {
    let mut message = Message::new(Some(jid));
    message.bodies.insert(String::new(), Body(text));
    message.type_ = MessageType::Chat;
    message.into()
}
pub fn make_muc_presence(id: &str, from: xmpp_parsers::Jid, to: xmpp_parsers::Jid) -> Element {
    let mut presence = Presence::new(PresenceType::None);
    presence.from = Some(from);
    presence.to = Some(to);
    presence.id = Some(id.to_string());
    presence.add_payload(xmpp_parsers::muc::Muc::new());

pub fn make_muc_message(to: xmpp_parsers::Jid, text: String) -> Element {
    let mut message = Message::new(Some(to.into_bare_jid()));
    message.bodies.insert(String::new(), Body(text));
    message.type_ = MessageType::Groupchat;
    message.into()
}
pub fn make_muc_presence_leave(from: xmpp_parsers::Jid, to: xmpp_parsers::Jid) -> Element {
    let mut presence = Presence::new(PresenceType::Unavailable);
    presence.from = Some(from);
    presence.to = Some(to);
    presence.into()
}
pub fn make_ping(id: &str, from: xmpp_parsers::Jid) -> Element {
    let mut ping = Iq::from_get(id, Ping);
    ping.to = Some(from.clone().into_domain_jid());
    ping.from = Some(from);
    ping.into()
}
pub fn make_pong(id: &str, from: xmpp_parsers::Jid, to: Option<xmpp_parsers::Jid>) -> Iq {
    let mut pong = Iq::from_result(id, None as Option<Roster>);
    pong.from = Some(from);
    pong.to = to;
    pong
}
pub fn make_iq_unsupported_error(
    id: String,
    from: xmpp_parsers::Jid,
    to: Option<xmpp_parsers::Jid>,
) -> Iq {
    let mut error = Iq::from_error(
        id,
        StanzaError {
            type_: ErrorType::Cancel,
            by: Some(from.clone()),
            defined_condition: DefinedCondition::ServiceUnavailable,
            texts: std::collections::BTreeMap::new(),
            other: None,
        },
    );
    error.from = Some(from);
    error.to = to;
    error
}

use tokio::prelude::future::{self, Either, Future};

use tokio::prelude::future::{self, Either};

use tokio::prelude::{Future, Stream};

use crate::config;

use tokio_xmpp::{Client, Event};

mod element_processor;

use config;

pub struct XmppConnection {
    account: std::rc::Rc<config::Account>,
    inner: Option<(stream::SplitSink<Client>, stream::SplitStream<Client>)>,
}
impl XmppConnection {
    fn new(account: config::Account) -> XmppConnection {
        XmppConnection {
            account: std::rc::Rc::new(account),
            inner: None,
        }
    }

mod xmpp_connection;
use xmpp_connection::MaybeXmppConnection;

    /// connects if nothing connected
    /// Error shoud be !
    fn connect<E: 'static>(self) -> impl Future<Item = Self, Error = E> {
        info!("xmpp connection...");
        let XmppConnection { account, inner } = self;
        if let Some(inner) = inner {
            Box::new(future::ok(XmppConnection {
                account,
                inner: Some(inner),
            })) as Box<Future<Item = _, Error = E>>
        } else {
            Box::new(future::loop_fn(account, |account| {
                info!("xmpp initialization...");
                let mut res_client = Client::new(&account.jid, &account.password);
                while let Err(e) = res_client {
                    error!("Cann't init xmpp client: {}", e);
                    res_client = Client::new(&account.jid, &account.password);
                }
                let client = res_client.expect("Cann't init xmpp client");
                info!("xmpp initialized");

pub use xmpp_connection::XmppCommand;

                // future to wait for online
                Self::online(client.split(), account)
                    .and_then(Self::self_presence)
                    .then(|r| match r {
                        Ok(conn) => future::ok(future::Loop::Break(conn)),
                        Err(acc) => future::ok(future::Loop::Continue(acc)),
                    })
            }))
        }
    }

struct XmppProcessState<F, S> {

    /// get connection and wait for online status and set presence
    /// returns error if something went wrong
    fn online(
        (sink, stream): (stream::SplitSink<Client>, stream::SplitStream<Client>),
        account: std::rc::Rc<config::Account>,
    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>> {
        Box::new(future::loop_fn(
            (sink, stream, account),
            |(sink, stream, account)| {
                stream.into_future().then(|r| match r {
                    Ok((event, stream)) => match event {
                        Some(Event::Online) => {
                            info!("Online");
                            future::ok(future::Loop::Break(XmppConnection {
                                account,
                                inner: Some((sink, stream)),
                            }))
                        }
                        Some(Event::Stanza(s)) => {
                            info!("xmpp stanza: {:?}", s);
                            future::ok(future::Loop::Continue((sink, stream, account)))
                        }
                        _ => {
                            warn!("Disconnected");
                            future::err(account)
                        }
                    },
                    Err((e, _)) => {
                        error!("xmpp receive error: {}", e);
                        future::err(account)
                    }
                })
            },
        ))
    }
    fn self_presence(self) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>> {
        let XmppConnection { account, inner } = self;
        if let Some((sink, stream)) = inner {
            use tokio::prelude::Sink;
            let presence = stanzas::make_presence(&account);
            info!("Sending presence...");
            Box::new(
                sink.send(presence)
                    .map_err(|e| {
                        error!("Error on send self-presence: {}", e);
                        "Cann't send self-presence".to_owned()
                    }).join(
                        future::loop_fn((account.clone(), stream), |(account, stream)| {
                            stream
                                .into_future()
                                .map_err(|(e, _)| {
                                    error!("Error on reading self-presence: {}", e);
                                    "Cann't read self-presence".to_owned()
                                }).and_then(|(event, stream)| match event {
                                    Some(event) => {
                                        if let tokio_xmpp::Event::Stanza(e) = event {
                                            info!("Get stanza: {:?}", e);
                                            if e.name() == "presence"
                                                && e.attr("from")
                                                    .map_or(false, |f| f == account.jid)
                                                && e.attr("to").map_or(false, |f| f == account.jid)
                                            {
                                                info!("Self presence");
                                                future::ok(future::Loop::Break(stream))
                                            } else {
                                                future::ok(future::Loop::Continue((
                                                    account, stream,
                                                )))
                                            }
                                        } else {
                                            future::err("Got wrong event".to_owned())
                                        }
                                    }
                                    None => future::err("Got closed stream".to_owned()),
                                })
                        }).map_err(|e| format!("waiting self-presence: {}", e)),
                    ).then(|r| match r {
                        Err(e) => {
                            error!("Self-presence waiting error: {}", e);
                            future::err(account)
                        }
                        Ok(inner) => future::ok(XmppConnection {
                            account,
                            inner: Some(inner),
                        }),
                    }),
            )
        } else {
            warn!("Don't gen connection on self-presence");
            Box::new(future::err(account)) as Box<Future<Item = _, Error = _>>
        }
    }
}
#[derive(Debug)]
pub struct XmppCommand;
struct XmppState<F, S> {

    conn: MaybeXmppConnection,

    conn: XmppConnection,

impl<F, S> XmppProcessState<F, S> {
    fn new(cmd_recv: S, signal: F, conn: MaybeXmppConnection) -> XmppProcessState<F, S> {
        XmppProcessState {

impl<F, S> XmppState<F, S> {
    fn new(cmd_recv: S, signal: F, conn: XmppConnection) -> XmppState<F, S> {
        XmppState {

) -> impl future::Future<Item = (), Error = failure::Error>

) -> impl future::Future<Item = (), Error = tokio::io::Error>

    F: future::Future + Clone + 'static,
    <F as hyper::rt::Future>::Error: std::fmt::Display + Into<failure::Error> + Send,
    S: stream::Stream<Item = XmppCommand> + 'static,

    F: future::Future<Item = ()> + 'static,
    S: stream::Stream<Item = XmppCommand> + 'static

    let conn = account.into();

    let signal = signal
        .map_err(|_| tokio::io::Error::new(tokio::io::ErrorKind::Other, "Wrong shutdown signal"));
    let conn = XmppConnection::new(account);

    future::loop_fn(XmppProcessState::new(cmd_recv, signal, conn), |s| {
        let XmppProcessState {

    future::loop_fn(XmppState::new(cmd_recv, signal, conn), |s| {
        let XmppState {

        conn.connect(signal.clone()).and_then(|conn| {
            info!("xmpp connected!");
            conn.processing(|_, _| Ok(false), cmd_recv.into_future())
                .then(|r| match r {
                    Ok((mut conn, r)) => match r {
                        Ok(Either::A(f)) => {
                            if let Some(cmd_recv) = f.into_inner() {
                                future::ok(future::Loop::Continue(XmppProcessState {
                                    cmd_recv,
                                    signal,
                                    conn: conn.into(),
                                }))
                            } else {
                                error!("Command receiver is gone");
                                future::ok(future::Loop::Break(Some(conn)))
                            }
                        }
                        Ok(Either::B((cmd, cmd_recv))) => {
                            if let Some(cmd) = cmd {
                                conn.process_command(cmd);
                                future::ok(future::Loop::Continue(XmppProcessState {
                                    cmd_recv,
                                    signal,
                                    conn: conn.into(),
                                }))
                            } else {
                                future::ok(future::Loop::Break(Some(conn)))
                            }
                        }
                        Err(_) => {
                            error!("Command receiver is broken");
                            future::ok(future::Loop::Break(Some(conn)))
                        }
                    },
                    Err((account, r)) => match r {
                        Ok(Either::A(f)) => {
                            if let Some(cmd_recv) = f.into_inner() {
                                future::ok(future::Loop::Continue(XmppProcessState {
                                    cmd_recv,
                                    signal,
                                    conn: account.into(),
                                }))
                            } else {
                                future::err(format_err!("Command receiver is gone"))
                            }
                        }
                        Ok(Either::B((cmd, cmd_recv))) => {
                            if let Some(_cmd) = cmd {
                                error!("Xmpp connection broken while get command");
                                future::ok(future::Loop::Continue(XmppProcessState {
                                    cmd_recv,
                                    signal,
                                    conn: account.into(),
                                }))
                            } else {
                                future::ok(future::Loop::Break(None))
                            }
                        }
                        Err(_) => future::err(format_err!("Command receiver is broken")),
                    },
                })
        })
    })
    .and_then(|opt_conn| {
        if let Some(conn) = opt_conn {
            Box::new(conn.shutdown()) as Box<dyn Future<Item = (), Error = _>>
        } else {
            Box::new(future::ok(()))
        }
    })

        signal
            .select2(conn.connect().and_then(|conn| {
                info!("xmpp connected!");
                cmd_recv
                    .into_future()
                    .map_err(|_| {
                        error!("Got error on recv cmd");
                        tokio::io::Error::new(tokio::io::ErrorKind::Other, "Receive cmd error")
                    }).map(|f| (f, conn))
            })).then(|r| {
                match r {
                    Ok(Either::A((_x, b))) => {
                        info!("Got signal");
                        // got signal, breaks
                        Box::new(b.map(|b| future::Loop::Break((Some((b.0).1), b.1))))
                            as Box<Future<Item = _, Error = _>>
                    }
                    Ok(Either::B((x, a))) => {
                        info!("Got cmd");
                        // got cmd, continue
                        Box::new(future::ok(future::Loop::Continue(XmppState::new(
                            (x.0).1,
                            a,
                            x.1,
                        )))) as Box<Future<Item = _, Error = _>>
                    }
                    Err(Either::A((e, b))) => {
                        // got signal error, breaks
                        error!("Signal error: {}", e);
                        Box::new(b.map(|b| future::Loop::Break((Some((b.0).1), b.1))))
                            as Box<Future<Item = _, Error = _>>
                    }
                    Err(Either::B((e, _a))) => {
                        // got cmd error, its bad
                        error!("Cmd error: {}", e);
                        Box::new(future::err(tokio::io::Error::new(
                            tokio::io::ErrorKind::Other,
                            "Cmd error",
                        ))) as Box<Future<Item = _, Error = _>>
                    }
                }
            })
    }).and_then(
        |(opt_cmd_recv, _conn): (Option<S>, XmppConnection)| {
            if let Some(cmd_recv) = opt_cmd_recv {
                // process left commands
                info!("Stop accepting commands");
                Box::new(cmd_recv.for_each(|_cmd| future::ok(())).map_err(|_| {
                    tokio::io::Error::new(tokio::io::ErrorKind::Other, "cmd receiver last error")
                })) as Box<Future<Item = (), Error = tokio::io::Error>>
            } else {
                Box::new(future::err(tokio::io::Error::new(
                    tokio::io::ErrorKind::Other,
                    "cmd receiver gone",
                )))
            }
        },
    )

type Func<S, T, E> = dyn Fn(&mut S, E) -> T + Sync;
pub struct Processor<S: 'static, T: 'static, E: Clone + 'static> {
    processors: Vec<Box<Func<S, Option<T>, E>>>,
    default: &'static Func<S, T, E>,
}
impl<S: 'static, T: 'static, E: Clone + 'static> Processor<S, T, E> {
    pub fn new<F>(f: &'static F) -> Processor<S, T, E>
    where
        F: Fn(&mut S, E) -> T + Sync + 'static,
    {
        Processor {
            processors: vec![],
            default: f,
        }
    }
    pub fn register<F, A>(&mut self, f: &'static F)
    where
        F: Fn(&mut S, A) -> T + Sync + 'static,
        A: std::convert::TryFrom<E>,
    {
        self.processors.push(Box::new(move |s, e: E| {
            use std::convert::TryInto;
            (e.try_into().ok() as Option<A>).map(|a| f(s, a))
        }));
    }
    pub fn process(&self, s: &mut S, e: E) -> T {
        for processor in self.processors.iter() {
            match processor(s, e.clone()) {
                Some(t) => return t,
                None => continue,
            }
        }
        (*self.default)(s, e)
    }
}

type Func<S, T, E> = dyn Fn(&mut S, E) -> T + Sync;
pub struct Processor<S: 'static, T: 'static, E: Clone + 'static> {
    processors: Vec<Box<Func<S, Option<T>, E>>>,
    default: &'static Func<S, T, E>,
}
impl<S: 'static, T: 'static, E: Clone + 'static> Processor<S, T, E> {
    pub fn new<F>(f: &'static F) -> Processor<S, T, E>
    where
        F: Fn(&mut S, E) -> T + Sync + 'static,
    {
        Processor {
            processors: vec![],
            default: f,
        }
    }
    pub fn register<F, A>(&mut self, f: &'static F)
    where
        F: Fn(&mut S, A) -> T + Sync + 'static,
        A: std::convert::TryFrom<E>,
    {
        self.processors.push(Box::new(move |s, e: E| {
            use std::convert::TryInto;
            (e.try_into().ok() as Option<A>).map(|a| f(s, a))
        }));
    }
    pub fn process(&self, s: &mut S, e: E) -> T {
        for processor in self.processors.iter() {
            match processor(s, e.clone()) {
                Some(t) => return t,
                None => continue,
            }
        }
        (*self.default)(s, e)
    }
}

use std::fmt::Debug;

    T: Debug,

        self.receiver.poll()

        self.receiver.poll();

    T: Debug,

#![deny(bare_trait_objects)]

extern crate minidom;

#[macro_use]
extern crate failure;

#[macro_use]
extern crate lazy_static;

use hyper::service::service_fn;

use tokio::prelude::{Future, Sink, Stream};

use tokio::prelude::Sink;

use crate::config::Config;

use config::Config;

use crate::xmpp::{xmpp_process, XmppCommand};

use xmpp::{xmpp_process, XmppCommand};

use crate::stoppable_receiver::stop_receiver;
fn body_to_string(req: Request<Body>) -> impl Future<Item = String, Error = failure::Error> {
    req.into_body()
        .map_err(std::convert::Into::into)
        .fold(String::new(), |mut acc, ch| {
            std::str::from_utf8(&*ch).map(|s| {
                acc.push_str(s);
                acc
            })
        })
}
struct ServiceCmd {
    cmd_send: tokio_channel::mpsc::Sender<XmppCommand>,
}
impl hyper::service::Service for ServiceCmd {
    type ReqBody = Body;
    type ResBody = Body;
    type Error = failure::Error;
    type Future =
        Box<dyn Future<Item = Response<Self::ResBody>, Error = Self::Error> + Send + 'static>;
    fn call(&mut self, req: Request<Self::ReqBody>) -> Self::Future {
        let xmpp_to_opt = req.headers().get("X-XMPP-To");
        let xmpp_to_res: Result<xmpp_parsers::Jid, failure::Error> = xmpp_to_opt.map_or_else(
            || Err(format_err!("No X-XMPP-To header")),
            |xmpp_to| {
                std::str::from_utf8(xmpp_to.as_bytes())
                    .map_err(std::convert::Into::into)
                    .and_then(|s| std::str::FromStr::from_str(s).map_err(std::convert::Into::into))
            },
        );
        let xmpp_muc_opt = req
            .headers()
            .get("X-XMPP-Muc")
            .map(|h| h.to_str().map(std::string::ToString::to_string));
        match (xmpp_muc_opt, xmpp_to_res) {
            (None, Err(err)) => {
                warn!("Unknown destination: {}", err);
                Box::new(tokio::prelude::future::result(
                    Response::builder()
                        .status(hyper::StatusCode::BAD_REQUEST)
                        .body(Body::from(format!("Unknown destination: {}", err)))
                        .map_err(std::convert::Into::into),
                )) as Box<dyn Future<Item = _, Error = _> + Send + 'static>
            }
            (None, Ok(xmpp_to)) => {
                info!("Got request. Reading body...");
                let cmd_send = self.cmd_send.clone();
                Box::new(body_to_string(req).and_then(move |message: String| {
                    if !message.is_empty() {
                        Box::new(
                            cmd_send
                                .clone()
                                .send(XmppCommand::Chat { xmpp_to, message })
                                .then(|r| match r {
                                    Ok(_) => tokio::prelude::future::ok(Response::new(Body::from(
                                        "Accepted",
                                    ))),
                                    Err(e) => {
                                        error!("Command sent error: {}", e);
                                        tokio::prelude::future::result(
                                            Response::builder()
                                                .status(hyper::StatusCode::BAD_REQUEST)
                                                .body(Body::from(format!(
                                                    "Command sent error: {}",
                                                    e
                                                ))),
                                        )
                                    }
                                })
                                .map_err(std::convert::Into::into),
                        )
                    } else {
                        warn!("Empty message");
                        Box::new(tokio::prelude::future::result(
                            Response::builder()
                                .status(hyper::StatusCode::BAD_REQUEST)
                                .body(Body::from("Empty message"))
                                .map_err(std::convert::Into::into),
                        ))
                            as Box<dyn Future<Item = _, Error = _> + Send + 'static>
                    }
                })) as Box<dyn Future<Item = _, Error = _> + Send + 'static>
            }
            (Some(Ok(muc_id)), _) => {
                info!("Got MUC request. Reading body...");
                let cmd_send = self.cmd_send.clone();
                Box::new(body_to_string(req).and_then(move |message: String| {
                    if !message.is_empty() {
                        Box::new(
                            cmd_send
                                .clone()
                                .send(XmppCommand::Chatroom { muc_id, message })
                                .then(|r| match r {
                                    Ok(_) => tokio::prelude::future::ok(Response::new(Body::from(
                                        "Accepted",
                                    ))),
                                    Err(e) => {
                                        error!("Command sent error: {}", e);
                                        tokio::prelude::future::result(
                                            Response::builder()
                                                .status(hyper::StatusCode::BAD_REQUEST)
                                                .body(Body::from(format!(
                                                    "Command sent error: {}",
                                                    e
                                                ))),
                                        )
                                    }
                                })
                                .map_err(std::convert::Into::into),
                        )
                    } else {
                        warn!("Empty message");
                        Box::new(tokio::prelude::future::result(
                            Response::builder()
                                .status(hyper::StatusCode::BAD_REQUEST)
                                .body(Body::from("Empty message"))
                                .map_err(std::convert::Into::into),
                        ))
                            as Box<dyn Future<Item = _, Error = _> + Send + 'static>
                    }
                })) as Box<dyn Future<Item = _, Error = _> + Send + 'static>
            }
            (Some(Err(err)), _) => {
                warn!("Unknown MUC destination: {}", err);
                Box::new(tokio::prelude::future::result(
                    Response::builder()
                        .status(hyper::StatusCode::BAD_REQUEST)
                        .body(Body::from(format!("Unknown MUC destination: {}", err)))
                        .map_err(std::convert::Into::into),
                )) as Box<dyn Future<Item = _, Error = _> + Send + 'static>
            }
        }
    }
}
struct MakeServiceCmd {
    cmd_send: tokio_channel::mpsc::Sender<XmppCommand>,
}
impl<Ctx> hyper::service::MakeService<Ctx> for MakeServiceCmd {
    type ReqBody = Body;
    type ResBody = Body;
    type Error = failure::Error;
    type Service = ServiceCmd;
    type Future = tokio::prelude::future::FutureResult<ServiceCmd, Self::MakeError>;
    type MakeError = hyper::http::Error;

use stoppable_receiver::stop_receiver;

    fn make_service(&mut self, _ctx: Ctx) -> Self::Future {
        tokio::prelude::future::ok(ServiceCmd {
            cmd_send: self.cmd_send.clone(),
        })
    }
}

        )
        .get_matches();

        ).get_matches();

    let config = Config::read(args.value_of("config").expect("Mandatory option config"))

    let config = Config::new(args.value_of("config").expect("Mandatory option config"))

        .map_err(|e| {
            error!("Cann't get CTRL+C signal: {}", e.0);
            e.0
        })

        .map_err(|e| error!("Cann't get CTRL+C signal: {}", e.0))

        .serve(MakeServiceCmd {
            cmd_send: cmd_send.clone(),
        })
        .with_graceful_shutdown(ctrl_c.clone().map(|_| ()))

        .serve(move || {
            let cmd_send = cmd_send.clone();
            service_fn(move |_req: Request<Body>| {
                info!("Got request");
                cmd_send.clone().send(XmppCommand {}).then(|r| match r {
                    Ok(_) => tokio::prelude::future::ok(Response::new(Body::from("Accepted"))),
                    Err(e) => {
                        error!("Command sent error: {}", e);
                        tokio::prelude::future::result(
                            Response::builder()
                                .status(hyper::StatusCode::BAD_REQUEST)
                                .body(Body::from(format!("Command sent error: {}", e))),
                        )
                    }
                })
            })
        }).with_graceful_shutdown(ctrl_c.clone().map(|_| ()))

    if let Some(ping) = config.account.ping {
        let ping = tokio::timer::Interval::new_interval(std::time::Duration::from_secs(ping));
        rt.spawn(
            ping.map_err(|e| {
                error!("Ping error: {}", e);
            })
            .for_each(move |_| {
                cmd_send
                    .clone()
                    .send(XmppCommand::Ping)
                    .map_err(|e| {
                        error!("Ping command error: {}", e);
                    })
                    .map(|_| ())
            })
            .select(
                ctrl_c
                    .clone()
                    .map(|_| ())
                    .map_err(|e| error!("ping server error: {}", e)),
            )
            .map(|_| ())
            .map_err(|_| ()),
        );
    }

    let xmpp_join = std::thread::spawn(move || -> Result<(), failure::Error> {

    let xmpp_join = std::thread::spawn(move || -> Result<(), tokio::io::Error> {

        let result = ctrt.block_on(xmpp_process(ctrl_c.clone(), recv, config.account));

        let result = ctrt.block_on(xmpp_process(
            ctrl_c.clone().map(|_| ()),
            recv,
            config.account,
        ));

use std::collections::HashMap;

    #[serde(deserialize_with = "deserialize_jid")]
    pub jid: xmpp_parsers::Jid,

    pub jid: String,

    #[serde(default, deserialize_with = "deserialize_jid_map")]
    pub chatrooms: HashMap<String, xmpp_parsers::Jid>,
    pub ping: Option<u64>,

    pub fn read<P: AsRef<Path>>(path: P) -> io::Result<Self> {

    pub fn new<P: AsRef<Path>>(path: P) -> io::Result<Config> {

}
fn deserialize_jid<'de, D>(deserializer: D) -> Result<xmpp_parsers::Jid, D::Error>
where
    D: serde::Deserializer<'de>,
{
    use serde::Deserialize;
    let s = String::deserialize(deserializer)?;
    std::str::FromStr::from_str(&s).map_err(serde::de::Error::custom)
}
fn deserialize_jid_map<'de, D>(
    deserializer: D,
) -> Result<HashMap<String, xmpp_parsers::Jid>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    use serde::Deserialize;
    let s = HashMap::<String, String>::deserialize(deserializer)?;
    let size = s.len();
    s.into_iter()
        .map(|(k, v)| (k, std::str::FromStr::from_str(&v)))
        .take_while(|(_k, r)| r.is_ok())
        .fold(Ok(HashMap::with_capacity(size)), |res, (k, r)| match res {
            Ok(mut res) => match r {
                Ok(v) => {
                    res.insert(k, v);
                    Ok(res)
                }
                Err(e) => Err(e),
            },
            Err(e) => Err(e),
        })
        .map_err(serde::de::Error::custom)

### XMPP client daemon
#### Sending messages
```
curl http://localhost:8083/ -H "X-XMPP-To: some@domain.org" -d "Test"
```
#### Sending messages to MUC
```
curl http://localhost:8083/ -H "X-XMPP-Muc: smac" -d "Test"
```

edition = "2018"

tokio-xmpp = "1.0.0"
failure = "0.1.5"

tokio-xmpp = "0.2"

toml = "0.5"

toml = "0.4"

env_logger = "0.6"
xmpp-parsers = "0.13"
minidom = "=0.10.0" # xmpp-parsers
lazy_static = "1.3.0" # ToDo: remove after const fn will be powerfull enough

env_logger = "0.5"
minidom = "=0.9.1"
xmpp-parsers = "0.11"

# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
[[package]]
name = "MacTypes-sys"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
 "libc 0.2.51 (registry+https://github.com/rust-lang/crates.io-index)",
]

version = "0.7.3"

version = "0.6.8"

 "memchr 2.2.0 (registry+https://github.com/rust-lang/crates.io-index)",

 "memchr 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)",

 "winapi 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",

 "winapi 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.3.10"

version = "0.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "arrayref"
version = "0.3.5"

version = "0.4.10"

version = "0.4.7"

 "nodrop 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)",

 "nodrop 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)",

 "libc 0.2.51 (registry+https://github.com/rust-lang/crates.io-index)",

 "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",

 "winapi 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",

 "winapi 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",

name = "autocfg"
version = "0.1.2"

name = "backtrace"
version = "0.2.3"

dependencies = [
 "backtrace-sys 0.1.24 (registry+https://github.com/rust-lang/crates.io-index)",
 "cfg-if 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
 "dbghelp-sys 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "kernel32-sys 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
 "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",
 "rustc-demangle 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)",
 "winapi 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
]

version = "0.3.15"

version = "0.3.9"

 "autocfg 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
 "backtrace-sys 0.1.28 (registry+https://github.com/rust-lang/crates.io-index)",
 "cfg-if 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
 "libc 0.2.51 (registry+https://github.com/rust-lang/crates.io-index)",
 "rustc-demangle 0.1.14 (registry+https://github.com/rust-lang/crates.io-index)",
 "winapi 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",

 "backtrace-sys 0.1.24 (registry+https://github.com/rust-lang/crates.io-index)",
 "cfg-if 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
 "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",
 "rustc-demangle 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)",
 "winapi 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.1.28"

version = "0.1.24"

 "cc 1.0.35 (registry+https://github.com/rust-lang/crates.io-index)",
 "libc 0.2.51 (registry+https://github.com/rust-lang/crates.io-index)",

 "cc 1.0.25 (registry+https://github.com/rust-lang/crates.io-index)",
 "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.10.1"

version = "0.9.3"

 "byteorder 1.3.1 (registry+https://github.com/rust-lang/crates.io-index)",

 "byteorder 1.2.7 (registry+https://github.com/rust-lang/crates.io-index)",
 "safemem 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.8.0"

version = "0.7.1"

 "byte-tools 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
 "crypto-mac 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "digest 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "opaque-debug 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",

 "byte-tools 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "crypto-mac 0.5.2 (registry+https://github.com/rust-lang/crates.io-index)",
 "digest 0.7.6 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
 "block-padding 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
 "byte-tools 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
 "byteorder 1.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
 "generic-array 0.12.0 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "block-padding"
version = "0.1.3"

version = "0.3.3"

 "byte-tools 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",

 "arrayref 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
 "byte-tools 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.3.1"

version = "0.2.0"

version = "1.3.1"

version = "1.2.7"

version = "0.4.12"

version = "0.4.10"

 "byteorder 1.3.1 (registry+https://github.com/rust-lang/crates.io-index)",

 "byteorder 1.2.7 (registry+https://github.com/rust-lang/crates.io-index)",

version = "1.0.35"

version = "1.0.25"

version = "0.1.7"

version = "0.1.6"

 "time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)",

 "time 0.1.40 (registry+https://github.com/rust-lang/crates.io-index)",

version = "2.33.0"

version = "2.32.0"

 "strsim 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "textwrap 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)",

 "strsim 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "textwrap 0.10.0 (registry+https://github.com/rust-lang/crates.io-index)",

name = "constant_time_eq"
version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]

 "libc 0.2.51 (registry+https://github.com/rust-lang/crates.io-index)",

 "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",

 "libc 0.2.51 (registry+https://github.com/rust-lang/crates.io-index)",

 "libc 0.2.43 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.7.1"

version = "0.6.1"

 "crossbeam-epoch 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
 "crossbeam-utils 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)",

 "crossbeam-epoch 0.5.2 (registry+https://github.com/rust-lang/crates.io-index)",
 "crossbeam-utils 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.7.1"

version = "0.5.2"

 "arrayvec 0.4.10 (registry+https://github.com/rust-lang/crates.io-index)",
 "cfg-if 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
 "crossbeam-utils 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)",
 "lazy_static 1.3.0 (registry+https://github.com/rust-lang/crates.io-index)",

 "arrayvec 0.4.7 (registry+https://github.com/rust-lang/crates.io-index)",
 "cfg-if 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
 "crossbeam-utils 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "lazy_static 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)",

name = "crossbeam-queue"
version = "0.1.2"

name = "crossbeam-utils"
version = "0.5.0"

dependencies = [
 "crossbeam-utils 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)",
]

name = "crossbeam-utils"
version = "0.6.5"

name = "crypto-mac"
version = "0.5.2"

 "cfg-if 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
 "lazy_static 1.3.0 (registry+https://github.com/rust-lang/crates.io-index)",

 "constant_time_eq 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
 "generic-array 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)",

name = "crypto-mac"
version = "0.7.0"

name = "dbghelp-sys"
version = "0.2.0"

 "generic-array 0.12.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "subtle 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)",

 "winapi 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
 "winapi-build 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.8.0"

version = "0.7.6"

 "generic-array 0.12.0 (registry+https://github.com/rust-lang/crates.io-index)",

 "generic-array 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.8.17"

version = "0.8.10"

 "cfg-if 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",

 "cfg-if 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.6.1"

version = "0.5.13"

 "humantime 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
 "regex 1.1.5 (registry+https://github.com/rust-lang/crates.io-index)",

 "humantime 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
 "log 0.4.5 (registry+https://github.com/rust-lang/crates.io-index)",
 "regex 1.0.5 (registry+https://github.com/rust-lang/crates.io-index)",

]
[[package]]
name = "error-chain"
version = "0.1.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
 "backtrace 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)",

 "backtrace 0.3.15 (registry+https://github.com/rust-lang/crates.io-index)",

 "backtrace 0.3.9 (registry+https://github.com/rust-lang/crates.io-index)",