O01eg/sendxmppd-rs - Change XGP44R5HBQAHBXEJKN4AB77OZ32B32FPUHSUILZDVNO4QLUOTW6QC

Rework stopping xmpp connection

Created by O01eg on December 29, 2018

XGP44R5HBQAHBXEJKN4AB77OZ32B32FPUHSUILZDVNO4QLUOTW6QC

Dependencies

In channels

main

Change contents

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

[3.34]

[4.31]

    account: std::rc::Rc<config::Account>,
    inner: Option<Client>,
}
pub struct XmppConnection {

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

∅:D[4.74] → [2.0:77]

    inner: Option<(stream::SplitSink<Client>, stream::SplitStream<Client>)>,

[4.74]

[3.356]

    inner: Client,
}
impl From<XmppConnection> for MaybeXmppConnection {
    fn from(from: XmppConnection) -> MaybeXmppConnection {
        MaybeXmppConnection {
            account: from.account,
            inner: Some(from.inner),
        }
    }

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

∅:D[5.554] → [5.554:579]

∅:D[5.579] → [2.78:152]

    /// Error shoud be !
    fn connect<E: 'static>(self) -> impl Future<Item = Self, Error = E> {

[5.554]

[5.653]

    /// don't connect if stop_future resolved
    fn connect<F>(
        self,
        stop_future: F,
    ) -> impl Future<Item = XmppConnection, Error = failure::Error>
    where
        F: future::Future + 'static,
        <F as hyper::rt::Future>::Error: Into<failure::Error> + Send,
    {

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

∅:D[5.782] → [2.153:320]

            Box::new(future::ok(MaybeXmppConnection {
                account,
                inner: Some(inner),
            })) as Box<Future<Item = _, Error = E>>

[5.782]

[5.944]

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

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

∅:D[5.961] → [5.961:1476]

            Box::new(future::loop_fn(account, |account| {
                info!("xmpp initialization...");
                let mut res_client = Client::new(&account.jid, &account.password);
                while let Err(e) = res_client {
                    error!("Cann't init xmpp client: {}", e);
                    res_client = Client::new(&account.jid, &account.password);
                }
                let client = res_client.expect("Cann't init xmpp client");
                info!("xmpp initialized");

[5.961]

[5.1476]

            Box::new(
                stop_future
                    .select2(
                        future::loop_fn(account, |account| {
                            info!("xmpp initialization...");
                            let res_client = Client::new(&account.jid, &account.password);
                            match res_client {
                                Err(_e) => Box::new(future::ok(future::Loop::Continue(account)))
                                    as Box<Future<Item = _, Error = _>>,
                                Ok(client) => {
                                    info!("xmpp initialized");

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

∅:D[5.1477] → [5.1477:1522]

∅:D[5.1522] → [2.321:426]

                // future to wait for online
                Self::online(client.split(), account)
                    .and_then(Self::self_presence)

[5.1477]

[2.426]

                                    // future to wait for online
                                    Box::new(
                                        XmppConnection {
                                            inner: client,
                                            account,
                                        }
                                        .online()
                                        .and_then(XmppConnection::self_presence)
                                        .then(
                                            |r| match r {
                                                Ok(conn) => future::ok(future::Loop::Break(conn)),
                                                Err(acc) => future::ok(future::Loop::Continue(acc)),
                                            },
                                        ),
                                    )
                                }
                            }
                        })
                        .map_err(|_: ()| ()),
                    )

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

∅:D[2.466] → [2.466:641]

∅:D[3.1142] → [5.1842:1858]

∅:D[2.641] → [5.1842:1858]

∅:D[5.1842] → [5.1842:1858]

                        Ok(conn) => future::ok(future::Loop::Break(conn)),
                        Err(acc) => future::ok(future::Loop::Continue(acc)),
                    })
            }))

[2.466]

[5.1858]

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

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

[5.1868]

    }
}
impl XmppConnection {
    /// base XMPP processing
    fn xmpp_processing(&mut self, _event: &Event) {}
    /// process event from xmpp stream
    /// returns from future when condition met
    /// or stop future was resolved
    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>,
        S: FnMut(&mut Self, &Event) -> bool,
    {
        future::loop_fn(
            (self, stop_future, stop_condition),
            |(xmpp, stop_future, mut stop_condition)| {
                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)))))
                            } else {
                                future::ok(future::Loop::Continue((xmpp, b, stop_condition)))
                            }
                        } else {
                            future::err((account, Ok(Either::A(b))))
                        }
                    }
                    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))))
                        }
                    }
                    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)))
                        }
                    }
                })
            },
        )

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

