o01eg/sendxmppd-rs - Change AGIW6YR3J5M3PIUP3UKFMBQT2VAK6CHDDSIMK4A3KJEXH5SKQEQAC

Use shared future for signal everywhere

Created by O01eg on December 29, 2018

AGIW6YR3J5M3PIUP3UKFMBQT2VAK6CHDDSIMK4A3KJEXH5SKQEQAC

Dependencies

In channels

main

Change contents

Deletion in src/xmpp/mod.rs at line 11 [4.17]
∅:D[3.34] → [3.34:106]
∅:D[5.434] → [4.210:211]
∅:D[3.106] → [4.210:211]
∅:D[4.210] → [4.210:211]
∅:D[6.3] → [6.3:31]
```
    account: std::rc::Rc<config::Account>,
    inner: Option<Client>,
}
pub struct XmppConnection {
```

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

∅:D[6.74] → [3.107:126]

∅:D[6.375] → [4.346:349]

∅:D[3.126] → [4.346:349]

∅:D[7.77] → [4.346:349]

∅:D[4.346] → [4.346:349]

∅:D[4.349] → [3.127:356]

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

[6.74]

[3.356]

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

Replacement in src/xmpp/mod.rs at line 25 [4.17]

∅:D[4.579] → [3.479:563]

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

[4.579]

[4.653]

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

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

∅:D[4.782] → [3.624:744]

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

[4.782]

[4.944]

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

Replacement in src/xmpp/mod.rs at line 46 [4.17]

∅:D[4.1522] → [3.745:1142]

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

[4.1522]

[4.1842]

                Self::online(client.split(), account)
                    .and_then(Self::self_presence)
                    .then(|r| match r {
                        Ok(conn) => future::ok(future::Loop::Break(conn)),
                        Err(acc) => future::ok(future::Loop::Continue(acc)),
                    })

Deletion in src/xmpp/mod.rs at line 55 [4.17]
∅:D[4.1874] → [3.1143:1145]
```
}
```

Deletion in src/xmpp/mod.rs at line 56 [4.17]

∅:D[4.1875] → [3.1146:1168]

∅:D[3.1168] → [2.0:1615]

∅:D[2.1615] → [8.533:555]

∅:D[8.533] → [8.533:555]

∅:D[8.555] → [2.1616:2063]

∅:D[2.2063] → [8.736:758]

∅:D[8.736] → [8.736:758]

∅:D[8.758] → [2.2064:2654]

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 58 [4.17]

∅:D[2.2767] → [2.2767:2865]

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

[2.2767]

[2.2865]

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

Replacement in src/xmpp/mod.rs at line 63 [4.17]

∅:D[2.2899] → [2.2899:3086]

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

[2.2899]

[2.3086]

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

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

∅:D[2.3180] → [2.3180:3256]

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

[2.3180]

[2.3256]

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

Replacement in src/xmpp/mod.rs at line 71 [4.17]

∅:D[2.3297] → [2.3297:3344]

                                inner: client,

[2.3297]

[2.3344]

                                inner: Some((sink, stream)),

Replacement in src/xmpp/mod.rs at line 76 [4.17]

∅:D[2.3513] → [2.3513:3595]

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

[2.3513]

[2.3595]

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

Replacement in src/xmpp/mod.rs at line 93 [4.17]

∅:D[5.745] → [6.888:942]

∅:D[6.942] → [3.1587:1649]

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

[5.745]

[5.784]

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

Replacement in src/xmpp/mod.rs at line 97 [4.17]

∅:D[5.785] → [3.1650:3187]

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

[5.785]

[5.2608]

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

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

∅:D[5.2657] → [3.3188:3286]

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

[5.2657]

[5.2743]

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

Deletion in src/xmpp/mod.rs at line 132 [4.17]

∅:D[5.2785] → [3.3287:3414]

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

Replacement in src/xmpp/mod.rs at line 133 [4.17]

∅:D[5.2823] → [3.3415:3996]

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

[5.2823]

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

Replacement in src/xmpp/mod.rs at line 181 [4.17]
∅:D[4.4311] → [4.4311:4355]
∅:D[4.4311] → [4.4311:4355]
∅:D[4.4311] → [4.4311:4355]
∅:D[4.4311] → [4.4311:4355]
```
    F: future::Future<Item = ()> + 'static,
```
[4.4311]
[11.465]
```
    F: future::Future + Clone + 'static,
    <F as hyper::rt::Future>::Error: std::fmt::Display,
```
Deletion in src/xmpp/mod.rs at line 185 [4.17]
∅:D[4.4357] → [11.519:594]
∅:D[9.136] → [4.4480:4481]
∅:D[10.557] → [4.4480:4481]
∅:D[11.594] → [4.4480:4481]
∅:D[4.4480] → [4.4480:4481]
```
    let signal = signal.map_err(|_| format_err!("Wrong shutdown signal"));
```

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

∅:D[4.5285] → [3.4189:4283]

                        Box::new(b.map(|b| future::Loop::Break((Some((b.0).1), b.1.into()))))

[4.5285]

[4.5372]

                        Box::new(b.map(|b| future::Loop::Break((Some((b.0).1), b.1))))

Replacement in src/xmpp/mod.rs at line 218 [4.17]

∅:D[4.5743] → [3.4284:4324]

                            x.1.into(),

[4.5743]

[4.5776]

                            x.1,

Replacement in src/xmpp/mod.rs at line 224 [4.17]

∅:D[4.6018] → [3.4325:4419]

                        Box::new(b.map(|b| future::Loop::Break((Some((b.0).1), b.1.into()))))

[4.6018]

[4.6105]

                        Box::new(b.map(|b| future::Loop::Break((Some((b.0).1), b.1))))

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

∅:D[6.4954] → [6.4954:5020]

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

[6.4954]

[6.5020]

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

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

∅:D[6.6389] → [6.6389:6536]

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

[6.6389]

[6.6536]

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