O01eg/sendxmppd-rs - Change CCLGGFKRUNXBM6IOGEYJOFULCSVOL6C5PFAGCKEWNXOZIKPWLLWQC

Move out XmppConnection into own file

Created by O01eg on April 14, 2019

CCLGGFKRUNXBM6IOGEYJOFULCSVOL6C5PFAGCKEWNXOZIKPWLLWQC

Dependencies

In channels

main

Change contents

File addition: xmpp_connection.rs (-rw-r--r--)

[3.7]

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 super::XmppCommand;
use super::stanzas;
use super::element_processor;
use crate::config;
#[derive(Default)]
struct XmppData {
    /// known roster data
    roster: HashMap<
        xmpp_parsers::Jid,
        (
            xmpp_parsers::roster::Subscription,
            xmpp_parsers::roster::Ask,
        ),
    >,
    /// ids counter
    counter: usize,
    /// map from id of adding item to roster and jid of item
    pending_add_roster_ids: HashMap<String, xmpp_parsers::Jid>,
    /// 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>,
}
struct XmppState {
    client: Client,
    data: XmppData,
}
pub struct XmppConnection {
    account: std::rc::Rc<config::Account>,
    state: XmppState,
}
struct XmppElementProcessor {
    incoming: element_processor::Processor<XmppConnection, bool, xmpp_parsers::Element>,
}
impl XmppElementProcessor {
    fn new() -> XmppElementProcessor {
        let mut incoming = element_processor::Processor::new(&|_, e| {
            warn!("Unknown stanza {:#?}", e);
            true
        });
        incoming.register(&XmppConnection::incoming_iq_processing);
        XmppElementProcessor { 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) => {
                let processors = XmppElementProcessor::new();
                processors.incoming.process(self, stanza.clone())
            }
            Event::Online => true,
            e => {
                warn!("Unexpected event {:?}", e);
                false
            }
        }
    }
    fn incoming_iq_processing(&mut self, iq: xmpp_parsers::iq::Iq) -> bool {
        use std::convert::TryInto;
        if let Some((_, jid)) = self.state.data.pending_add_roster_ids.remove_entry(&iq.id) {
            if let xmpp_parsers::iq::IqType::Result(None) = iq.payload {
                if self.state.data.roster.contains_key(&jid) {
                    info!("Jid {} updated to roster", jid);
                } else {
                    info!("Jid {} added in roster", jid);
                    self.state.data.roster.insert(
                        jid.clone(),
                        (
                            xmpp_parsers::roster::Subscription::None,
                            xmpp_parsers::roster::Ask::None,
                        ),
                    );
                }
                self.process_jid(&jid);
            } else {
                warn!(
                    "Wrong payload when adding {} to roster: {:?}",
                    jid, iq.payload
                );
            }
        }
        match iq.payload {
            xmpp_parsers::iq::IqType::Set(element) => {
                if let Some(roster) =
                    element.try_into().ok() as Option<xmpp_parsers::roster::Roster>
                {
                    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);
                    }
                }
            }
            xmpp_parsers::iq::IqType::Error(e) => {
                error!("iq error: {:?}", e);
                return false;
            }
            xmpp_parsers::iq::IqType::Get(element) => {
                if let Some(_ping) = element.try_into().ok() as Option<xmpp_parsers::ping::Ping> {
                    let pong = stanzas::make_pong(&iq.id, self.state.client.jid.clone(), iq.from);
                    self.state.data.send_queue.push_back(pong);
                }
            }
            _ => (), // ignore
        }
        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)| {
                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 {:?}", 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: {:?}", s);
                Ok(false)
            }
            _ => {
                error!("Disconnected while online");
                Err(())
            }
        }
    }
    fn process_initial_roster(&mut self, event: Event, id_init_roster: &str) -> Result<bool, ()> {
        if let Event::Stanza(s) = event {
            use std::convert::TryInto;
            match s.try_into() as Result<xmpp_parsers::iq::Iq, _> {
                Ok(iq) => {
                    if iq.id == id_init_roster {
                        match iq.payload {
                            xmpp_parsers::iq::IqType::Error(_e) => {
                                error!("Get error instead of roster");
                                Err(())
                            }
                            xmpp_parsers::iq::IqType::Result(Some(result)) => {
                                match result.try_into() as Result<xmpp_parsers::roster::Roster, _> {
                                    Ok(roster) => {
                                        self.state.data.roster.clear();
                                        info!("Got first roster:");
                                        for i in roster.items {
                                            info!(" >>> {:?}", i);
                                            self.state
                                                .data
                                                .roster
                                                .insert(i.jid, (i.subscription, i.ask));
                                        }
                                        Ok(true)
                                    }
                                    Err(e) => {
                                        error!("Cann't parse roster: {}", e);
                                        Err(())
                                    }
                                }
                            }
                            _ => {
                                error!("Unknown result of roster");
                                Err(())
                            }
                        }
                    } else {
                        Ok(false)
                    }
                }
                Err(_e) => Ok(false),
            }
        } else {
            error!("Wrong event while waiting roster");
            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... {:?}", get_roster);
        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, event| conn.process_initial_roster(event, &id_init_roster),
                    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... {:?}", 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, 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());
                    self.state
                        .data
                        .pending_add_roster_ids
                        .insert(id_add_roster, xmpp_to.clone());
                    info!("Adding jid to roster... {:?}", add_roster);
                    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);
            }
        }
    }
    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... {:?}", 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... {:?}", 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,
            })
    }
}