∅:D[8.2767] → [2.642:860]

    fn online(
        (sink, stream): (stream::SplitSink<Client>, stream::SplitStream<Client>),
        account: std::rc::Rc<config::Account>,
    ) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>> {

[8.2767]

[8.2865]

    fn online(self) -> impl Future<Item = XmppConnection, Error = std::rc::Rc<config::Account>> {

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

∅:D[8.2899] → [2.861:1051]

            (sink, stream, account),
            |(sink, stream, account)| {
                stream.into_future().then(|r| match r {
                    Ok((event, stream)) => match event {

[8.2899]

[8.3086]

            (self.inner, self.account),
            |(client, account)| {
                client.into_future().then(|r| match r {
                    Ok((event, client)) => match event {

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

∅:D[8.3180] → [2.1052:1133]

                            future::ok(future::Loop::Break(MaybeXmppConnection {

[8.3180]

[8.3256]

                            future::ok(future::Loop::Break(XmppConnection {

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

∅:D[8.3297] → [2.1134:1195]

                                inner: Some((sink, stream)),

[8.3297]

[8.3344]

                                inner: client,

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

∅:D[8.3513] → [2.1196:1284]

                            future::ok(future::Loop::Continue((sink, stream, account)))

[8.3513]

[8.3595]

                            future::ok(future::Loop::Continue((client, account)))

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

∅:D[6.745] → [2.1285:1428]

        let MaybeXmppConnection { account, inner } = self;
        if let Some((sink, stream)) = inner {
            use tokio::prelude::Sink;

[6.745]

[6.784]

        let XmppConnection { account, inner } = self;
        let client = inner;
        use tokio::prelude::Sink;

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

∅:D[6.785] → [2.1429:3348]

            let presence = stanzas::make_presence(&account);
            info!("Sending presence...");
            Box::new(
                sink.send(presence)
                    .map_err(|e| {
                        error!("Error on send self-presence: {}", e);
                        "Cann't send self-presence".to_owned()
                    })
                    .join(
                        future::loop_fn((account.clone(), stream), |(account, stream)| {
                            stream
                                .into_future()
                                .map_err(|(e, _)| {
                                    error!("Error on reading self-presence: {}", e);
                                    "Cann't read self-presence".to_owned()
                                })
                                .and_then(|(event, stream)| 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(stream))
                                            } else {
                                                future::ok(future::Loop::Continue((
                                                    account, stream,
                                                )))
                                            }

[6.785]

[6.2608]

        let presence = stanzas::make_presence(&account);
        info!("Sending presence...");
        let account2 = account.clone();
        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))

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

∅:D[6.2657] → [2.3349:3435]

                                            future::err("Got wrong event".to_owned())

[6.2657]

[6.2743]

                                            future::ok(future::Loop::Continue((account, client)))

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

[6.2785]

                                    } else {
                                        future::err("Got wrong event".to_owned())

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

∅:D[6.2823] → [2.3436:4300]

                                    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(MaybeXmppConnection {
                            account,
                            inner: Some(inner),
                        }),
                    }),
            )
        } else {
            warn!("Don't gen connection on self-presence");
            Box::new(future::err(account)) as Box<Future<Item = _, Error = _>>
        }

[6.2823]

[5.3783]

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

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

∅:D[2.4342] → [2.4342:4398]

    <F as hyper::rt::Future>::Error: std::fmt::Display,

[2.4342]

[9.465]

    <F as hyper::rt::Future>::Error: std::fmt::Display + Into<failure::Error> + Send,

Replacement in src/xmpp/mod.rs at line 291 [5.17]
∅:D[5.4692] → [5.4692:4761]
∅:D[5.4692] → [5.4692:4761]
∅:D[5.4692] → [5.4692:4761]
∅:D[5.4692] → [5.4692:4761]
```
        signal
            .select2(conn.connect().and_then(|conn| {
```
[5.4692]
[5.4761]
```
        conn.connect(signal.clone())
            .and_then(|conn| {
```

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

∅:D[9.675] → [9.675:731]

                    .map(|f| (f, conn))
            }))

[9.675]

[9.731]

                    .map(|(cmd, cmd_recv)| (cmd, cmd_recv, conn))
            })

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

∅:D[5.5146] → [5.5146:5285]

∅:D[5.5285] → [2.4399:4486]

∅:D[4.3375] → [5.5372:5743]

∅:D[3.4283] → [5.5372:5743]

∅:D[2.4486] → [5.5372:5743]

∅:D[7.5034] → [5.5372:5743]

∅:D[5.5372] → [5.5372:5743]

∅:D[5.5743] → [2.4487:4520]

∅:D[4.3416] → [5.5776:6018]

∅:D[3.4324] → [5.5776:6018]

∅:D[2.4520] → [5.5776:6018]

∅:D[7.5068] → [5.5776:6018]

∅:D[5.5776] → [5.5776:6018]

∅:D[5.6018] → [2.4521:4608]

∅:D[4.3511] → [5.6105:6169]

∅:D[3.4419] → [5.6105:6169]

∅:D[2.4608] → [5.6105:6169]

∅:D[7.5156] → [5.6105:6169]

∅:D[5.6105] → [5.6105:6169]

                    Ok(Either::A((_x, b))) => {
                        info!("Got signal");
                        // got signal, breaks
                        Box::new(b.map(|b| future::Loop::Break((Some((b.0).1), b.1))))
                            as Box<Future<Item = _, Error = _>>
                    }
                    Ok(Either::B((x, a))) => {
                        info!("Got cmd");
                        // got cmd, continue
                        Box::new(future::ok(future::Loop::Continue(XmppState::new(
                            (x.0).1,
                            a,
                            x.1,
                        )))) as Box<Future<Item = _, Error = _>>
                    }
                    Err(Either::A((e, b))) => {
                        // got signal error, breaks
                        error!("Signal error: {}", e);
                        Box::new(b.map(|b| future::Loop::Break((Some((b.0).1), b.1))))
                            as Box<Future<Item = _, Error = _>>

[5.5146]

[5.6169]

                    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 317 [5.17]
∅:D[5.6191] → [5.6191:6240]
∅:D[5.6191] → [5.6191:6240]
∅:D[5.6191] → [5.6191:6240]
∅:D[5.6191] → [5.6191:6240]
∅:D[5.6191] → [5.6191:6240]
```
                    Err(Either::B((e, _a))) => {
```
[5.6191]
[5.6240]
```
                    Err(e) => {
```

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

∅:D[5.6342] → [9.756:892]

                        Box::new(future::err(format_err!("Cmd error")))
                            as Box<Future<Item = _, Error = _>>

[5.6342]

[5.6572]

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

Replacement in src/xmpp/mod.rs at line 335 [5.17]
∅:D[9.1354] → [9.1354:1422]
∅:D[9.1354] → [9.1354:1422]
∅:D[9.1354] → [9.1354:1422]
∅:D[9.1354] → [9.1354:1422]
```
            Box::new(future::err(format_err!("cmd receiver gone")))
```
[9.1354]
[9.1422]
```
            Box::new(future::ok(()))
```