use tokio_xmpp::{Client, Event, Packet};
use tokio::prelude::future::{self, Either};
use tokio::prelude::stream;
use tokio::prelude::{Future, Stream};

use futures_util::future::Either;

use tokio_xmpp::{AsyncClient, Event, Packet};
use log::{error, info, warn};

    Chatroom {
        muc_id: String,
        message: String,
    },
    // Send ping request to the server to test connection

    Chatroom { muc_id: String, message: String },
    /// Send ping request to the server to test connection

    jid: xmpp_parsers::Jid,

    jid: xmpp_parsers::BareJid,

                conn.process_jid(&self.jid);

        true // ignore

        info!("ping successed");
        true

    }
}
/// data for awaiting iq answer
struct IqWait {
    /// time of sending iq
    sent: Instant,
    /// timeout
    timeout: Duration,
    /// handler for iq processing
    handler: Box<dyn IqHandler>,
}
impl IqWait {
    pub fn new<T: IqHandler + 'static>(timeout_secs: u64, handler: T) -> IqWait {
        IqWait {
            sent: Instant::now(),
            timeout: Duration::from_secs(timeout_secs),
            handler: Box::new(handler),
        }

        xmpp_parsers::Jid,

        xmpp_parsers::BareJid,

    /// ToDo: remove it as it used only for initialization

    /// ToDo: remove it as it is used only for initialization

    /// if self-presence accepted
    /// ToDo: remove it as it is used only for initialization
    self_presence: bool,

    pending_ids: HashMap<String, (Instant, Box<dyn IqHandler>)>,

    pending_ids: HashMap<String, IqWait>,

    client: Client,

