O01eg/sendxmppd-rs - Change BWDUANCV77MCLCYMRS2UNFIFW4ZC3KB2KEGEUI77FRO7KW6TUZJQC

Second part of processing result is only about stop_future

Created by O01eg on December 31, 2018

BWDUANCV77MCLCYMRS2UNFIFW4ZC3KB2KEGEUI77FRO7KW6TUZJQC

Dependencies

In channels

main

Change contents

Insertion in src/xmpp/stanzas.rs at line 2 [3.13]
[3.57]
[3.57]
```
use xmpp_parsers::iq::Iq;
```
Insertion in src/xmpp/stanzas.rs at line 4 [3.13]
[3.141]
[3.141]
```
use xmpp_parsers::roster::Roster;
```

Insertion in src/xmpp/stanzas.rs at line 14 [3.13]

[3.419]


pub fn make_get_roster(id: &str) -> Element {
    let mut get_roster = Iq::from_get(Roster {
        ver: None,
        items: vec![],
    });
    get_roster.id = Some(id.to_string());
    get_roster.into()
}

Insertion in src/xmpp/mod.rs at line 7 [5.17]
[4.48]
[3.420]
```
use std::collections::HashMap;
```
Insertion in src/xmpp/mod.rs at line 11 [5.17]
[3.434]
[6.320]
```
const ID_GET_ROSTER: &str = "id_get_roster0";
```

Replacement in src/xmpp/mod.rs at line 14 [5.17]

∅:D[6.321] → [7.0:33]

pub struct MaybeXmppConnection {

[6.321]

[6.354]

struct XmppState {
    client: Client,
    roster: HashMap<jid::Jid, ()>,
}
struct MaybeXmppConnection {

Replacement in src/xmpp/mod.rs at line 21 [5.17]
∅:D[6.397] → [7.34:61]
```
    inner: Option<Client>,
```
[6.397]
[6.425]
```
    state: Option<XmppState>,
```
Replacement in src/xmpp/mod.rs at line 24 [5.17]
∅:D[8.106] → [7.62:90]
```
pub struct XmppConnection {
```
[8.106]
[9.31]
```
struct XmppConnection {
```
Replacement in src/xmpp/mod.rs at line 26 [5.17]
∅:D[9.74] → [7.91:110]
```
    inner: Client,
```
[9.74]
[6.448]
```
    state: XmppState,
```

Replacement in src/xmpp/mod.rs at line 33 [5.17]

∅:D[6.627] → [7.111:148]

            inner: Some(from.inner),

[6.627]

[6.666]

            state: Some(from.state),

Replacement in src/xmpp/mod.rs at line 42 [5.17]
∅:D[3.527] → [7.149:174]
```
            inner: None,
```
[3.527]
[5.499]
```
            state: None,
```

Replacement in src/xmpp/mod.rs at line 53 [5.17]

∅:D[6.1011] → [10.0:37]

        F: future::Future + 'static,

[6.1011]

[6.1056]

        F: future::Future + Clone + 'static,

Replacement in src/xmpp/mod.rs at line 57 [5.17]

∅:D[5.690] → [7.175:234]

        let MaybeXmppConnection { account, inner } = self;

[5.690]

[5.744]

        let MaybeXmppConnection { account, state } = self;

Replacement in src/xmpp/mod.rs at line 59 [5.17]

∅:D[5.745] → [7.235:340]

