O01eg/sendxmppd-rs - Change FWJDW3G5KT66GHFS7IPPSFS3G5COH2BNN57GHTVT533FE7467QZAC

Allow process xmpp incoming stanzas with futures

Created by O01eg on January 1, 2019

FWJDW3G5KT66GHFS7IPPSFS3G5COH2BNN57GHTVT533FE7467QZAC

Dependencies

In channels

main

Change contents

Insertion in src/xmpp/mod.rs at line 7 [5.17]
[3.48]
[4.420]
```
use std::collections::HashMap;
```
Insertion in src/xmpp/mod.rs at line 16 [5.17]
[6.393]
[6.428]
```
    roster: HashMap<jid::Jid, ()>,
```

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

∅:D[2.19] → [2.19:132]

impl From<config::Account> for MaybeXmppConnection {
    fn from(from: config::Account) -> MaybeXmppConnection {

[2.19]

[2.132]

impl MaybeXmppConnection {
    fn new(account: config::Account) -> MaybeXmppConnection {

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

∅:D[2.162] → [2.162:207]

            account: std::rc::Rc::new(from),

[2.162]

[2.207]

            account: std::rc::Rc::new(account),

Deletion in src/xmpp/mod.rs at line 45 [5.17]
∅:D[7.77] → [8.126:128]
∅:D[9.682] → [8.126:128]
∅:D[8.126] → [8.126:128]
```
}
```
Deletion in src/xmpp/mod.rs at line 46 [5.17]
∅:D[8.129] → [2.233:372]
∅:D[2.372] → [9.772:802]
∅:D[9.772] → [9.772:802]
∅:D[9.802] → [2.373:400]
∅:D[2.400] → [6.578:603]
∅:D[4.527] → [6.578:603]
∅:D[6.603] → [5.499:515]
∅:D[10.535] → [5.499:515]
∅:D[11.111] → [5.499:515]
∅:D[9.829] → [5.499:515]
∅:D[12.174] → [5.499:515]
∅:D[5.499] → [5.499:515]
∅:D[5.515] → [2.401:403]
∅:D[2.403] → [5.515:516]
∅:D[5.515] → [5.515:516]
∅:D[5.516] → [2.404:431]
```
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 {
```

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

∅:D[5.554] → [2.432:483]

    /// don't connect only if stop_future resolved

[5.554]

[9.876]

    /// don't connect if stop_future resolved

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

∅:D[11.435] → [2.484:557]

                                            state: XmppState { client },

[11.435]

[11.494]

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

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

∅:D[6.1403] → [2.558:621]

                                        .map(|(conn, _)| conn)

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

[6.1468]

[6.1838]

                                        .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 124 [5.17]

∅:D[9.3788] → [6.2022:2073]

    fn xmpp_processing(&mut self, event: &Event) {

[9.3788]

[6.2073]

    fn xmpp_processing(
        self,
        event: &Event,
    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>> {

Insertion in src/xmpp/mod.rs at line 129 [5.17]
[6.2124]
[6.2124]
```
        future::ok(self)
```

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

∅:D[9.3983] → [2.622:658]

    /// or stop future was resolved

[9.3983]

[9.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 142 [5.17]

∅:D[9.4137] → [2.659:841]

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

[9.4137]

[9.4319]

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

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

∅:D[9.4335] → [9.4335:4375]

∅:D[9.4375] → [2.842:943]

        F: Future<Item = T, Error = E>,
        E: Into<failure::Error>,
        S: FnMut(&mut Self, Event) -> Result<bool, failure::Error>,

[9.4335]

[9.4476]

        F: Future<Item = T, Error = E> + 'static,
        S: FnMut(&mut Self, Event) -> Result<bool, ()> + 'static,
        T: 'static,
        E: 'static,

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

∅:D[9.4612] → [2.944:1639]

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

[9.4612]

[2.1639]

                let XmppConnection {
                    state: XmppState { client, roster },
                    account,
                } = xmpp;
                client.into_future().select2(stop_future).then(|r| match r {
                    Ok(Either::A(((event, client), b))) => {
                        if let Some(event) = event {
                            let xmpp = XmppConnection {
                                state: XmppState { client, roster },
                                account,
                            };
                            Box::new(xmpp.xmpp_processing(&event).then(|r| match r {
                                Ok(mut xmpp) => match stop_condition(&mut xmpp, event) {

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

∅:D[2.2096] → [2.2096:2179]

∅:D[2.2179] → [6.3025:3059]

∅:D[6.3025] → [6.3025:3059]

∅:D[6.3059] → [2.2180:2290]

∅:D[2.2290] → [10.3255:3285]

∅:D[6.3149] → [10.3255:3285]

∅:D[13.742] → [10.3255:3285]

∅:D[14.1375] → [10.3255:3285]

∅:D[10.3255] → [10.3255:3285]

∅:D[10.3285] → [2.2291:2940]

                                    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)),
                                )))
                            } else {
                                future::err((account, Ok(Either::B(t))))
                            }

[2.2096]

[2.2940]

                                    Err(_e) => future::err((xmpp.account, Ok(Either::A(b)))),
                                },
                                Err(account) => future::err((account, Ok(Either::A(b)))),
                            })) as Box<dyn Future<Item = _, Error = _>>
                        } else {
                            Box::new(future::err((account, Ok(Either::A(b)))))

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

∅:D[2.2966] → [2.2966:3076]

∅:D[2.3076] → [14.1409:1478]

∅:D[14.1409] → [14.1409:1478]

∅:D[14.1478] → [10.3285:3311]

∅:D[10.3285] → [10.3285:3311]

∅:D[10.3311] → [2.3077:3695]

∅:D[7.2179] → [15.4505:4531]

∅:D[2.3695] → [15.4505:4531]

∅:D[16.2179] → [15.4505:4531]

∅:D[9.5636] → [15.4505:4531]

∅:D[14.2058] → [15.4505:4531]

∅:D[8.5335] → [15.4505:4531]

∅:D[15.4505] → [15.4505:4531]

∅:D[15.4531] → [2.3696:3719]

                        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 },
                                        account,
                                    },
                                    Err(e.into()),
                                )))
                            } else {
                                future::err((account, Err(e.into())))
                            }
                        }
                    })