Insertion in src/xmpp/stanzas.rs at line 47 [4.13]

[2.812]

    presence.to = Some(jid);
    presence.into()
}
pub fn make_allow_subscribe(jid: xmpp_parsers::Jid) -> Element {
    let mut presence = Presence::new(PresenceType::Subscribed);

Replacement in src/xmpp/mod.rs at line 0 [3.17]

∅:D[3.17] → [3.18:62]

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

[3.17]

[3.62]

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

Deletion in src/xmpp/mod.rs at line 2 [3.17]
∅:D[3.90] → [3.90:128]
∅:D[3.163] → [3.163:164]
∅:D[3.164] → [2.2363:2404]
```
use tokio::prelude::{Future, Stream};
use tokio_xmpp::{Client, Event, Packet};
```
Deletion in src/xmpp/mod.rs at line 4 [3.17]
∅:D[5.48] → [6.1192:1193]
∅:D[6.1193] → [2.2405:2448]
∅:D[7.306] → [4.420:434]
∅:D[8.489] → [4.420:434]
∅:D[9.305] → [4.420:434]
∅:D[10.43] → [4.420:434]
∅:D[11.262] → [4.420:434]
∅:D[12.32] → [4.420:434]
∅:D[13.34] → [4.420:434]
∅:D[14.48] → [4.420:434]
∅:D[6.1236] → [4.420:434]
∅:D[15.32] → [4.420:434]
∅:D[16.32] → [4.420:434]
∅:D[5.48] → [4.420:434]
∅:D[2.2448] → [4.420:434]
∅:D[3.210] → [4.420:434]
```
use std::collections::{HashMap, VecDeque};
mod stanzas;
```

Deletion in src/xmpp/mod.rs at line 6 [3.17]

∅:D[11.433] → [9.430:431]

∅:D[9.430] → [9.430:431]

∅:D[9.431] → [6.1237:1274]

∅:D[6.1274] → [2.2449:3031]

∅:D[17.329] → [6.1786:1848]

∅:D[11.490] → [6.1786:1848]

∅:D[18.1104] → [6.1786:1848]

∅:D[19.378] → [6.1786:1848]

∅:D[2.3031] → [6.1786:1848]

∅:D[6.1786] → [6.1786:1848]

∅:D[6.1848] → [11.491:524]

∅:D[11.524] → [2.3032:3121]

∅:D[11.611] → [6.1848:1851]

∅:D[2.3121] → [6.1848:1851]

∅:D[6.1848] → [6.1848:1851]

∅:D[6.1851] → [11.612:679]

∅:D[11.679] → [2.3122:3336]

∅:D[2.3336] → [11.795:847]

∅:D[11.795] → [11.795:847]

∅:D[11.847] → [6.1851:1880]

∅:D[6.1851] → [6.1851:1880]

∅:D[9.460] → [20.354:397]

∅:D[21.392] → [20.354:397]

∅:D[6.1880] → [20.354:397]

∅:D[22.33] → [20.354:397]

∅:D[23.33] → [20.354:397]

∅:D[20.354] → [20.354:397]

∅:D[20.397] → [6.1881:1911]

∅:D[9.491] → [20.425:427]

∅:D[21.423] → [20.425:427]

∅:D[6.1911] → [20.425:427]

∅:D[22.61] → [20.425:427]

∅:D[24.27] → [20.425:427]

∅:D[23.61] → [20.425:427]

∅:D[20.425] → [20.425:427]

∅:D[20.427] → [25.105:106]

∅:D[25.105] → [25.105:106]

∅:D[25.106] → [6.1912:1936]

∅:D[26.101] → [27.31:74]

∅:D[9.516] → [27.31:74]

∅:D[21.448] → [27.31:74]

∅:D[6.1936] → [27.31:74]

∅:D[22.90] → [27.31:74]

∅:D[23.90] → [27.31:74]

∅:D[27.31] → [27.31:74]

∅:D[27.74] → [6.1937:1959]

∅:D[9.539] → [20.448:627]

∅:D[21.471] → [20.448:627]

∅:D[6.1959] → [20.448:627]

∅:D[22.110] → [20.448:627]

∅:D[24.47] → [20.448:627]

∅:D[23.110] → [20.448:627]

∅:D[20.448] → [20.448:627]

∅:D[20.627] → [6.1960:1997]

∅:D[6.1997] → [28.0:19]

∅:D[22.148] → [28.0:19]

∅:D[9.577] → [28.0:19]

∅:D[28.19] → [6.1998:2111]

∅:D[29.172] → [28.132:162]

∅:D[30.113] → [28.132:162]

∅:D[12.158] → [28.132:162]

∅:D[31.677] → [28.132:162]

∅:D[6.2111] → [28.132:162]

∅:D[22.238] → [28.132:162]

∅:D[28.132] → [28.132:162]

∅:D[28.162] → [6.2112:2422]

∅:D[6.2422] → [31.748:764]

∅:D[22.312] → [31.748:764]

∅:D[31.748] → [31.748:764]

∅:D[31.764] → [6.2423:2425]

∅:D[6.2425] → [31.766:767]

∅:D[31.766] → [31.766:767]

∅:D[31.767] → [6.2426:2453]

∅:D[28.431] → [3.516:554]

∅:D[30.430] → [3.516:554]

∅:D[31.994] → [3.516:554]

∅:D[6.2453] → [3.516:554]

∅:D[3.516] → [3.516:554]

∅:D[3.554] → [6.2454:2505]

∅:D[28.483] → [20.876:1011]

∅:D[29.268] → [20.876:1011]

∅:D[30.482] → [20.876:1011]

∅:D[12.254] → [20.876:1011]

∅:D[31.1046] → [20.876:1011]

∅:D[6.2505] → [20.876:1011]

∅:D[22.359] → [20.876:1011]

∅:D[20.876] → [20.876:1011]

∅:D[20.1011] → [9.604:649]

∅:D[9.649] → [20.1056:1132]

∅:D[21.581] → [20.1056:1132]

∅:D[32.37] → [20.1056:1132]

∅:D[24.149] → [20.1056:1132]

∅:D[20.1056] → [20.1056:1132]

∅:D[26.648] → [3.653:690]

∅:D[33.282] → [3.653:690]

∅:D[34.302] → [3.653:690]

∅:D[27.460] → [3.653:690]

∅:D[35.563] → [3.653:690]

∅:D[36.282] → [3.653:690]

∅:D[37.152] → [3.653:690]

∅:D[38.257] → [3.653:690]

∅:D[20.1132] → [3.653:690]

∅:D[39.109] → [3.653:690]

∅:D[25.781] → [3.653:690]

∅:D[3.653] → [3.653:690]

∅:D[3.690] → [6.2506:2565]

∅:D[33.337] → [3.744:745]

∅:D[9.709] → [3.744:745]

∅:D[21.641] → [3.744:745]

∅:D[35.623] → [3.744:745]

∅:D[36.337] → [3.744:745]

∅:D[40.213] → [3.744:745]

∅:D[6.2565] → [3.744:745]

∅:D[22.419] → [3.744:745]

∅:D[38.312] → [3.744:745]

∅:D[24.209] → [3.744:745]

∅:D[20.1193] → [3.744:745]

∅:D[25.841] → [3.744:745]

∅:D[23.234] → [3.744:745]

∅:D[3.744] → [3.744:745]

∅:D[3.745] → [6.2566:2671]

∅:D[6.2671] → [2.3337:3393]

∅:D[7.598] → [3.944:961]

∅:D[26.769] → [3.944:961]

∅:D[41.1197] → [3.944:961]

∅:D[33.500] → [3.944:961]

∅:D[34.355] → [3.944:961]

∅:D[11.900] → [3.944:961]

∅:D[27.581] → [3.944:961]

∅:D[42.1155] → [3.944:961]

∅:D[35.744] → [3.944:961]

∅:D[36.500] → [3.944:961]

∅:D[37.320] → [3.944:961]

∅:D[6.2727] → [3.944:961]

∅:D[22.577] → [3.944:961]

∅:D[38.475] → [3.944:961]

∅:D[20.1354] → [3.944:961]

∅:D[43.216] → [3.944:961]

∅:D[39.162] → [3.944:961]

∅:D[25.962] → [3.944:961]

∅:D[44.216] → [3.944:961]

∅:D[2.3393] → [3.944:961]

∅:D[3.944] → [3.944:961]

∅:D[3.961] → [20.1355:1405]

∅:D[20.1405] → [9.816:845]

∅:D[9.845] → [20.1434:1464]

∅:D[21.777] → [20.1434:1464]

∅:D[20.1434] → [20.1434:1464]

∅:D[20.1464] → [9.846:912]

∅:D[9.912] → [20.1530:1591]

∅:D[21.844] → [20.1530:1591]

∅:D[32.99] → [20.1530:1591]

∅:D[24.377] → [20.1530:1591]

∅:D[20.1530] → [20.1530:1591]

∅:D[20.1591] → [2.3394:3584]

∅:D[26.1391] → [3.1476:1477]

∅:D[41.1388] → [3.1476:1477]

∅:D[33.1016] → [3.1476:1477]

∅:D[45.612] → [3.1476:1477]

∅:D[21.998] → [3.1476:1477]

∅:D[34.1088] → [3.1476:1477]

∅:D[11.1320] → [3.1476:1477]

∅:D[46.349] → [3.1476:1477]

∅:D[42.1346] → [3.1476:1477]

∅:D[36.1016] → [3.1476:1477]

∅:D[6.2918] → [3.1476:1477]

∅:D[20.2010] → [3.1476:1477]

∅:D[39.678] → [3.1476:1477]

∅:D[25.1618] → [3.1476:1477]

∅:D[44.636] → [3.1476:1477]

∅:D[2.3584] → [3.1476:1477]

∅:D[3.1476] → [3.1476:1477]

∅:D[3.1477] → [2.3585:3789]

∅:D[41.1593] → [9.1065:1066]

∅:D[45.765] → [9.1065:1066]

∅:D[11.1473] → [9.1065:1066]

∅:D[42.1551] → [9.1065:1066]

∅:D[6.3055] → [9.1065:1066]

∅:D[44.789] → [9.1065:1066]

∅:D[2.3789] → [9.1065:1066]

∅:D[9.1065] → [9.1065:1066]

∅:D[9.1066] → [2.3790:4201]

∅:D[41.2005] → [20.3128:3162]

∅:D[45.2431] → [20.3128:3162]

∅:D[11.3139] → [20.3128:3162]

∅:D[42.1963] → [20.3128:3162]

∅:D[6.3467] → [20.3128:3162]

∅:D[44.2455] → [20.3128:3162]

∅:D[2.4201] → [20.3128:3162]

∅:D[20.3128] → [20.3128:3162]

∅:D[20.3162] → [2.4202:5219]

∅:D[41.3023] → [20.3192:3287]

∅:D[45.2462] → [20.3192:3287]

∅:D[11.3170] → [20.3192:3287]

∅:D[42.2981] → [20.3192:3287]

∅:D[6.4405] → [20.3192:3287]

∅:D[44.2486] → [20.3192:3287]

∅:D[2.5219] → [20.3192:3287]

∅:D[20.3192] → [20.3192:3287]

∅:D[33.1167] → [34.2319:2359]

∅:D[20.3287] → [34.2319:2359]

∅:D[34.2319] → [34.2319:2359]

∅:D[34.2359] → [20.3288:3733]

∅:D[26.2883] → [3.1858:1868]

∅:D[33.1359] → [3.1858:1868]

∅:D[34.2804] → [3.1858:1868]

∅:D[36.1359] → [3.1858:1868]

∅:D[20.3733] → [3.1858:1868]

∅:D[39.1137] → [3.1858:1868]

∅:D[25.3372] → [3.1858:1868]

∅:D[3.1858] → [3.1858:1868]

∅:D[3.1868] → [26.2884:2890]

∅:D[26.2890] → [20.3734:3736]

∅:D[20.3736] → [26.2997:2998]

∅:D[25.3375] → [26.2997:2998]

∅:D[26.2997] → [26.2997:2998]

∅:D[26.2998] → [20.3737:3788]

∅:D[20.3788] → [2.5220:5324]

∅:D[47.1289] → [6.4545:4606]

∅:D[41.3163] → [6.4545:4606]

∅:D[48.666] → [6.4545:4606]

∅:D[11.3306] → [6.4545:4606]

∅:D[18.1209] → [6.4545:4606]

∅:D[43.395] → [6.4545:4606]

∅:D[2.5324] → [6.4545:4606]

∅:D[6.4545] → [6.4545:4606]

∅:D[6.4606] → [11.3307:3369]

∅:D[11.3369] → [2.5325:5391]

∅:D[47.4784] → [6.8539:8553]

∅:D[41.6248] → [6.8539:8553]

∅:D[48.3049] → [6.8539:8553]

∅:D[11.3603] → [6.8539:8553]

∅:D[18.2217] → [6.8539:8553]

∅:D[43.1135] → [6.8539:8553]

∅:D[2.5391] → [6.8539:8553]

∅:D[6.8539] → [6.8539:8553]

∅:D[6.8553] → [2.5392:5427]

∅:D[47.4820] → [6.8600:8670]

∅:D[41.6296] → [6.8600:8670]

∅:D[48.3097] → [6.8600:8670]

∅:D[11.3651] → [6.8600:8670]

∅:D[18.2253] → [6.8600:8670]

∅:D[43.1171] → [6.8600:8670]

∅:D[2.5427] → [6.8600:8670]

∅:D[6.8600] → [6.8600:8670]

∅:D[6.8670] → [2.5428:7935]

∅:D[47.4843] → [6.8712:8726]

∅:D[41.6339] → [6.8712:8726]

∅:D[48.3140] → [6.8712:8726]

∅:D[11.3694] → [6.8712:8726]

∅:D[18.2276] → [6.8712:8726]

∅:D[43.1194] → [6.8712:8726]

∅:D[2.7935] → [6.8712:8726]

∅:D[6.8712] → [6.8712:8726]

∅:D[6.8726] → [2.7936:7967]

∅:D[2.7967] → [6.8726:8736]

∅:D[6.8726] → [6.8726:8736]

∅:D[6.8736] → [2.7968:7981]

∅:D[49.2431] → [9.2124:2130]

∅:D[29.871] → [9.2124:2130]

∅:D[12.1026] → [9.2124:2130]

∅:D[31.1227] → [9.2124:2130]

∅:D[6.8736] → [9.2124:2130]

∅:D[22.970] → [9.2124:2130]

∅:D[16.920] → [9.2124:2130]

∅:D[2.7981] → [9.2124:2130]

∅:D[9.2124] → [9.2124:2130]

∅:D[9.2130] → [20.3896:3983]

∅:D[21.1590] → [20.3896:3983]

∅:D[32.285] → [20.3896:3983]

∅:D[24.970] → [20.3896:3983]

∅:D[20.3896] → [20.3896:3983]

∅:D[20.3983] → [6.8737:8888]

∅:D[28.658] → [20.4019:4137]

∅:D[9.2282] → [20.4019:4137]

∅:D[30.709] → [20.4019:4137]

∅:D[12.1178] → [20.4019:4137]

∅:D[6.8888] → [20.4019:4137]

∅:D[22.1007] → [20.4019:4137]

∅:D[16.1072] → [20.4019:4137]

∅:D[20.4019] → [20.4019:4137]

∅:D[20.4137] → [6.8889:9010]

∅:D[28.841] → [20.4319:4335]

∅:D[21.1773] → [20.4319:4335]

∅:D[30.892] → [20.4319:4335]

∅:D[12.1300] → [20.4319:4335]

∅:D[6.9010] → [20.4319:4335]

∅:D[22.1190] → [20.4319:4335]

∅:D[32.407] → [20.4319:4335]

∅:D[24.1092] → [20.4319:4335]

∅:D[16.1194] → [20.4319:4335]

∅:D[20.4319] → [20.4319:4335]

∅:D[20.4335] → [6.9011:9167]

∅:D[28.943] → [20.4476:4612]

∅:D[9.2339] → [20.4476:4612]

∅:D[21.1875] → [20.4476:4612]

∅:D[30.1034] → [20.4476:4612]

∅:D[12.1457] → [20.4476:4612]

∅:D[6.9167] → [20.4476:4612]

∅:D[22.1333] → [20.4476:4612]

∅:D[32.453] → [20.4476:4612]

∅:D[24.1138] → [20.4476:4612]

∅:D[16.1351] → [20.4476:4612]

∅:D[20.4476] → [20.4476:4612]

∅:D[20.4612] → [6.9168:9205]

∅:D[6.9205] → [2.7982:8041]

∅:D[11.3750] → [6.9264:9319]

∅:D[2.8041] → [6.9264:9319]

∅:D[6.9264] → [6.9264:9319]

∅:D[6.9319] → [2.8042:14798]

∅:D[7.7546] → [26.5580:5595]

∅:D[8.6633] → [26.5580:5595]

∅:D[28.3719] → [26.5580:5595]

∅:D[10.6982] → [26.5580:5595]

∅:D[21.4691] → [26.5580:5595]

∅:D[11.6822] → [26.5580:5595]

∅:D[29.3484] → [26.5580:5595]

∅:D[30.3510] → [26.5580:5595]

∅:D[12.3851] → [26.5580:5595]

∅:D[31.4185] → [26.5580:5595]

∅:D[6.16291] → [26.5580:5595]

∅:D[22.3480] → [26.5580:5595]

∅:D[15.2864] → [26.5580:5595]

∅:D[2.14798] → [26.5580:5595]

∅:D[26.5580] → [26.5580:5595]

∅:D[26.5595] → [20.6613:6623]

∅:D[50.3945] → [3.3349:3355]

∅:D[33.2690] → [3.3349:3355]

∅:D[51.1027] → [3.3349:3355]

∅:D[36.2690] → [3.3349:3355]

∅:D[38.1842] → [3.3349:3355]

∅:D[20.6623] → [3.3349:3355]

∅:D[52.1060] → [3.3349:3355]

∅:D[25.6318] → [3.3349:3355]

∅:D[53.1761] → [3.3349:3355]

∅:D[3.3349] → [3.3349:3355]


#[derive(Default)]
struct XmppData {
    /// known roster data
    roster: HashMap<
        xmpp_parsers::Jid,
        (
            xmpp_parsers::roster::Subscription,
            xmpp_parsers::roster::Ask,
        ),
    >,
    /// ids counter
    counter: usize,
    /// map from id of adding item to roster and jid of item
    pending_add_roster_ids: HashMap<String, xmpp_parsers::Jid>,
    /// 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>,
}
struct XmppState {
    client: Client,
    data: XmppData,
}
struct XmppElementProcessor {
    incoming: element_processor::Processor<XmppConnection, bool, xmpp_parsers::Element>,
}
impl XmppElementProcessor {
    fn new() -> XmppElementProcessor {
        let mut incoming = element_processor::Processor::new(&|_, e| {
            warn!("Unknown stanza {:#?}", e);
            true
        });
        incoming.register(&XmppConnection::incoming_iq_processing);
        XmppElementProcessor { incoming }
    }
}
struct MaybeXmppConnection {
    account: std::rc::Rc<config::Account>,
    state: Option<XmppState>,
}
struct XmppConnection {
    account: std::rc::Rc<config::Account>,
    state: 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
    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) => {
                let processors = XmppElementProcessor::new();
                processors.incoming.process(self, stanza.clone())
            }
            Event::Online => true,
            e => {
                warn!("Unexpected event {:?}", e);
                false
            }
        }
    }
    fn incoming_iq_processing(&mut self, iq: xmpp_parsers::iq::Iq) -> bool {
        use std::convert::TryInto;
        if let Some((_, jid)) = self.state.data.pending_add_roster_ids.remove_entry(&iq.id) {
            if let xmpp_parsers::iq::IqType::Result(None) = iq.payload {
                if self.state.data.roster.contains_key(&jid) {
                    info!("Jid {} updated to roster", jid);
                } else {
                    info!("Jid {} added in roster", jid);
                    self.state.data.roster.insert(
                        jid.clone(),
                        (
                            xmpp_parsers::roster::Subscription::None,
                            xmpp_parsers::roster::Ask::None,
                        ),
                    );
                }
                self.process_jid(&jid);
            } else {
                warn!(
                    "Wrong payload when adding {} to roster: {:?}",
                    jid, iq.payload
                );
            }
        }
        match iq.payload {
            xmpp_parsers::iq::IqType::Set(element) => {
                if let Some(roster) =
                    element.try_into().ok() as Option<xmpp_parsers::roster::Roster>
                {
                    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);
                    }
                }
            }
            xmpp_parsers::iq::IqType::Error(e) => {
                error!("iq error: {:?}", e);
                return false;
            }
            xmpp_parsers::iq::IqType::Get(element) => {
                if let Some(_ping) = element.try_into().ok() as Option<xmpp_parsers::ping::Ping> {
                    let pong = stanzas::make_pong(&iq.id, self.state.client.jid.clone(), iq.from);
                    self.state.data.send_queue.push_back(pong);
                }
            }
            _ => (), // ignore
        }
        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
    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)| {
                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 {:?}", 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)))
                                    }
                                }
                            }),
                    )
                }
            },
        )
    }