    client: AsyncClient,

struct XmppElementProcessor {
    incoming: element_processor::Processor<XmppConnection, bool, xmpp_parsers::Element>,

trait IqRequestHandler {
    fn process(
        self: Box<Self>,
        conn: &mut XmppConnection,
        id: String,
        from: Option<xmpp_parsers::Jid>,
    ) -> xmpp_parsers::iq::Iq;
}
struct IqRequestUnknown {
    element: xmpp_parsers::Element,
    type_: &'static str,
}
impl IqRequestHandler for IqRequestUnknown {
    fn process(
        self: Box<Self>,
        conn: &mut XmppConnection,
        id: String,
        from: Option<xmpp_parsers::Jid>,
    ) -> xmpp_parsers::iq::Iq {
        warn!(
            "Unsupported IQ {} request from {:?}: {}",
            self.type_,
            from,
            String::from(&self.element)
        );
        stanzas::make_iq_unsupported_error(id, conn.account.jid.clone(), from)
    }
}
struct IqSetRoster {}
impl IqRequestHandler for IqSetRoster {
    fn process(
        self: Box<Self>,
        conn: &mut XmppConnection,
        id: String,
        from: Option<xmpp_parsers::Jid>,
    ) -> xmpp_parsers::iq::Iq {
        info!("Got roster push {} from {:?}", id, from);
        stanzas::make_roster_push_answer(id, conn.account.jid.clone(), from)
    }
}
struct IqGetPing {}
impl IqRequestHandler for IqGetPing {
    fn process(
        self: Box<Self>,
        conn: &mut XmppConnection,
        id: String,
        from: Option<xmpp_parsers::Jid>,
    ) -> xmpp_parsers::iq::Iq {
        info!("Got ping {} from {:?}", id, from);
        stanzas::make_pong(id, conn.account.jid.clone(), from)
    }
}
struct IqGetVersion {}
impl IqRequestHandler for IqGetVersion {
    fn process(
        self: Box<Self>,
        conn: &mut XmppConnection,
        id: String,
        from: Option<xmpp_parsers::Jid>,
    ) -> xmpp_parsers::iq::Iq {
        info!("Got version query {} from {:?}", id, from);
        stanzas::make_version(id, conn.account.jid.clone(), from)
    }
}
struct IqGetDiscoInfo {}
impl IqRequestHandler for IqGetDiscoInfo {
    fn process(
        self: Box<Self>,
        conn: &mut XmppConnection,
        id: String,
        from: Option<xmpp_parsers::Jid>,
    ) -> xmpp_parsers::iq::Iq {
        info!("Got disco query {} from {:?}", id, from);
        stanzas::make_disco_info_result(id, conn.account.jid.clone(), from)
    }
}
struct IqGetDiscoItems {}
impl IqRequestHandler for IqGetDiscoItems {
    fn process(
        self: Box<Self>,
        conn: &mut XmppConnection,
        id: String,
        from: Option<xmpp_parsers::Jid>,
    ) -> xmpp_parsers::iq::Iq {
        info!("Got disco items query {} from {:?}", id, from);
        stanzas::make_disco_items_result(id, conn.account.jid.clone(), from)
    }
}
struct IqGetDiscoCommands {}
impl IqRequestHandler for IqGetDiscoCommands {
    fn process(
        self: Box<Self>,
        conn: &mut XmppConnection,
        id: String,
        from: Option<xmpp_parsers::Jid>,
    ) -> xmpp_parsers::iq::Iq {
        info!("Got disco commands query {} from {:?}", id, from);
        stanzas::make_disco_items_commands(id, conn.account.jid.clone(), from)
    }

impl XmppElementProcessor {
    fn new() -> XmppElementProcessor {

lazy_static::lazy_static! {
    static ref INCOMING: element_processor::Processor<XmppConnection, bool, xmpp_parsers::Element> = {

        XmppElementProcessor { incoming }
    }

        incoming.register(&XmppConnection::incoming_presence_processing);
        incoming.register(&XmppConnection::incoming_message_processing);
        incoming
    };
    static ref INCOMING_IQ_SET: element_processor::Processor<XmppConnection, Box<dyn IqRequestHandler>, xmpp_parsers::Element> = {
        let mut iq_set =
            element_processor::Processor::new(&|_conn: &mut XmppConnection, element| {
                Box::new(IqRequestUnknown {
                    element,
                    type_: "set",
                }) as Box<dyn IqRequestHandler>
            });
        iq_set.register(&XmppConnection::incoming_iq_processing_set_roster);
        iq_set
    };
    static ref INCOMING_IQ_GET: element_processor::Processor<XmppConnection, Box<dyn IqRequestHandler>, xmpp_parsers::Element> = {
        let mut iq_get =
            element_processor::Processor::new(&|_conn: &mut XmppConnection, element| {
                Box::new(IqRequestUnknown {
                    element,
                    type_: "get",
                }) as Box<dyn IqRequestHandler>
            });
        iq_get.register(&XmppConnection::incoming_iq_processing_get_ping);
        iq_get.register(&XmppConnection::incoming_iq_processing_get_disco_info);
        iq_get.register(&XmppConnection::incoming_iq_processing_get_disco_items);
        iq_get.register(&XmppConnection::incoming_iq_processing_get_version);
        iq_get
    };

    pub fn connect<F>(
        self,
        stop_future: F,
    ) -> impl Future<Item = XmppConnection, Error = failure::Error>

    pub async fn connect<F>(self, stop_future: F) -> Result<XmppConnection, failure::Error>

        F: future::Future + Clone + 'static,
        <F as hyper::rt::Future>::Error: Into<failure::Error> + Send,

        F: std::future::Future<Output = ()> + Clone + 'static,

        info!("xmpp connection...");

        info!("xmpp check connection...");

            Box::new(future::ok(XmppConnection { account, state }))
                as Box<dyn Future<Item = _, Error = _>>

            Ok(XmppConnection { account, state })

            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");

            loop {
                let pin_stop_future = stop_future.clone();
                tokio::pin!(pin_stop_future);
                let connect = futures_util::future::select(
                    pin_stop_future,
                    Box::pin(async {
                        info!("xmpp initialization...");
                        let mut client = AsyncClient::new_with_jid(
                            account.jid.clone(),
                            account.password.clone(),
                        );
                        client.set_reconnect(false);
                        let stop_future = stop_future.clone();
                        let (connected, stopped) = XmppConnection {
                            state: XmppState {
                                client,
                                data: std::default::Default::default(),
                            },
                            account: account.clone(),
                        }
                        .processing(XmppConnection::online, Box::pin(stop_future), true)
                        .await;
                        let (connected, stopped) = match (connected, stopped) {
                            (Ok(connection), Either::Left(stop_future)) => {
                                connection.initial_roster(stop_future).await
                            }
                            (connected, stopped) => (connected, stopped),
                        };

                            let stop_future2 = stop_future.clone();
                            let stop_future3 = stop_future.clone();
                            let stop_future4 = stop_future.clone();

                        let (connected, stopped) = match (connected, stopped) {
                            (Ok(connection), Either::Left(stop_future)) => {
                                connection.self_presence(stop_future).await
                            }
                            (connected, stopped) => (connected, stopped),
                        };

                            // 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"))

                        match (connected, stopped) {
                            (Ok(connection), Either::Left(stop_future)) => {
                                connection.enter_mucs(stop_future).await
                            }
                            (connected, stopped) => (connected, stopped),

            )

                )
                .await;
                match connect {
                    Either::Left((_, _)) => {
                        break Err(failure::format_err!("Stop XMPP connection"))
                    }
                    Either::Right(((_, Either::Right(_)), _)) => {
                        break Err(failure::format_err!("Stop XMPP connection(2)"))
                    }
                    Either::Right(((result, Either::Left(_)), _)) => match result {
                        Ok(connection) => break Ok(connection),
                        Err(account) => {}
                    },
                }
            }

            Event::Stanza(stanza) => {
                let processors = XmppElementProcessor::new();
                processors.incoming.process(self, stanza.clone())
            }
            Event::Online => true,
            e => {
                warn!("Unexpected event {:?}", e);

            Event::Stanza(stanza) => INCOMING.process(self, stanza.clone()),
            Event::Online { .. } => true,
            Event::Disconnected(error) => {
                error!("Unexpected event {:?}", error);

            }
        }
    }
    /// Process roster push
    /// see RFC 6212 2.1.6. Roster Push
    fn incoming_iq_processing_set_roster(
        &mut self,
        roster: xmpp_parsers::roster::Roster,
    ) -> Box<dyn IqRequestHandler> {
        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));

        Box::new(IqSetRoster {})

        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
            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);