[2.2966]

[15.5580]

                    }
                    Ok(Either::B((t, a))) => Box::new(if let Some(client) = a.into_inner() {
                        future::ok(future::Loop::Break((
                            XmppConnection {
                                state: XmppState { client, roster },
                                account,
                            },
                            Ok(Either::B(t)),
                        )))
                    } else {
                        future::err((account, Ok(Either::B(t))))
                    }),
                    Err(Either::A((e, b))) => {
                        warn!("XMPP error: {}", e.0);
                        Box::new(future::err((account, Ok(Either::A(b)))))
                    }
                    Err(Either::B((e, a))) => Box::new(if let Some(client) = a.into_inner() {
                        future::ok(future::Loop::Break((
                            XmppConnection {
                                state: XmppState { client, roster },
                                account,
                            },
                            Err(e),
                        )))
                    } else {
                        future::err((account, Err(e)))
                    }),
                })

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

∅:D[9.6691] → [2.3720:3839]

    /// returns error if something went wrong
    fn online(&mut self, event: Event) -> Result<bool, failure::Error> {

[9.6691]

[6.4011]

    /// returns error if something went wrong and xmpp connection is broken
    fn online(&mut self, event: Event) -> Result<bool, ()> {

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

∅:D[6.4339] → [2.3840:3902]

                Err(format_err!("Disconnected while online"))

[6.4339]

[6.4363]

                Err(())

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

∅:D[6.4394] → [2.3903:3992]

    fn process_initial_roster(&mut self, event: Event) -> Result<bool, failure::Error> {

[6.4394]

[6.4471]

    fn process_initial_roster(&mut self, event: Event) -> Result<bool, ()> {

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

∅:D[6.4859] → [2.3993:4077]

                                    Err(format_err!("Get error instead of roster"))

[6.4859]

[6.4978]

                                    error!("Get error instead of roster");
                                    Err(())

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

∅:D[6.5907] → [2.4078:4176]

                                        Err(e) => Err(format_err!("Cann't parse roster: {}", e)),

[6.5907]

                                        Err(e) => {
                                            error!("Cann't parse roster: {}", e);
                                            Err(())
                                        }

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

∅:D[6.6168] → [2.4177:4260]

                                _ => Err(format_err!("Unknown result of roster")),

[6.6168]

                                _ => {
                                    error!("Unknown result of roster");
                                    Err(())
                                }

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

∅:D[6.6299] → [2.4261:4326]

                        Err(format_err!("Iq stanza without id"))

[6.6299]

[9.8819]

                        error!("Iq stanza without id");
                        Err(())

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

∅:D[6.6475] → [2.4327:4392]

            Err(format_err!("Wrong event while waiting roster"))

[6.6475]

[6.6551]

            error!("Wrong event while waiting roster");
            Err(())

Replacement in src/xmpp/mod.rs at line 288 [5.17]
∅:D[6.6717] → [6.6717:6747]
∅:D[6.6717] → [6.6717:6747]
∅:D[6.6747] → [2.4393:4426]
```
        F: Future<Error = E>,
        E: Into<failure::Error>,
```
[6.6717]
[6.6747]
```
        F: Future<Error = E> + 'static,
        E: 'static,
```

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

∅:D[6.6753] → [2.4427:4481]

        let XmppConnection { account, state } = self;

[6.6753]

[6.6870]

        let XmppConnection {
            account,
            state: XmppState { client, roster },
        } = self;

Replacement in src/xmpp/mod.rs at line 301 [5.17]
∅:D[6.7065] → [2.4482:4516]
```
        state
            .client
```
[6.7065]
[6.7080]
```
        client
```

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

∅:D[6.7201] → [2.4517:4586]

                (account2, Err(failure::SyncFailure::new(e).into()))

[6.7201]

[6.7226]

                account2

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

∅:D[6.7312] → [2.4587:4636]

                    state: XmppState { client },

[6.7312]

[6.7369]

                    state: XmppState { client, roster },

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

∅:D[6.7497] → [2.4637:5125]

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

[6.7497]

[6.7778]

                .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 327 [5.17]

∅:D[6.7942] → [6.7942:8005]

        F: Future<Error = E>,
        E: Into<failure::Error>,

[6.7942]

[6.8005]

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

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

∅:D[6.8011] → [2.5126:5180]

        let XmppConnection { account, state } = self;

[6.8011]

[9.10624]

        let XmppConnection {
            account,
            state: XmppState { client, roster },
        } = self;

Replacement in src/xmpp/mod.rs at line 340 [5.17]
∅:D[6.8183] → [2.5181:5215]
```
        state
            .client
```
[6.8183]
[6.8198]
```
        client
```

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

∅:D[6.8315] → [2.5216:5285]

                (account2, Err(failure::SyncFailure::new(e).into()))

[6.8315]

[6.8340]

                account2

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

∅:D[6.8426] → [2.5286:5335]

                    state: XmppState { client },

[6.8426]

[6.8483]

                    state: XmppState { client, roster },

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

∅:D[6.9105] → [2.5336:5424]

                            Err(format_err!("Wrong event while waiting self-presence"))

[6.9105]

[6.9220]

                            error!("Wrong event while waiting self-presence");
                            Err(())

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

[6.9320]

[2.5425]

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

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

∅:D[2.5440] → [2.5440:5679]

∅:D[2.5679] → [6.9601:9616]

∅:D[13.2504] → [6.9601:9616]

            .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 382 [5.17]
∅:D[5.3817] → [2.5680:5712]
```
struct XmppProcessState<F, S> {
```
[5.3817]
[17.86]
```
struct XmppState<F, S> {
```

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

∅:D[5.3920] → [2.5713:5866]

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 408 [5.17]
∅:D[5.4357] → [2.5867:5898]
```
    let conn = account.into();
```
[5.4357]
[5.4526]
```
    let conn = MaybeXmppConnection::new(account);
```

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

∅:D[5.4527] → [2.5899:6003]

    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[5.4692] → [2.6004:6783]

        conn.connect(signal.clone()).and_then(|conn| {
            info!("xmpp connected!");
            conn.processing(|_, _| Ok(false), cmd_recv.into_future())
                .then(|r| match r {
                    Ok((conn, r)) => match r {
                        Ok(Either::A(f)) => {
                            if let Some(cmd_recv) = f.into_inner() {
                                future::ok(future::Loop::Continue(XmppProcessState {
                                    cmd_recv,
                                    signal,
                                    conn: conn.into(),
                                }))
                            } else {
                                future::err(format_err!("Command receiver is gone"))
                            }

[5.4692]

[9.13130]

        conn.connect(signal.clone())
            .and_then(|conn| {
                info!("xmpp connected!");
                cmd_recv
                    .into_future()
                    .map_err(|_| {
                        error!("Got error on recv cmd");
                        format_err!("Receive cmd error")
                    })
                    .map(|(cmd, cmd_recv)| (cmd, cmd_recv, conn))
            })
            .then(|r| {
                match r {
                    Ok((cmd, cmd_recv, conn)) => {
                        if let Some(_cmd) = cmd {
                            info!("Got cmd");
                            // got cmd, continue
                            future::ok(future::Loop::Continue(XmppState::new(
                                cmd_recv,
                                signal,
                                conn.into(),
                            )))
                        } else {
                            future::ok(future::Loop::Break((None, conn.into())))

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

∅:D[9.13156] → [2.6784:8876]

                        Ok(Either::B((cmd, cmd_recv))) => {
                            if let Some(_cmd) = cmd {
                                info!("Got command");
                                future::ok(future::Loop::Continue(XmppProcessState {
                                    cmd_recv,
                                    signal,
                                    conn: conn.into(),
                                }))
                            } else {
                                future::ok(future::Loop::Break(()))
                            }
                        }
                        Err(_) => future::err(format_err!("Command receiver is broken")),
                    },
                    Err((account, r)) => match r {
                        Ok(Either::A(f)) => {
                            if let Some(cmd_recv) = f.into_inner() {
                                future::ok(future::Loop::Continue(XmppProcessState {
                                    cmd_recv,
                                    signal,
                                    conn: account.into(),
                                }))
                            } else {
                                future::err(format_err!("Command receiver is gone"))
                            }
                        }
                        Ok(Either::B((cmd, cmd_recv))) => {
                            if let Some(_cmd) = cmd {
                                error!("Xmpp connection broken while get command");
                                future::ok(future::Loop::Continue(XmppProcessState {
                                    cmd_recv,
                                    signal,
                                    conn: account.into(),
                                }))
                            } else {
                                future::ok(future::Loop::Break(()))
                            }
                        }
                        Err(_) => future::err(format_err!("Command receiver is broken")),
                    },
                })
        })

[9.13156]

[6.10245]

                    }
                    Err(e) => {
                        // got cmd error, its bad
                        error!("Cmd error: {}", e);
                        future::err(e)
                    }
                }
            })
    })
    .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(()))
        }