        if let Some(inner) = inner {
            Box::new(future::ok(XmppConnection { account, inner }))

[5.745]

[6.1302]

        if let Some(state) = state {
            Box::new(future::ok(XmppConnection { account, state }))

Insertion in src/xmpp/mod.rs at line 65 [5.17]
[6.1405]
[6.1434]
```
                    .clone()
```

Replacement in src/xmpp/mod.rs at line 67 [5.17]

∅:D[6.1464] → [10.38:99]

                        future::loop_fn(account, |account| {

[6.1464]

[6.1530]

                        future::loop_fn(account, move |account| {

Insertion in src/xmpp/mod.rs at line 76 [5.17]

[5.1477]

[6.2164]

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

Replacement in src/xmpp/mod.rs at line 82 [5.17]

∅:D[11.435] → [7.341:400]

                                            inner: client,

[11.435]

[11.494]

                                            state: XmppState {
                                                client,
                                                roster: HashMap::new(),
                                            },

Replacement in src/xmpp/mod.rs at line 88 [5.17]

∅:D[11.589] → [10.100:231]

                                        .online()
                                        .and_then(XmppConnection::self_presence)

[11.589]

[11.720]

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

Replacement in src/xmpp/mod.rs at line 124 [5.17]

∅:D[6.3788] → [10.232:285]

    fn xmpp_processing(&mut self, _event: &Event) {}

[6.3788]

[6.3896]

    fn xmpp_processing(&mut self, event: &Event) {
        info!("Incoming xmpp event: {:?}", event);
    }

Replacement in src/xmpp/mod.rs at line 130 [5.17]

∅:D[6.3983] → [6.3983:4019]

    /// or stop future was resolved

[6.3983]

[6.4019]

    /// or stop future was resolved.
    /// Return item if connection was preserved or error otherwise.
    /// Second part is a state of stop_future

Replacement in src/xmpp/mod.rs at line 143 [5.17]

∅:D[6.4375] → [10.408:453]

        S: FnMut(&mut Self, &Event) -> bool,

[6.4375]

[6.4476]

        S: FnMut(&mut Self, Event) -> Result<bool, ()>,

Replacement in src/xmpp/mod.rs at line 148 [5.17]

∅:D[6.4612] → [7.401:539]

                let XmppConnection { inner, account } = xmpp;
                inner.into_future().select2(stop_future).then(|r| match r {

[6.4612]

[13.534]

                let XmppConnection {
                    state: XmppState { client, roster },
                    account,
                } = xmpp;
                client.into_future().select2(stop_future).then(|r| match r {

Replacement in src/xmpp/mod.rs at line 156 [5.17]

∅:D[13.708] → [7.540:587]

                                inner: client,

[13.708]

[13.777]

                                state: XmppState { client, roster },

Replacement in src/xmpp/mod.rs at line 160 [5.17]

∅:D[13.907] → [10.454:742]

                            if stop_condition(&mut xmpp, &event) {
                                future::ok(future::Loop::Break((xmpp, Ok(Either::A(b)))))
                            } else {
                                future::ok(future::Loop::Continue((xmpp, b, stop_condition)))

[13.907]

[12.3255]

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

Replacement in src/xmpp/mod.rs at line 174 [5.17]

∅:D[13.1548] → [7.588:650]

                        if let Some(inner) = a.into_inner() {

[13.1548]

[13.1611]

                        if let Some(client) = a.into_inner() {

Replacement in src/xmpp/mod.rs at line 176 [5.17]

∅:D[13.1672] → [7.651:718]

                                XmppConnection { inner, account },

[13.1672]

[13.1874]

                                XmppConnection {
                                    state: XmppState { client, roster },
                                    account,
                                },

Replacement in src/xmpp/mod.rs at line 186 [5.17]

∅:D[13.2270] → [7.719:808]

                    Err(Either::A((_e, b))) => future::err((account, Ok(Either::A(b)))),

[13.2270]

                    Err(Either::A((e, b))) => {
                        warn!("XMPP error: {}", e.0);
                        future::err((account, Ok(Either::A(b))))
                    }

Replacement in src/xmpp/mod.rs at line 191 [5.17]

∅:D[13.2318] → [7.809:871]

                        if let Some(inner) = a.into_inner() {

[13.2318]

[13.2381]

                        if let Some(client) = a.into_inner() {

Replacement in src/xmpp/mod.rs at line 193 [5.17]

∅:D[13.2442] → [7.872:939]

                                XmppConnection { inner, account },

[13.2442]

[10.743]

                                XmppConnection {
                                    state: XmppState { client, roster },
                                    account,
                                },

Replacement in src/xmpp/mod.rs at line 209 [5.17]

∅:D[6.6691] → [6.6691:6737]

∅:D[6.6737] → [10.844:1453]

    /// returns error if something went wrong
    fn online(self) -> impl Future<Item = XmppConnection, Error = std::rc::Rc<config::Account>> {
        Box::new(future::loop_fn(
            (self.inner, self.account),
            |(client, account)| {
                client.into_future().then(|r| match r {
                    Ok((event, client)) => match event {
                        Some(Event::Online) => {
                            info!("Online");
                            future::ok(future::Loop::Break(XmppConnection {
                                account,
                                inner: client,
                            }))

[6.6691]

[10.1453]

    /// 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) -> Result<bool, ()> {
        if let Event::Stanza(s) = event {
            use try_from::TryInto;
            match s.try_into() as Result<xmpp_parsers::iq::Iq, _> {
                Ok(iq) => {
                    if let Some(id) = iq.id {
                        if id == ID_GET_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) => {
                                            info!("Got roster:");
                                            for i in roster.items {
                                                info!(" >>> {:?}", i);
                                            }
                                            Ok(true)
                                        }
                                        Err(e) => {
                                            error!("Cann't parse roster: {}", e);
                                            Err(())
                                        }
                                    }
                                }
                                _ => {
                                    error!("Unknown result of roster");
                                    Err(())
                                }
                            }
                        } else {
                            Ok(false)

Replacement in src/xmpp/mod.rs at line 264 [5.17]

∅:D[10.1479] → [10.1479:1829]

∅:D[14.5337] → [15.5558:5584]

∅:D[12.6802] → [15.5558:5584]

∅:D[10.1829] → [15.5558:5584]

∅:D[11.3094] → [15.5558:5584]

∅:D[6.8724] → [15.5558:5584]

∅:D[15.5558] → [15.5558:5584]

∅:D[15.5584] → [10.1830:1996]

                        Some(Event::Stanza(s)) => {
                            info!("xmpp stanza: {:?}", s);
                            future::ok(future::Loop::Continue((client, account)))
                        }
                        _ => {
                            warn!("Disconnected");
                            future::err(account)
                        }
                    },
                    Err((e, _)) => {
                        error!("xmpp receive error: {}", e);
                        future::err(account)

[10.1479]

[6.8819]

                    } else {
                        error!("Iq stanza without id");
                        Err(())

Insertion in src/xmpp/mod.rs at line 268 [5.17]

[6.8841]

[10.1997]

                }
                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>,
    {
        let XmppConnection {
            account,
            state: XmppState { client, roster },
        } = self;
        use tokio::prelude::Sink;
        let get_roster = stanzas::make_get_roster(ID_GET_ROSTER);
        let account2 = account.clone();
        info!("Quering roster... {:?}", get_roster);
        client
            .send(get_roster)
            .map_err(move |e| {
                error!("Error on querying roster: {}", e);
                account2
            })
            .and_then(move |client| {
                XmppConnection {
                    state: XmppState { client, roster },
                    account,
                }
                .processing(XmppConnection::process_initial_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),

Replacement in src/xmpp/mod.rs at line 312 [5.17]
∅:D[10.2016] → [10.2016:2042]
```
            },
        ))
```
[10.2016]
[12.8518]
```
            })
```

Replacement in src/xmpp/mod.rs at line 315 [5.17]

∅:D[12.8525] → [10.2043:2220]

    fn self_presence(self) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>> {
        let XmppConnection { account, inner } = self;
        let client = inner;

[12.8525]

[6.10624]

    fn self_presence<F, E>(
        self,
        stop_future: F,
    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
    where
        F: Future<Error = E>,
        E: Into<failure::Error>,
    {
        let XmppConnection {
            account,
            state: XmppState { client, roster },
        } = self;

Deletion in src/xmpp/mod.rs at line 330 [5.17]
∅:D[6.10716] → [10.2221:2259]
```
        info!("Sending presence...");
```

Replacement in src/xmpp/mod.rs at line 331 [5.17]

∅:D[6.10756] → [10.2260:2485]

        Box::new(
            client
                .send(presence)
                .map_err(|e| {
                    error!("Error on send self-presence: {}", e);
                    "Cann't send self-presence".to_owned()

[6.10756]

[10.2485]

        info!("Sending presence... {:?}", presence);
        client
            .send(presence)
            .map_err(|e| {
                error!("Error on send self-presence: {}", e);
                account2
            })
            .and_then(move |client| {
                XmppConnection {
                    state: XmppState { client, roster },
                    account,
                }
                .processing(
                    move |conn, event| {
                        if let Event::Stanza(s) = event {
                            if s.name() == "presence"
                                && s.attr("from").map_or(false, |f| f == conn.account.jid)
                                && s.attr("to").map_or(false, |f| f == conn.account.jid)
                            {
                                Ok(true)
                            } else {
                                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),

Replacement in src/xmpp/mod.rs at line 368 [5.17]

∅:D[10.2504] → [10.2504:4597]

                .and_then(move |client| {
                    future::loop_fn((account2.clone(), client), |(account, client)| {
                        client
                            .into_future()
                            .map_err(|(e, _)| {
                                error!("Error on reading self-presence: {}", e);
                                "Cann't read self-presence".to_owned()
                            })
                            .and_then(|(event, client)| match event {
                                Some(event) => {
                                    if let tokio_xmpp::Event::Stanza(e) = event {
                                        info!("Get stanza: {:?}", e);
                                        if e.name() == "presence"
                                            && e.attr("from").map_or(false, |f| f == account.jid)
                                            && e.attr("to").map_or(false, |f| f == account.jid)
                                        {
                                            info!("Self presence");
                                            future::ok(future::Loop::Break(client))
                                        } else {
                                            future::ok(future::Loop::Continue((account, client)))
                                        }
                                    } else {
                                        future::err("Got wrong event".to_owned())
                                    }
                                }
                                None => future::err("Got closed stream".to_owned()),
                            })
                    })
                    .map_err(|e| format!("waiting self-presence: {}", e))
                })
                .then(|r| match r {
                    Err(e) => {
                        error!("Self-presence waiting error: {}", e);
                        future::err(account)
                    }
                    Ok(inner) => future::ok(XmppConnection { account, inner }),
                }),
        )

[10.2504]

[6.12130]

})

Replacement in src/xmpp/mod.rs at line 432 [5.17]

∅:D[6.13049] → [2.54:118]

                            future::ok(future::Loop::Break(()))

[6.13049]

[6.13130]

                            future::ok(future::Loop::Break((None, conn.into())))

Replacement in src/xmpp/mod.rs at line 438 [5.17]

∅:D[5.6342] → [10.4664:4726]

                        future::err(format_err!("Cmd error"))

[5.6342]

[5.6572]

                        future::err(e)

Insertion in src/xmpp/mod.rs at line 443 [5.17]

[4.900]

[5.7379]

    .and_then(|(opt_cmd_recv, _conn): (Option<S>, MaybeXmppConnection)| {
        if let Some(cmd_recv) = opt_cmd_recv {
            // process left commands
            info!("Stop accepting commands");
            Box::new(
                cmd_recv
                    .for_each(|_cmd| future::ok(()))
                    .map_err(|_| format_err!("cmd receiver last error")),
            ) as Box<Future<Item = (), Error = failure::Error>>
        } else {
            Box::new(future::ok(()))
        }
    })

Insertion in Cargo.toml at line 23 [16.11995]
[9.7111]
[14.8552]
```
try_from = "=0.2.2" # dependency of xmpp-parsers
```

Insertion in Cargo.lock at line 1110 [16.12201]

[9.44807]

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