        INCOMING_IQ_SET
            .process(self, element)
            .process(self, id, from)
    }
    /// Process ping request
    /// see XEP-0199: XMPP Ping
    fn incoming_iq_processing_get_ping(
        &mut self,
        _ping: xmpp_parsers::ping::Ping,
    ) -> Box<dyn IqRequestHandler> {
        Box::new(IqGetPing {})
    }
    /// Process disco query
    /// see XEP-0030: Service Discovery
    fn incoming_iq_processing_get_disco_info(
        &mut self,
        disco: xmpp_parsers::disco::DiscoInfoQuery,
    ) -> Box<dyn IqRequestHandler> {
        if let Some(ref node) = disco.node {
            warn!("Unsupported node {}", node);
            Box::new(IqRequestUnknown {
                element: disco.into(),
                type_: "get",
            })
        } else {
            Box::new(IqGetDiscoInfo {})
        }
    }
    /// Process disco items query
    /// see XEP-0030: Service Discovery
    fn incoming_iq_processing_get_disco_items(
        &mut self,
        disco: xmpp_parsers::disco::DiscoItemsQuery,
    ) -> Box<dyn IqRequestHandler> {
        match &disco.node {
            Some(node) if node == "http://jabber.org/protocol/commands" => {
                Box::new(IqGetDiscoCommands {})

            return stanzas::make_roster_push_answer(id, self.state.client.jid.clone(), from);

            Some(node) => {
                warn!("Unsupported node {}", node);
                Box::new(IqRequestUnknown {
                    element: disco.into(),
                    type_: "get",
                })
            }
            None => Box::new(IqGetDiscoItems {}),

        warn!(
            "Unsupported IQ set request from {:?}: {}",
            from,
            String::from(&element)
        );
        stanzas::make_iq_unsupported_error(id, self.state.client.jid.clone(), from)

    }
    /// Process version query
    /// see XEP-0092: Software Version
    fn incoming_iq_processing_get_version(
        &mut self,
        _version: xmpp_parsers::version::VersionQuery,
    ) -> Box<dyn IqRequestHandler> {
        Box::new(IqGetVersion {})

        use std::convert::TryInto;
        if let Some(_ping) = element.clone().try_into().ok() as Option<xmpp_parsers::ping::Ping> {
            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)

        INCOMING_IQ_GET
            .process(self, element)
            .process(self, id, from)

                if let Some((_, handler)) = self.state.data.pending_ids.remove_entry(&iq.id) {
                    return handler.1.error(self, e);

                if let Some((_, data)) = self.state.data.pending_ids.remove_entry(&iq.id) {
                    return data.handler.error(self, e);

                if let Some((_, handler)) = self.state.data.pending_ids.remove_entry(&iq.id) {
                    return handler.1.result(self, opt_element);

                if let Some((_, data)) = self.state.data.pending_ids.remove_entry(&iq.id) {
                    let now = Instant::now();
                    if data.sent + data.timeout > now {
                        info!(
                            "Timeout for {} was {:?} expected {:?}",
                            iq.id,
                            now - data.sent,
                            data.timeout
                        );
                        return data.handler.result(self, opt_element);
                    } else {
                        warn!(
                            "Timeout for {} was {:?} expected {:?}",
                            iq.id,
                            now - data.sent,
                            data.timeout
                        );
                        return data.handler.timeout(self);
                    }

    /// 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,

    fn incoming_presence_processing(&mut self, presence: xmpp_parsers::presence::Presence) -> bool {
        if presence.from.as_ref() == Some(&self.account.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
    }
    /// first data: return Ok(conn) if still connected, Err(account) if disconnected
    /// second data: true if stopped
    pub async fn processing<S, F, O>(
        mut self,
        mut stop_condition: S,

    ) -> impl Future<
        Item = (Self, Result<Either<F, T>, E>),
        Error = (std::rc::Rc<config::Account>, Result<Either<F, T>, E>),
    >

        stop_fatal: bool,
    ) -> (Result<Self, std::rc::Rc<config::Account>>, Either<F, O>)

        F: Future<Item = T, Error = E> + 'static,

        T: 'static,
        E: 'static,

        F: std::future::Future<Output = O> + Unpin,

        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))))

        let account = self.account.clone();
        let mut stop_future = stop_future;
        loop {
            if let Some(send_element) = self.state.data.send_queue.pop_front() {
                info!("Sending {}", String::from(&send_element));
                let stop = {
                    let send_future = self.state.client.send_stanza(send_element);
                    tokio::pin!(send_future);
                    match futures_util::future::select(stop_future, send_future).await {
                        Either::Left((stop, send_future)) => {
                            if stop_fatal {
                                warn!("Sending interrupted!");
                            } else {
                                match send_future.await {
                                    Ok(_) => {
                                        info!("Sent!");

                                })),
                                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)))

                                    Err(e) => {
                                        error!("Send stanza error: {}", e);
                                        break (Err(self.account), Either::Right(stop));

                                })),
                            }),
                    ) 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,
                                        };

                                }
                            }
                            stop
                        }
                        Either::Right((Ok(_), f)) => {
                            info!("Sent!");
                            stop_future = f;
                            continue;
                        }
                        Either::Right((Err(e), f)) => {
                            error!("Send stanza error: {}", e);
                            break (Err(self.account), Either::Left(f));
                        }
                    }
                };
                break (Ok(self), Either::Right(stop));
            } else {
                use futures_util::StreamExt;

                                        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)))
                                    }
                                }
                            }),
                    )

                let recv_future = self.state.client.next();
                tokio::pin!(recv_future);
                match futures_util::future::select(stop_future, recv_future).await {
                    Either::Left((stop, s)) => break (Ok(self), Either::Right(stop)),
                    Either::Right((None, f)) => {
                        break (Err(self.account), Either::Left(f));
                    }
                    Either::Right((Some(event), f)) => {
                        stop_future = f;
                        if (self.xmpp_processing(&event)) {
                            match stop_condition(&mut self, event) {
                                Ok(true) => break (Ok(self), Either::Left(stop_future)),
                                Ok(false) => {}
                                Err(_) => break (Err(self.account), Either::Left(stop_future)),
                            }
                        } else {
                            break (Err(self.account), Either::Left(stop_future));
                        }
                    }

            },
        )

            }
        }

            Event::Online => {

            Event::Online { .. } => {

            _ => {
                error!("Disconnected while online");

            Event::Disconnected(e) => {
                error!("Disconnected while waiting online: {:?}", e);

    fn initial_roster<F, E>(
        self,

    async fn initial_roster<F>(
        mut self,

    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>

    ) -> (Result<Self, std::rc::Rc<config::Account>>, Either<F, ()>)

        F: Future<Error = E> + 'static,
        E: 'static,

        F: std::future::Future<Output = ()> + Unpin,

        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);

        self.state.data.counter += 1;
        let id_init_roster = format!("id_init_roster{}", self.state.data.counter);

        let account2 = account.clone();

        data.pending_ids.insert(

        self.state.data.pending_ids.insert(

            (
                Instant::now() + Duration::from_secs(60),
                Box::new(InitRosterIqHandler {}),
            ),

            IqWait::new(60, 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,

        let stop_future = {
            let send_future = self.state.client.send_stanza(get_roster);
            tokio::pin!(send_future);
            match futures_util::future::select(stop_future, send_future).await {
                Either::Left(((), _)) => None,
                Either::Right((Ok(_), stop_future)) => Some(stop_future),
                Either::Right((Err(e), f)) => {
                    error!("Send initial roster stanza error: {}", e);
                    return (Err(self.account), Either::Left(f));

                .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),
                })
            })

            }
        };
        if let Some(stop_future) = stop_future {
            self.processing(
                move |conn, _| Ok(conn.state.data.roster_init),
                stop_future,
                true,
            )
            .await
        } else {
            (Ok(self), Either::Right(()))
        }

    fn self_presence<F, E>(
        self,

    async fn self_presence<F>(
        mut self,

    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>

    ) -> (Result<Self, std::rc::Rc<config::Account>>, Either<F, ()>)

        F: Future<Error = E> + 'static,
        E: Into<failure::Error> + 'static,

        F: std::future::Future<Output = ()> + Unpin,

        let XmppConnection {
            account,
            state: XmppState { client, mut data },
        } = self;
        use tokio::prelude::Sink;

        self.state.data.counter += 1;

        data.counter += 1;
        let id_presence = format!("id_init_presence{}", data.counter);
        let presence = stanzas::make_presence(id_presence, xmpp_parsers::presence::Show::None, "Online!".to_string());
        let account2 = account.clone();

        let id_presence = format!("id_init_presence{}", self.state.data.counter);
        let presence = stanzas::make_presence(id_presence, None, "Online!".to_string());

        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,

        let stop_future = {
            let send_future = self.state.client.send_stanza(presence);
            tokio::pin!(send_future);
            match futures_util::future::select(stop_future, send_future).await {
                Either::Left(((), send_future)) => None,
                Either::Right((Ok(_), stop_future)) => Some(stop_future),
                Either::Right((Err(e), f)) => {
                    error!("Send self-presence stanza error: {}", e);
                    return (Err(self.account), Either::Left(f));

                .processing(
                    move |conn, event| {
                        if let Event::Stanza(s) = event {
                            use std::convert::TryInto;
                            match s.try_into() as Result<xmpp_parsers::presence::Presence, _> {
                                Ok(presence) => {
                                    Ok(presence.from.as_ref() == Some(&conn.state.client.jid))
                                }
                                Err(e) => {
                                    warn!("Not a self-presence: {}", e);
                                    Ok(false)
                                }
                            }
                        } else {
                            error!("Wrong event while waiting self-presence");
                            Err(())
                        }
                    },
                    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);
                }

        };
        if let Some(stop_future) = stop_future {
            self.processing(
                move |conn, _| Ok(conn.state.data.self_presence),
                stop_future,
                true,
            )
            .await
        } else {
            (Ok(self), Either::Right(()))

    pub fn process_command(&mut self, cmd: XmppCommand) {
        info!("Got command");

    pub fn process_command(mut self, cmd: XmppCommand) -> MaybeXmppConnection {

            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::ChatroomPresence { muc_id, show, message } => {
                if let Some(muc) = self.state.data.mucs.get(&muc_id) {
                    self.state.data.counter += 1;
                    let id_presence = format!("id_presence{}", self.state.data.counter);
                    self.state
                        .data
                        .send_queue
                        .push_back(stanzas::make_muc_presence(
                            &id_presence,
                            self.account.jid.clone(),
                            muc.clone(),
                            show,
                            Some(message),
                        ));
                } else {
                    error!("Not found MUC {}", muc_id);
                }
            }

                self.state
                    .data
                    .send_queue
                    .push_back(stanzas::make_presence(id_presence, show, message));

                self.state.data.send_queue.push_back(stanzas::make_presence(
                    id_presence,
                    Some(show),
                    message,
                ));

                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 {}),
                    ),

                let ping = stanzas::make_ping(
                    &id_ping,
                    self.state
                        .client
                        .bound_jid()
                        .map_or_else(|| self.account.jid.clone(), |j| j.clone()),

                self.state.data.send_queue.push_back(ping);
                self.state
                    .data
                    .pending_ids
                    .insert(id_ping, IqWait::new(30, PingIqHandler {}));

                    .filter_map(|(id, (timeout, _))| {
                        if now >= *timeout {

                    .filter_map(|(id, data)| {
                        if now >= data.sent + data.timeout {
                            warn!(
                                "Timeout for {} was {:?} expected {:?}",
                                id,
                                now - data.sent,
                                data.timeout
                            );

                    if let Some((_, handler)) = self.state.data.pending_ids.remove(&id) {
                        correct &= handler.timeout(&mut self);

                    if let Some(data) = self.state.data.pending_ids.remove(&id) {
                        correct &= data.handler.timeout(&mut self);

            _ => todo!(),

        self.into()

    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>>

    async fn enter_mucs<F>(
        mut self,
        mut stop_future: F,
    ) -> (Result<Self, std::rc::Rc<config::Account>>, Either<F, ()>)

        F: Future<Error = E> + 'static,
        E: Into<failure::Error> + 'static,

        F: std::future::Future<Output = ()> + Unpin,

        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(),
                    xmpp_parsers::presence::Show::None,
                    None,
                );
                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,
            })

        for chatroom in &self.account.chatrooms {
            self.state.data.counter += 1;
            let id_muc_presence = format!("id_muc_presence{}", self.state.data.counter);
            let muc_presence = stanzas::make_muc_presence(
                &id_muc_presence,
                self.account.jid.clone(),
                chatroom.1.clone(),
                None,
                None,
            );
            info!("Sending muc presence... {}", String::from(&muc_presence));
            let opt_stop_future = {
                let send_future = self.state.client.send_stanza(muc_presence);
                tokio::pin!(send_future);
                match futures_util::future::select(stop_future, send_future).await {
                    Either::Left(((), _)) => None,
                    Either::Right((Ok(_), stop_future)) => Some(stop_future),
                    Either::Right((Err(e), f)) => {
                        error!("Send self-presence stanza error: {}", e);
                        return (Err(self.account), Either::Left(f));
                    }
                }
            };
            if let Some(f) = opt_stop_future {
                stop_future = f;
            } else {
                return (Ok(self), Either::Right(()));
            }
        }
        (Ok(self), Either::Left(stop_future))

use xmpp_parsers::disco::{DiscoInfoResult, DiscoItemsResult, Feature, Identity};

use xmpp_parsers::version::VersionResult;
use xmpp_parsers::{BareJid, Jid};

pub fn make_presence(id: String, show: PresenceShow, text: String) -> Element {

pub fn make_presence(id: String, show: Option<PresenceShow>, text: String) -> Element {

    presence
        .statuses
        .insert(String::from("en"), text);

    presence.statuses.insert(String::from("en"), text);

pub fn make_add_roster(id: &str, jid: xmpp_parsers::Jid) -> Element {

pub fn make_add_roster(id: &str, jid: xmpp_parsers::BareJid) -> Element {

pub fn make_ask_subscribe(jid: xmpp_parsers::Jid) -> Element {

pub fn make_ask_subscribe(id: String, jid: xmpp_parsers::Jid) -> Element {

    presence.id = Some(id);

pub fn make_allow_subscribe(jid: xmpp_parsers::Jid) -> Element {

pub fn make_allow_subscribe(id: String, jid: xmpp_parsers::Jid) -> Element {

    presence.id = Some(id);

    show: PresenceShow,

    show: Option<PresenceShow>,

    ) -> Element {

) -> Element {

        presence
            .statuses
            .insert(String::from("en"), text);

        presence.statuses.insert(String::from("en"), text);

    let mut message = Message::new(Some(to.into_bare_jid()));

    let mut message = Message::new(Some(Jid::Bare(to.into())));

    let mut ping = Iq::from_get(id, Ping);
    ping.to = Some(from.clone().into_domain_jid());
    ping.from = Some(from);

    let ping = Iq::from_get(id, Ping)
        .with_to(Jid::Bare(BareJid::domain(from.clone().domain())))
        .with_from(from);

pub fn make_pong(id: &str, from: xmpp_parsers::Jid, to: Option<xmpp_parsers::Jid>) -> Iq {

pub fn make_pong(id: String, from: xmpp_parsers::Jid, to: Option<xmpp_parsers::Jid>) -> Iq {

pub fn make_version(id: String, from: xmpp_parsers::Jid, to: Option<xmpp_parsers::Jid>) -> Iq {
    let mut version = Iq::from_result(
        id,
        Some(VersionResult {
            name: "SendXmppDRust".to_string(),
            version: "0.1.0".to_string(),
            os: None,
        }),
    );
    version.from = Some(from);
    version.to = to;
    version
}

pub fn make_disco_info_result(
    id: String,
    from: xmpp_parsers::Jid,
    to: Option<xmpp_parsers::Jid>,
) -> Iq {
    let mut result = Iq::from_result(
        id,
        Some(DiscoInfoResult {
            node: None,
            identities: vec![Identity {
                category: "client".to_string(),
                type_: "bot".to_string(),
                name: None,
                lang: None,
            }],
            features: vec![
                Feature {
                    var: "http://jabber.org/protocol/disco#info".to_string(),
                },
                Feature {
                    var: "urn:xmpp:ping".to_string(),
                },
                Feature {
                    var: "http://jabber.org/protocol/commands".to_string(),
                },
            ],
            extensions: vec![],
        }),
    );
    result.from = Some(from);
    result.to = to;
    result
}
pub fn make_disco_items_result(
    id: String,
    from: xmpp_parsers::Jid,
    to: Option<xmpp_parsers::Jid>,
) -> Iq {
    let mut result = Iq::from_result(
        id,
        Some(DiscoItemsResult {
            node: None,
            items: vec![],
        }),
    );
    result.from = Some(from);
    result.to = to;
    result
}
pub fn make_disco_items_commands(
    id: String,
    from: xmpp_parsers::Jid,
    to: Option<xmpp_parsers::Jid>,
) -> Iq {
    let mut result = Iq::from_result(
        id,
        Some(DiscoItemsResult {
            node: Some("http://jabber.org/protocol/commands".to_string()),
            items: vec![],
        }),
    );
    result.from = Some(from);
    result.to = to;
    result
}

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

use futures_util::future::Either;
use log::{error, info, warn};

pub fn xmpp_process<F, S>(

pub async fn xmpp_process<F, S>(

    cmd_recv: S,

    mut cmd_recv: S,

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

) -> Result<(), failure::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: std::future::Future<Output = ()> + Clone + 'static,
    S: tokio::stream::Stream<Item = XmppCommand> + Unpin,

    let conn = account.into();

    use futures_util::StreamExt;

    future::loop_fn(XmppProcessState::new(cmd_recv, signal, conn), |s| {
        let XmppProcessState {
            cmd_recv,
            signal,
            conn,
        } = s;
        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);

    let mut maybe_conn: MaybeXmppConnection = account.into();
    let mut recv_future = cmd_recv.next();

                                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(()))

    loop {
        match maybe_conn.connect(signal.clone()).await {
            Ok(conn) => match conn.processing(|_, _| Ok(false), recv_future, false).await {
                (Ok(conn), Either::Left(f)) => {
                    maybe_conn = conn.into();
                    recv_future = f;
                }
                (Ok(conn), Either::Right(None)) => break Ok(()),
                (Ok(conn), Either::Right(Some(cmd))) => {
                    info!("Get command: {:?}", cmd);
                    maybe_conn = conn.process_command(cmd);
                    recv_future = cmd_recv.next();
                }
                (Err(acc), Either::Left(f)) => {
                    maybe_conn = acc.into();
                    recv_future = f;
                }
                (Err(acc), Either::Right(None)) => break Ok(()),
                (Err(acc), Either::Right(Some(cmd))) => {
                    warn!("Get command for broken connection: {:?}", cmd);
                    maybe_conn = acc.into();
                    recv_future = cmd_recv.next();
                }
            },
            Err(e) => break Err(e),

    })

    }

type Func<S, T, E> = dyn Fn(&mut S, E) -> T;
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 + 'static,
    {
        Processor {
            processors: vec![],
            default: f,
        }
    }
    pub fn register<F, A>(&mut self, f: &'static F)
    where
        F: Fn(&mut S, A) -> T + '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;

use std::future::Future;
use std::pin::Pin;
use tokio::stream::Stream;
use tokio::sync::mpsc::Receiver;

use tokio::prelude::{Async, Future, Poll, Stream};
use tokio_channel::mpsc::Receiver;

use pin_project::pin_project;

pub struct StoppableReceiver<T, F>
where
    F: Future,
{

#[pin_project(project = StoppableReceiverProj)]
pub struct StoppableReceiver<T, F: Future<Output = ()>> {
    #[pin]

    #[pin]

impl<T, F> Stream for StoppableReceiver<T, F>
where
    F: Future,
    T: Debug,
{

impl<T, F: Future<Output = ()>> Stream for StoppableReceiver<T, F> {

    type Error = ();

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        if !self.finished {
            match self.stop_future.poll() {
                Err(_) => {
                    self.receiver.close();
                    self.finished = true;
                }
                Ok(Async::Ready(_)) => {
                    self.receiver.close();
                    self.finished = true;
                }
                Ok(Async::NotReady) => {}
            }

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut std::task::Context,
    ) -> std::task::Poll<Option<Self::Item>> {
        let StoppableReceiverProj {
            mut receiver,
            stop_future,
            finished,
        } = self.project();
        if *finished {
            return std::task::Poll::Ready(None);
        }
        if let std::task::Poll::Ready(_) = stop_future.poll(cx) {
            *finished = true;
            receiver.close();
            return std::task::Poll::Ready(None);

        self.receiver.poll()

        receiver.poll_recv(cx)

pub fn stop_receiver<T, F>(receiver: Receiver<T>, stop_future: F) -> StoppableReceiver<T, F>
where
    F: Future,
    T: Debug,
{

pub fn stop_receiver<T, F: Future<Output = ()>>(
    receiver: Receiver<T>,
    stop_future: F,
) -> StoppableReceiver<T, F> {

extern crate clap;
extern crate hyper;
#[macro_use]
extern crate log;
extern crate env_logger;
#[macro_use]
extern crate serde_derive;
extern crate tokio;
extern crate tokio_channel;
extern crate tokio_signal;
extern crate tokio_xmpp;
#[macro_use]
extern crate failure;
extern crate toml;
extern crate xmpp_parsers;

use tokio::runtime::current_thread;
use tokio::runtime::Runtime;

use log::{error, info, warn};

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

fn body_to_string(req: Request<Body>) -> impl Future<Item = String, Error = failure::Error> {
    req.into_body()

/// Return future with body converted to String
async fn body_to_string(req: Request<Body>) -> Result<String, failure::Error> {
    let whole_body = hyper::body::to_bytes(req.into_body()).await?;
    std::str::from_utf8(&whole_body)
        .map(std::string::ToString::to_string)

        .fold(String::new(), |mut acc, ch| {
            std::str::from_utf8(&*ch).map(|s| {
                acc.push_str(s);
                acc
            })
        })

    cmd_send: tokio_channel::mpsc::Sender<XmppCommand>,

    cmd_send: tokio::sync::mpsc::Sender<XmppCommand>,

impl hyper::service::Service for ServiceCmd {
    type ReqBody = Body;
    type ResBody = Body;

impl hyper::service::Service<Request<Body>> for ServiceCmd {
    type Response = Response<Body>;

    type Future =
        Box<dyn Future<Item = Response<Self::ResBody>, Error = Self::Error> + Send + 'static>;

    type Future = std::pin::Pin<
        Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
    >;
    fn poll_ready(
        &mut self,
        _: &mut std::task::Context,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        std::task::Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: Request<Self::ReqBody>) -> Self::Future {

    fn call(&mut self, req: Request<Body>) -> Self::Future {

            || Err(format_err!("No X-XMPP-To header")),

            || Err(failure::format_err!("No X-XMPP-To header")),

        let xmpp_pres_opt = req
            .headers()
            .get("X-XMPP-Presence");

        let xmpp_pres_opt = req.headers().get("X-XMPP-Presence");

            .map_or_else(|| Err(format_err!("No X-XMPP-Presence header")),
            |show| {
                std::str::from_utf8(show.as_bytes())
                    .map_err(std::convert::Into::into)
                    .and_then(|s| std::str::FromStr::from_str(s)
                        .map_err(|e| format_err!("Incorrect presence {}", e) ))
            },
        );

            .map_or_else(
                || Err(failure::format_err!("No X-XMPP-Presence header")),
                |show| {
                    std::str::from_utf8(show.as_bytes())
                        .map_err(std::convert::Into::into)
                        .and_then(|s| {
                            std::str::FromStr::from_str(s)
                                .map_err(|e| failure::format_err!("Incorrect presence {}", e))
                        })
                },
            );

                Box::new(tokio::prelude::future::result(

                Box::pin(async move {

                            "Unknown destination: {}\nUnknown destination: {}",
                            err,
                            err2,

                            "Unknown destination: {}\nUnknown destination2: {}",
                            err, err2,

                        .map_err(std::convert::Into::into),
                )) as Box<dyn Future<Item = _, Error = _> + Send + 'static>

                        .map_err(std::convert::Into::into)
                })

                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::Presence { show, 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>
            }
            (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)), _, Ok(show)) => {
                info!("Got MUC presence 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::ChatroomPresence { muc_id, show, 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>

                let mut cmd_send = self.cmd_send.clone();
                Box::pin(async move {
                    match body_to_string(req).await {
                        Ok(message) => {
                            match cmd_send.send(XmppCommand::Presence { show, message }).await {
                                Ok(_) => Response::builder()
                                    .body(Body::from(format!("Accepted")))
                                    .map_err(std::convert::Into::into),
                                Err(e) => {
                                    error!("Cann't send presence command: {}", e);
                                    Response::builder()
                                        .status(hyper::StatusCode::BAD_REQUEST)
                                        .body(Body::from(format!("Cann't get presence text")))
                                        .map_err(std::convert::Into::into)
                                }
                            }
                        }
                        Err(e) => Response::builder()
                            .status(hyper::StatusCode::BAD_REQUEST)
                            .body(Body::from(format!("Cann't get presence text")))
                            .map_err(std::convert::Into::into),

                })) 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>
            }

            (None, Ok(xmpp_to), _) => Box::pin(async {
                Response::builder()
                    .status(hyper::StatusCode::BAD_REQUEST)
                    .body(Body::from(format!("SOON")))
                    .map_err(std::convert::Into::into)
            }),
            (Some(Ok(muc_id)), _, Ok(show)) => Box::pin(async {
                Response::builder()
                    .status(hyper::StatusCode::BAD_REQUEST)
                    .body(Body::from(format!("SOON")))
                    .map_err(std::convert::Into::into)
            }),
            (Some(Ok(muc_id)), _, _) => Box::pin(async {
                Response::builder()
                    .status(hyper::StatusCode::BAD_REQUEST)
                    .body(Body::from(format!("SOON")))
                    .map_err(std::convert::Into::into)
            }),
            (Some(Err(err)), _, _) => Box::pin(async {
                Response::builder()
                    .status(hyper::StatusCode::BAD_REQUEST)
                    .body(Body::from(format!("SOON")))
                    .map_err(std::convert::Into::into)
            }),

    cmd_send: tokio_channel::mpsc::Sender<XmppCommand>,

    cmd_send: tokio::sync::mpsc::Sender<XmppCommand>,

impl<Ctx> hyper::service::MakeService<Ctx> for MakeServiceCmd {
    type ReqBody = Body;
    type ResBody = Body;

impl<T> hyper::service::Service<T> for MakeServiceCmd {
    type Response = ServiceCmd;

    type Service = ServiceCmd;
    type Future = tokio::prelude::future::FutureResult<ServiceCmd, Self::MakeError>;
    type MakeError = hyper::http::Error;

    type Future = std::pin::Pin<
        Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
    >;
    fn poll_ready(
        &mut self,
        _: &mut std::task::Context,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        std::task::Poll::Ready(Ok(()))
    }

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

    fn call(&mut self, _: T) -> Self::Future {
        let cmd_send = self.cmd_send.clone();
        let fut = async move { Ok(ServiceCmd { cmd_send }) };
        Box::pin(fut)

fn main() {

/// Entry point of daemon
#[tokio::main]
pub async fn main() -> Result<(), failure::Error> {

    use hyper::rt::{Future, Stream};
    let ctrl_c = tokio_signal::ctrl_c()
        .flatten_stream()
        .into_future()
        .map(|_| ())
        .map_err(|e| {
            error!("Cann't get CTRL+C signal: {}", e.0);
            e.0
        })
        .shared();

    use futures_util::future::FutureExt;
    let ctrl_c = async {
        tokio::signal::ctrl_c().await;
    }
    .shared();

    let (cmd_send, cmd_recv) = tokio_channel::mpsc::channel(10);

    let (mut cmd_send, cmd_recv) = tokio::sync::mpsc::channel(10);

        .with_graceful_shutdown(ctrl_c.clone().map(|_| ()))
        .map_err(|e| error!("server error: {}", e));

        .with_graceful_shutdown(ctrl_c.clone());

    let mut rt = Runtime::new().expect("Cann't start tokio");

        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 mut ping = tokio::time::interval(std::time::Duration::from_secs(ping));
        let mut ctrl_c = ctrl_c.clone();
        tokio::spawn(async move {
            loop {
                let tick = ping.tick();
                tokio::pin!(tick);
                match futures_util::future::select(ctrl_c, tick).await {
                    futures_util::future::Either::Left((_, _)) => break,
                    futures_util::future::Either::Right((_, ctrl_c__)) => {
                        ctrl_c = ctrl_c__;
                        info!("Tick");
                        cmd_send.clone().send(XmppCommand::Ping).await;
                        cmd_send.send(XmppCommand::TimeoutCleanup).await;
                    }
                }
            }
        });

    rt.spawn(http_server);
    let xmpp_join = std::thread::spawn(move || -> Result<(), failure::Error> {
        let recv = stop_receiver(cmd_recv, ctrl_c.clone().map(|_| ()));
        // Launch single-threaded runtime
        let mut ctrt = current_thread::Runtime::new()?;
        let result = ctrt.block_on(xmpp_process(ctrl_c.clone(), recv, config.account));
        info!("Stopping xmpp thread");
        result
    });

    let xmpp_join = std::thread::Builder::new()
        .name("XMPP Thread".into())
        .spawn(move || -> Result<(), failure::Error> {
            let recv = stop_receiver(cmd_recv, ctrl_c.clone());
            // Launch single-threaded runtime
            let mut ctrt = tokio::runtime::Builder::new()
                .basic_scheduler()
                .enable_time()
                .enable_io()
                .build()?;
            let result = ctrt.block_on(xmpp_process(ctrl_c.clone(), recv, config.account));
            info!("Stopping xmpp thread");
            result
        })
        .expect("XMPP Thread");

    rt.shutdown_on_idle().wait().expect("Shutdown error");

    http_server.await;

    xmpp_join
        .join()
        .expect("Join xmpp thread")
        .expect("Result xmpp thread");

    xmpp_join.join().unwrap();
    Ok(())

use serde_derive::Deserialize;

### XMPP client daemon

# XMPP client daemon
## XMPP
### XMPP Core
Ensures granted subscription before delivering messages
### XEPs
#### XEP-0030: Service Discovery
Answers to `http://jabber.org/protocol/disco#info` query.
Answers to `http://jabber.org/protocol/disco#items` query.
#### XEP-0045: Multi-User Chat
Enters to all MUC in `account.chatrooms` at start.
#### XEP-0050: Ad-Hoc Commands
Answers empty commands list. (Awaiting for implementation in xmpp_parser).
#### XEP-0092: Software Version
Answers version.
#### XEP-0199: XMPP Ping
Sends ping each `account.ping` seconds to XMPP server.
Answers to incoming pings.
#### XEP-0203: Delayed Delivery
Ignores delayed messages.

#### Sending messages

## HTTP API

### Sending messages

#### Sending messages to MUC

### Sending messages to MUC

tokio = "0.1"
tokio-channel = "0.1"
tokio-signal = "0.2"
tokio-xmpp = "1.0.0"
failure = "0.1.5"
hyper = "0.12"

futures-util = "0.3.5"
tokio-xmpp = { git = "https://gitlab.com/xmpp-rs/xmpp-rs" }
failure = "0.1.8"
hyper = "0.13"

clap = "2.32"

clap = "2.33"

env_logger = "0.6"
xmpp-parsers = "0.13"
minidom = "=0.10.0" # xmpp-parsers

env_logger = "0.7"
xmpp-parsers = "0.17"
minidom = "=0.12.0" # xmpp-parsers
lazy_static = "1.4.0" # ToDo: remove after const fn will be powerfull enough
pin-project = "0.4"
[dependencies.tokio]
version = "0.2"
features = ["signal", "sync", "rt-threaded"]
[patch.crates-io]
xmpp-parsers = { git = "https://gitlab.com/xmpp-rs/xmpp-rs" }
minidom = { git = "https://gitlab.com/xmpp-rs/xmpp-rs" }

name = "MacTypes-sys"
version = "2.1.0"

name = "addr2line"
version = "0.13.0"

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

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

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

version = "0.7.3"

version = "0.7.13"

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

 "memchr 2.3.3 (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.9 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.3.10"

version = "0.4.7"

name = "arrayvec"
version = "0.4.10"

name = "async-trait"
version = "0.1.40"

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

 "proc-macro2 1.0.20 (registry+https://github.com/rust-lang/crates.io-index)",
 "quote 1.0.7 (registry+https://github.com/rust-lang/crates.io-index)",
 "syn 1.0.40 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.2.11"

version = "0.2.14"

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

 "hermit-abi 0.1.15 (registry+https://github.com/rust-lang/crates.io-index)",
 "libc 0.2.76 (registry+https://github.com/rust-lang/crates.io-index)",
 "winapi 0.3.9 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.1.2"

version = "1.0.1"

version = "0.3.15"

version = "0.3.50"

 "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)",

 "addr2line 0.13.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "cfg-if 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)",
 "libc 0.2.76 (registry+https://github.com/rust-lang/crates.io-index)",
 "miniz_oxide 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
 "object 0.20.0 (registry+https://github.com/rust-lang/crates.io-index)",
 "rustc-demangle 0.1.16 (registry+https://github.com/rust-lang/crates.io-index)",

name = "backtrace-sys"
version = "0.1.28"

name = "base64"
version = "0.10.1"

 "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)",

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

version = "0.10.1"

version = "0.12.3"

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

version = "1.0.4"

version = "1.2.1"

version = "0.8.0"

version = "0.8.1"

 "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)",

 "digest 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)",
 "opaque-debug 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)",

version = "0.7.0"

version = "0.7.3"

 "block-padding 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",

 "block-padding 0.1.5 (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)",

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

version = "0.1.3"

version = "0.1.5"

version = "1.3.1"

version = "1.3.4"

version = "0.4.12"

version = "0.5.6"

dependencies = [
 "byteorder 1.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
 "iovec 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
]

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