Replacement in src/xmpp/mod.rs at line 7 [3.17]

∅:D[3.3356] → [20.6624:6691]

∅:D[20.6691] → [6.16292:16429]

∅:D[28.3839] → [9.4011:4339]

∅:D[30.3630] → [9.4011:4339]

∅:D[12.3989] → [9.4011:4339]

∅:D[6.16429] → [9.4011:4339]

∅:D[22.3601] → [9.4011:4339]

∅:D[16.4756] → [9.4011:4339]

∅:D[9.4011] → [9.4011:4339]

∅:D[9.4339] → [6.16430:16484]

    /// 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: {:?}", s);
                Ok(false)
            }
            _ => {
                error!("Disconnected while online");
                Err(())
            }
        }
    }

[3.3356]

[6.16484]

mod stanzas;

Replacement in src/xmpp/mod.rs at line 9 [3.17]

∅:D[6.16485] → [2.14799:14898]

∅:D[11.6900] → [6.16562:16604]

∅:D[19.1320] → [6.16562:16604]

∅:D[2.14898] → [6.16562:16604]

∅:D[6.16562] → [6.16562:16604]

∅:D[6.16604] → [2.14899:14938]

∅:D[47.6453] → [6.16639:16735]

∅:D[11.6936] → [6.16639:16735]

