O01eg/sendxmppd-rs - Change PVCRPP3BXTLRRT2VK2BCKFCRJXUOX3AEF2A6UCPOSZYEKQEMBWDQC

Some servers don't send to in initial presence

Created by O01eg on December 31, 2018

PVCRPP3BXTLRRT2VK2BCKFCRJXUOX3AEF2A6UCPOSZYEKQEMBWDQC

Dependencies

In channels

main

Change contents

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

∅:D[3.554] → [2.0:109]

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

[3.554]

[3.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 + Clone + 'static,
        <F as hyper::rt::Future>::Error: Into<failure::Error> + Send,
    {

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

∅:D[4.910] → [2.110:162]

                as Box<Future<Item = _, Error = E>>

[4.910]

[3.944]

                as Box<Future<Item = _, Error = _>>

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

∅:D[3.961] → [2.163:678]

            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");

[3.961]

[3.1476]

            Box::new(
                stop_future
                    .clone()
                    .select2(
                        future::loop_fn(account, move |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");
                                    let stop_future2 = stop_future.clone();

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

∅:D[3.1477] → [2.679:1137]

                // future to wait for online
                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)),
                })
            }))

[3.1477]

[3.1858]

                                    // future to wait for online
                                    Box::new(
                                        XmppConnection {
                                            inner: client,
                                            account,
                                        }
                                        .processing(XmppConnection::online, stop_future.clone())
                                        .map(|(conn, _)| conn)
                                        .map_err(|(acc, _)| acc)
                                        .and_then(|conn| conn.self_presence(stop_future2))
                                        .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"))
                        }
                    }),
            )

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

∅:D[4.3427] → [2.1138:1190]

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

[4.3427]

[4.3535]

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

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

∅:D[4.3776] → [2.1191:1312]

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

[4.3776]

[4.3958]

        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 126 [3.17]

∅:D[4.4014] → [2.1313:1358]

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

[4.4014]

[4.4116]

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

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

∅:D[4.4701] → [2.1359:1647]

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

[4.4701]

[4.5203]

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

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

∅:D[4.5802] → [2.1648:1837]

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

[4.5802]

[4.5891]

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

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

∅:D[4.6129] → [2.1838:1878]

                                Err(e),

[4.6129]

[4.6176]

                                Err(e.into()),

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

∅:D[4.6241] → [2.1879:1938]

                            future::err((account, Err(e)))

[4.6241]

[5.5513]

                            future::err((account, Err(e.into())))

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

∅:D[4.6432] → [2.1939:2924]

∅:D[2.2924] → [6.1664:1690]

∅:D[6.1664] → [6.1664:1690]

∅:D[6.1690] → [2.2925:3091]

∅:D[2.3091] → [6.1784:1806]

∅:D[6.1784] → [6.1784:1806]

∅:D[6.1806] → [2.3092:3137]

    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)
                        }
                    },
                    Err((e, _)) => {
                        error!("xmpp receive error: {}", e);
                        future::err(account)
                    }
                })
            },
        ))

[4.6432]

[7.3687]

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

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

∅:D[7.3694] → [2.3138:3233]

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

[7.3694]

[2.3233]

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

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

∅:D[4.7343] → [2.3355:5692]

        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.7343]

[3.3783]

        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 {
                    inner: 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)
                            {
                                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 277 [3.17]

∅:D[3.4311] → [2.5693:5737]

    F: future::Future<Item = ()> + 'static,

[3.4311]

[8.465]

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

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

∅:D[3.4357] → [2.5738:5814]

    let signal = signal.map_err(|_| format_err!("Wrong shutdown signal"));

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

∅:D[3.4692] → [2.5815:5884]

        signal
            .select2(conn.connect().and_then(|conn| {

[3.4692]

[3.4761]

        conn.connect(signal.clone())
            .and_then(|conn| {

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

∅:D[8.675] → [2.5885:5941]

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

[8.675]

[8.731]

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

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

∅:D[3.5146] → [2.5942:6239]

                    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.into()))))
                            as Box<Future<Item = _, Error = _>>

[3.5146]

[2.6239]

                    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 315 [3.17]

∅:D[2.6261] → [2.6261:6986]

∅:D[2.6986] → [9.7045:7067]

∅:D[4.9763] → [9.7045:7067]

∅:D[9.7045] → [9.7045:7067]

∅:D[9.7067] → [2.6987:7036]

                    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.into(),
                        )))) 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.into()))))
                            as Box<Future<Item = _, Error = _>>
                    }
                    Err(Either::B((e, _a))) => {

[2.6261]

[3.6240]

                    Err(e) => {

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

∅:D[3.6342] → [2.7037:7173]

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

[3.6342]

[3.6572]

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

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

∅:D[8.1354] → [2.7174:7242]

            Box::new(future::err(format_err!("cmd receiver gone")))

[8.1354]

[8.1422]

            Box::new(future::ok(()))

Replacement in src/main.rs at line 63 [11.17]

∅:D[10.4954] → [2.7243:7309]

        .map_err(|e| error!("Cann't get CTRL+C signal: {}", e.0))

[10.4954]

[10.5020]

        .map_err(|e| {
            error!("Cann't get CTRL+C signal: {}", e.0);
            e.0
        })

Replacement in src/main.rs at line 100 [11.17]

∅:D[10.6389] → [2.7310:7457]

        let result = ctrt.block_on(xmpp_process(
            ctrl_c.clone().map(|_| ()),
            recv,
            config.account,
        ));

[10.6389]

[10.6536]

        let result = ctrt.block_on(xmpp_process(ctrl_c.clone(), recv, config.account));