O01eg/sendxmppd-rs - Change OANBCLN5TD5VQTSAQXSIXU6IUCXMKQSML2UJGQCKVTOCENEVABIAC

Move xmpp client into XmppState

Created by O01eg on December 31, 2018

OANBCLN5TD5VQTSAQXSIXU6IUCXMKQSML2UJGQCKVTOCENEVABIAC

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 9 [5.17]
[3.434]
[4.320]
```
const ID_GET_ROSTER: &str = "id_get_roster0";
```

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

∅:D[4.321] → [4.321:354]

pub struct MaybeXmppConnection {

[4.321]

[4.354]

struct XmppState {
    client: Client,
}
struct MaybeXmppConnection {

Replacement in src/xmpp/mod.rs at line 18 [5.17]
∅:D[4.397] → [2.0:27]
```
    inner: Option<Client>,
```
[4.397]
[4.425]
```
    state: Option<XmppState>,
```
Replacement in src/xmpp/mod.rs at line 21 [5.17]
∅:D[6.106] → [7.73:101]
∅:D[7.73] → [7.73:101]
```
pub struct XmppConnection {
```
[6.106]
[8.31]
```
struct XmppConnection {
```
Replacement in src/xmpp/mod.rs at line 23 [5.17]
∅:D[8.74] → [2.28:47]
```
    inner: Client,
```
[8.74]
[4.448]
```
    state: XmppState,
```

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

∅:D[4.627] → [2.48:85]

            inner: Some(from.inner),

[4.627]

[4.666]

            state: Some(from.state),

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

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

∅:D[4.1011] → [2.112:149]

        F: future::Future + 'static,

[4.1011]

[4.1056]

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

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

∅:D[5.690] → [2.150:209]

        let MaybeXmppConnection { account, inner } = self;

[5.690]

[5.744]

        let MaybeXmppConnection { account, state } = self;

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

∅:D[5.745] → [2.210:315]

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

[5.745]

[4.1302]

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

Insertion in src/xmpp/mod.rs at line 62 [5.17]
[4.1405]
[4.1434]
```
                    .clone()
```

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

∅:D[4.1464] → [2.316:377]