∅:D[2.14938] → [6.16639:16735]

∅:D[6.16639] → [6.16639:16735]

∅:D[6.16735] → [2.14939:15999]

∅:D[47.7514] → [6.18104:18146]

∅:D[48.6606] → [6.18104:18146]

∅:D[11.8330] → [6.18104:18146]

∅:D[54.2869] → [6.18104:18146]

∅:D[2.15999] → [6.18104:18146]

∅:D[6.18104] → [6.18104:18146]

∅:D[6.18146] → [2.16000:16049]

∅:D[47.7564] → [6.18374:18412]

∅:D[48.6835] → [6.18374:18412]

∅:D[54.3131] → [6.18374:18412]

∅:D[2.16049] → [6.18374:18412]

∅:D[6.18374] → [6.18374:18412]

∅:D[6.18412] → [2.16050:16262]

∅:D[47.7777] → [48.6991:7025]

∅:D[2.16262] → [48.6991:7025]

∅:D[48.6991] → [48.6991:7025]

∅:D[48.7025] → [2.16263:16436]

∅:D[11.8520] → [47.7921:7951]

∅:D[2.16436] → [47.7921:7951]

∅:D[47.7921] → [47.7921:7951]

∅:D[47.7951] → [6.18736:18791]

∅:D[48.7097] → [6.18736:18791]

∅:D[11.8592] → [6.18736:18791]

∅:D[6.18736] → [6.18736:18791]

∅:D[6.18791] → [2.16437:16471]

∅:D[47.7986] → [6.18879:18957]

∅:D[48.7186] → [6.18879:18957]

∅:D[11.8681] → [6.18879:18957]

∅:D[54.3340] → [6.18879:18957]

∅:D[2.16471] → [6.18879:18957]

∅:D[6.18879] → [6.18879:18957]

∅:D[28.3902] → [9.4363:4377]

∅:D[30.3693] → [9.4363:4377]

∅:D[12.4014] → [9.4363:4377]

∅:D[6.18957] → [9.4363:4377]

∅:D[22.3664] → [9.4363:4377]

∅:D[16.4781] → [9.4363:4377]

∅:D[9.4363] → [9.4363:4377]

∅:D[9.4377] → [6.18958:19051]

∅:D[6.19051] → [9.4377:4387]

∅:D[9.4377] → [9.4377:4387]

∅:D[9.7793] → [21.8518:8524]

∅:D[32.2042] → [21.8518:8524]

∅:D[21.8518] → [21.8518:8524]

∅:D[21.8524] → [6.19052:19318]

∅:D[6.19318] → [2.16472:16523]

∅:D[11.8729] → [6.19369:19421]