                        future::loop_fn(account, |account| {

[4.1464]

[4.1530]

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

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

[4.2010]

[5.1476]


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

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

∅:D[2.435] → [2.435:494]

                                            inner: client,

[2.435]

[2.494]

                                            state: XmppState { client },

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

∅:D[2.589] → [2.589:720]

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

[2.589]

[2.720]

                                        .processing(XmppConnection::online, stop_future.clone())
                                        .map(|(conn, _)| conn)
                                        .map_err(|(acc, _)| acc)
                                        .and_then(|conn| conn.initial_roster(stop_future2))
                                        .and_then(|conn| conn.self_presence(stop_future3))

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

∅:D[4.3788] → [2.917:970]

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

[4.3788]

[4.3896]

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

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

∅:D[4.4137] → [2.971:1092]

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

[4.4137]

[4.4319]

        Item = (Self, Result<Either<F, T>, failure::Error>),
        Error = (
            std::rc::Rc<config::Account>,
            Result<Either<F, T>, failure::Error>,
        ),

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

∅:D[4.4375] → [2.1093:1138]

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

[4.4375]

[4.4476]

        E: Into<failure::Error>,
        S: FnMut(&mut Self, Event) -> Result<bool, failure::Error>,

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

∅:D[4.4612] → [2.1139:1277]

∅:D[2.1277] → [4.4752:4866]

∅:D[4.4752] → [4.4752:4866]

∅:D[4.4866] → [2.1278:1457]

∅:D[2.1457] → [4.4945:5003]

∅:D[4.4945] → [4.4945:5003]

∅:D[4.5003] → [2.1458:1615]

                let XmppConnection { inner, account } = xmpp;
                inner.into_future().select2(stop_future).then(|r| match r {
                    Ok(Either::A(((event, client), b))) => {
                        if let Some(event) = event {
                            let mut xmpp = XmppConnection {
                                inner: client,
                                account,
                            };
                            xmpp.xmpp_processing(&event);
                            if stop_condition(&mut xmpp, &event) {
                                future::ok(future::Loop::Break((xmpp, Ok(Either::A(b)))))

[4.4612]

[2.1615]

                let XmppConnection { state, account } = xmpp;
                state
                    .client
                    .into_future()
                    .select2(stop_future)
                    .then(|r| match r {
                        Ok(Either::A(((event, client), b))) => {
                            if let Some(event) = event {
                                let mut xmpp = XmppConnection {
                                    state: XmppState { client },
                                    account,
                                };
                                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, Err(e))),
                                }
                            } 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 },
                                        account,
                                    },
                                    Ok(Either::B(t)),
                                )))

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

∅:D[2.1652] → [2.1652:1746]

                                future::ok(future::Loop::Continue((xmpp, b, stop_condition)))

[2.1652]

[4.5504]

                                future::err((account, Ok(Either::B(t))))

Deletion in src/xmpp/mod.rs at line 180 [5.17]

∅:D[4.5534] → [4.5534:5636]

                        } else {
                            future::err((account, Ok(Either::A(b))))

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

∅:D[7.4531] → [4.5637:5706]

∅:D[4.5706] → [2.1747:1809]

∅:D[2.1809] → [4.5769:5830]

∅:D[4.5769] → [4.5769:5830]

∅:D[4.5830] → [2.1810:1877]

∅:D[2.1877] → [4.5898:6082]

∅:D[4.5898] → [4.5898:6082]

                    }
                    Ok(Either::B((t, a))) => {
                        if let Some(inner) = a.into_inner() {
                            future::ok(future::Loop::Break((
                                XmppConnection { inner, account },
                                Ok(Either::B(t)),
                            )))
                        } else {
                            future::err((account, Ok(Either::B(t))))

[7.4531]

[7.4974]

                        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 185 [5.17]

∅:D[7.5000] → [4.6083:6242]

∅:D[4.6242] → [2.1878:1940]

∅:D[2.1940] → [4.6305:6366]

∅:D[4.6305] → [4.6305:6366]

∅:D[4.6366] → [2.1941:2048]

∅:D[2.2048] → [4.6481:6546]

∅:D[4.6481] → [4.6481:6546]

∅:D[4.6546] → [2.2049:2108]

                    }
                    Err(Either::A((_e, b))) => future::err((account, Ok(Either::A(b)))),
                    Err(Either::B((e, a))) => {
                        if let Some(inner) = a.into_inner() {
                            future::ok(future::Loop::Break((
                                XmppConnection { inner, account },
                                Err(e),
                            )))
                        } else {
                            future::err((account, Err(e)))

[7.5000]

[7.5513]

                        Err(Either::B((e, a))) => {
                            if let Some(client) = a.into_inner() {
                                future::ok(future::Loop::Break((
                                    XmppConnection {
                                        state: XmppState { client },
                                        account,
                                    },
                                    Err(e.into()),
                                )))
                            } else {
                                future::err((account, Err(e.into())))
                            }

Replacement in src/xmpp/mod.rs at line 198 [5.17]
∅:D[9.2678] → [7.5539:5580]
∅:D[10.2678] → [7.5539:5580]
∅:D[7.5539] → [7.5539:5580]
```
                    }
                })
```
[7.5539]
[7.5580]
```
                    })
```

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

∅:D[4.6737] → [2.2109:3094]

    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,
                            }))
                        }
                        Some(Event::Stanza(s)) => {
                            info!("xmpp stanza: {:?}", s);
                            future::ok(future::Loop::Continue((client, account)))
                        }
                        _ => {
                            warn!("Disconnected");
                            future::err(account)

[4.6737]

[11.5558]

    fn online(&mut self, event: Event) -> Result<bool, failure::Error> {
        match event {
            Event::Online => {
                info!("Online!");
                Ok(true)
            }
            Event::Stanza(s) => {
                warn!("Stanza before online: {:?}", s);
                Ok(false)
            }
            _ => {
                error!("Disconnected while online");
                Err(format_err!("Disconnected while online"))
            }
        }
    }
    fn process_initial_roster(&mut self, event: Event) -> Result<bool, failure::Error> {
        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) => {
                                    Err(format_err!("Get error instead of roster"))
                                }
                                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) => Err(format_err!("Cann't parse roster: {}", e)),
                                    }
                                }
                                _ => Err(format_err!("Unknown result of roster")),
                            }
                        } else {
                            Ok(false)

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

∅:D[11.5584] → [2.3095:3261]

                    },
                    Err((e, _)) => {
                        error!("xmpp receive error: {}", e);
                        future::err(account)

[11.5584]

[4.8819]

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

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

∅:D[4.8841] → [2.3262:3307]

                })
            },
        ))