∅:D[2.16523] → [6.19369:19421]

∅:D[6.19369] → [6.19369:19421]

∅:D[6.19421] → [21.8524:8525]

∅:D[21.8524] → [21.8524:8525]

∅:D[21.8525] → [2.16524:16691]

∅:D[11.8796] → [6.19641:19750]

∅:D[2.16691] → [6.19641:19750]

∅:D[6.19641] → [6.19641:19750]

∅:D[6.19750] → [2.16692:16738]

∅:D[41.11465] → [6.19780:20084]

∅:D[11.8827] → [6.19780:20084]

∅:D[2.16738] → [6.19780:20084]

∅:D[6.19780] → [6.19780:20084]

∅:D[6.20084] → [2.16739:16911]

∅:D[11.8909] → [6.20165:20486]

∅:D[19.1638] → [6.20165:20486]

∅:D[2.16911] → [6.20165:20486]

∅:D[6.20165] → [6.20165:20486]

∅:D[6.20486] → [9.7794:7942]

∅:D[21.8525] → [9.7794:7942]

∅:D[9.7942] → [6.20487:20570]

∅:D[30.5010] → [9.8005:8011]

∅:D[12.5460] → [9.8005:8011]

∅:D[6.20570] → [9.8005:8011]

∅:D[22.3728] → [9.8005:8011]

∅:D[16.6431] → [9.8005:8011]

∅:D[9.8005] → [9.8005:8011]

∅:D[9.8011] → [6.20571:20686]

∅:D[28.5180] → [20.10624:10716]

∅:D[9.8128] → [20.10624:10716]

∅:D[21.8796] → [20.10624:10716]

∅:D[30.5065] → [20.10624:10716]

∅:D[12.5578] → [20.10624:10716]

∅:D[6.20686] → [20.10624:10716]

∅:D[22.3811] → [20.10624:10716]

∅:D[32.2220] → [20.10624:10716]

∅:D[24.3485] → [20.10624:10716]

∅:D[15.3174] → [20.10624:10716]

∅:D[16.6651] → [20.10624:10716]

∅:D[20.10624] → [20.10624:10716]

∅:D[22.3850] → [20.10716:10756]

∅:D[32.2259] → [20.10716:10756]

∅:D[24.3524] → [20.10716:10756]

∅:D[20.10716] → [20.10716:10756]

∅:D[20.10756] → [6.20687:20740]

    fn process_initial_roster(&mut self, event: Event, id_init_roster: &str) -> Result<bool, ()> {
        if let Event::Stanza(s) = event {
            use std::convert::TryInto;
            match s.try_into() as Result<xmpp_parsers::iq::Iq, _> {
                Ok(iq) => {
                    if iq.id == id_init_roster {
                        match iq.payload {
                            xmpp_parsers::iq::IqType::Error(_e) => {
                                error!("Get error instead of roster");
                                Err(())
                            }
                            xmpp_parsers::iq::IqType::Result(Some(result)) => {
                                match result.try_into() as Result<xmpp_parsers::roster::Roster, _> {
                                    Ok(roster) => {
                                        self.state.data.roster.clear();
                                        info!("Got first roster:");
                                        for i in roster.items {
                                            info!(" >>> {:?}", i);
                                            self.state
                                                .data
                                                .roster
                                                .insert(i.jid, (i.subscription, i.ask));
                                        }
                                        Ok(true)
                                    }
                                    Err(e) => {
                                        error!("Cann't parse roster: {}", e);
                                        Err(())
                                    }
                                }
                            }
                            _ => {
                                error!("Unknown result of roster");
                                Err(())
                            }
                        }
                    } else {
                        Ok(false)
                    }
                }
                Err(_e) => Ok(false),
            }
        } else {
            error!("Wrong event while waiting roster");
            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... {:?}", get_roster);
        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, event| conn.process_initial_roster(event, &id_init_roster),
                    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... {:?}", presence);

[6.16485]

[6.20740]

mod xmpp_connection;
use xmpp_connection::MaybeXmppConnection;

Deletion in src/xmpp/mod.rs at line 12 [3.17]

∅:D[6.20741] → [16.6652:6667]

∅:D[9.8183] → [16.6652:6667]

∅:D[16.6667] → [2.16912:16956]

∅:D[41.11510] → [9.8226:8315]

∅:D[11.8938] → [9.8226:8315]

∅:D[2.16956] → [9.8226:8315]

∅:D[9.8226] → [9.8226:8315]

∅:D[9.8315] → [6.20742:20767]

∅:D[28.5285] → [9.8340:8426]

∅:D[30.5170] → [9.8340:8426]

∅:D[12.5620] → [9.8340:8426]

∅:D[6.20767] → [9.8340:8426]

∅:D[22.3920] → [9.8340:8426]

∅:D[16.6693] → [9.8340:8426]

∅:D[9.8340] → [9.8340:8426]

∅:D[9.8426] → [6.20768:20823]

∅:D[28.5335] → [9.8483:8658]

∅:D[29.3966] → [9.8483:8658]

∅:D[30.5220] → [9.8483:8658]

∅:D[12.5678] → [9.8483:8658]

∅:D[31.4551] → [9.8483:8658]

∅:D[55.777] → [9.8483:8658]

∅:D[6.20823] → [9.8483:8658]

∅:D[22.3956] → [9.8483:8658]

∅:D[16.6853] → [9.8483:8658]

∅:D[9.8483] → [9.8483:8658]

∅:D[9.8658] → [2.16957:17484]

∅:D[47.8514] → [9.9042:9105]

∅:D[11.9323] → [9.9042:9105]

∅:D[6.21208] → [9.9042:9105]

∅:D[22.4480] → [9.9042:9105]

∅:D[2.17484] → [9.9042:9105]

∅:D[9.9042] → [9.9042:9105]

∅:D[9.9105] → [6.21209:21324]

∅:D[28.5424] → [9.9220:9320]

∅:D[30.5309] → [9.9220:9320]

∅:D[12.5794] → [9.9220:9320]

∅:D[6.21324] → [9.9220:9320]

∅:D[22.4569] → [9.9220:9320]

∅:D[16.6969] → [9.9220:9320]

∅:D[9.9220] → [9.9220:9320]

∅:D[9.9320] → [6.21325:21624]

∅:D[6.21624] → [22.4570:4585]

∅:D[9.9320] → [22.4570:4585]

∅:D[22.4585] → [6.21625:21632]

∅:D[6.21632] → [2.17485:19528]

∅:D[11.9463] → [6.21686:21716]

∅:D[2.19528] → [6.21686:21716]

∅:D[6.21686] → [6.21686:21716]

∅:D[6.21716] → [2.19529:20613]

∅:D[45.5300] → [6.23186:23196]

∅:D[17.755] → [6.23186:23196]

∅:D[11.9665] → [6.23186:23196]

∅:D[56.831] → [6.23186:23196]

∅:D[18.4593] → [6.23186:23196]

∅:D[43.3544] → [6.23186:23196]

∅:D[2.20613] → [6.23186:23196]

∅:D[6.23186] → [6.23186:23196]

∅:D[6.23196] → [2.20614:23149]

∅:D[11.9691] → [18.5042:5048]

∅:D[2.23149] → [18.5042:5048]

∅:D[18.5042] → [18.5042:5048]

∅:D[31.5470] → [3.3789:3792]

∅:D[20.12136] → [3.3789:3792]

∅:D[44.3125] → [3.3789:3792]

∅:D[3.3789] → [3.3789:3792]

        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, 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()));
                    }
                } 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());
                    self.state
                        .data
                        .pending_add_roster_ids
                        .insert(id_add_roster, xmpp_to.clone());
                    info!("Adding jid to roster... {:?}", add_roster);
                    self.state.data.send_queue.push_back(add_roster);
                }
            }
        }
    }
    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);
            }
        }
    }
    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... {:?}", 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... {:?}", 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,
            })
    }
}

Replacement in src/xmpp/element_processor.rs at line 0 [11.11301]

∅:D[11.11301] → [2.24939:26015]

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

[11.11301]

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