[4.8841]

[5.3783]

                }
                Err(_e) => Ok(false),
            }
        } else {
            Err(format_err!("Wrong event while waiting roster"))
        }

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

∅:D[4.10352] → [2.3308:3485]

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

[4.10352]

[4.10624]

    fn initial_roster<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 } = self;
        use tokio::prelude::Sink;
        let get_roster = stanzas::make_get_roster(ID_GET_ROSTER);
        let account2 = account.clone();
        info!("Quering roster... {:?}", get_roster);
        state
            .client
            .send(get_roster)
            .map_err(move |e| {
                error!("Error on querying roster: {}", e);
                (account2, Err(failure::SyncFailure::new(e).into()))
            })
            .and_then(move |client| {
                XmppConnection {
                    state: XmppState { client },
                    account,
                }
                .processing(XmppConnection::process_initial_roster, stop_future)
            })
            .then(|r| match r {
                Err((account, e)) => {
                    error!(
                        "Cann't wait roster: {}",
                        e.err().map_or_else(
                            || std::borrow::Cow::Borrowed("None"),
                            |e| e.to_string().into()
                        )
                    );
                    future::err(account)
                }
                Ok((conn, _)) => future::ok(conn),
            })
    }
    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 } = self;

Deletion in src/xmpp/mod.rs at line 319 [5.17]
∅:D[4.10716] → [2.3486:3524]
```
        info!("Sending presence...");
```

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

∅:D[4.10756] → [2.3525:5862]

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

[4.10756]

[4.12130]

        info!("Sending presence... {:?}", presence);
        state
            .client
            .send(presence)
            .map_err(|e| {
                error!("Error on send self-presence: {}", e);
                (account2, Err(failure::SyncFailure::new(e).into()))
            })
            .and_then(move |client| {
                XmppConnection {
                    state: XmppState { client },
                    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 {
                            Err(format_err!("Wrong event while waiting self-presence"))
                        }
                    },
                    stop_future,
                )
            })
            .then(|r| match r {
                Err((account, _e)) => {
                    error!("Cann't wait self-presence");
                    future::err(account)
                }
                Ok((conn, _)) => future::ok(conn),
            })

Replacement in src/xmpp/mod.rs at line 365 [5.17]
∅:D[5.3817] → [2.5863:5895]
```
struct XmppProcessState<F, S> {
```
[5.3817]
[12.86]
```
struct XmppState<F, S> {
```

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

∅:D[5.3920] → [2.5896:6049]

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

[5.3920]

[5.4058]

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

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

∅:D[5.4527] → [2.6050:6154]

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

[5.4527]

[5.4617]

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

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

∅:D[4.12779] → [2.6155:6240]

                            future::ok(future::Loop::Continue(XmppProcessState::new(

[4.12779]

[4.12857]

                            future::ok(future::Loop::Continue(XmppState::new(

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

∅:D[5.6342] → [2.6241:6303]

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

[5.6342]

[5.6572]

                        future::err(e)

Insertion in Cargo.toml at line 23 [13.11995]
[8.7111]
[9.8552]
```
try_from = "=0.2.2" # dependency of xmpp-parsers
```

Insertion in Cargo.lock at line 1110 [13.12201]

[8.44807]

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