o01eg/sendxmppd-rs - Change OGMBXBKPWAX4A4CY2LJITNKUYHWBFE4EVBHI4NRZO5VNZ5KXM2VQC

Move online to XmppConnection

Created by O01eg on December 29, 2018

OGMBXBKPWAX4A4CY2LJITNKUYHWBFE4EVBHI4NRZO5VNZ5KXM2VQC

Dependencies

In channels

main

Change contents

Insertion in src/xmpp/mod.rs at line 9 [4.17]

[3.434]

[4.210]


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

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

∅:D[5.74] → [6.0:77]

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

[5.74]

[4.346]

    inner: Client,

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

∅:D[4.349] → [6.78:182]

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

[4.349]

[3.479]

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

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

∅:D[4.579] → [6.183:257]

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

[4.579]

[4.653]

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

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

∅:D[4.690] → [6.258:312]

        let XmppConnection { account, inner } = self;

[4.690]

[4.744]

        let MaybeXmppConnection { account, inner } = self;

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

∅:D[4.782] → [6.313:475]

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

[4.782]

[4.944]

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

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

∅:D[4.1522] → [6.476:581]

∅:D[5.643] → [4.1627:1842]

∅:D[6.581] → [4.1627:1842]

∅:D[4.1627] → [4.1627: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)),
                    })

[4.1522]

[4.1842]

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

Insertion in src/xmpp/mod.rs at line 71 [4.17]
[4.1874]
[4.1874]
```
}
```
Insertion in src/xmpp/mod.rs at line 73 [4.17]
[4.1875]
[4.1875]
```
impl XmppConnection {
```

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

∅:D[4.1988] → [4.1988:2003]

∅:D[4.2003] → [6.582:664]

∅:D[6.664] → [3.528:575]

∅:D[7.119] → [3.528:575]

∅:D[4.2085] → [3.528:575]

∅:D[3.575] → [6.665:739]

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

[4.1988]

[6.739]

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

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

∅:D[6.773] → [6.773:963]

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

[6.773]

[6.963]

            (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 86 [4.17]

∅:D[6.1174] → [6.1174:1235]

                                inner: Some((sink, stream)),

[6.1174]

[6.1235]

                                inner: client,

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

∅:D[6.1404] → [6.1404:1492]

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

[6.1404]

[6.1492]

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

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

∅:D[5.942] → [6.1843:1927]

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

[5.942]

[3.784]

        let client = inner;
        use tokio::prelude::Sink;

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

∅:D[3.785] → [6.1928:3847]

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

[3.785]

[3.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 141 [4.17]

∅:D[3.2657] → [6.3848:3934]

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

[3.2657]

[3.2743]

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

Insertion in src/xmpp/mod.rs at line 143 [4.17]

[3.2785]

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

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

∅:D[3.2823] → [6.3935:4794]

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

[3.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(XmppConnection { account, inner }),
                }),
        )

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

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

pub struct XmppCommand {
    pub xmpp_to: String,
    pub message: String,
}

[8.60]

[4.3816]

pub struct XmppCommand;

Replacement in src/xmpp/mod.rs at line 169 [4.17]
∅:D[4.3891] → [6.4795:4821]
```
    conn: XmppConnection,
```
[4.3891]
[4.3917]
```
    conn: MaybeXmppConnection,
```

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

∅:D[8.133] → [6.4822:4900]

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

[8.133]

[4.4038]

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

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

∅:D[4.4481] → [6.4901:4946]

    let conn = XmppConnection::new(account);

[4.4481]

[4.4526]

    let conn = MaybeXmppConnection::new(account);

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

∅:D[4.5285] → [6.4947:5034]

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

[4.5285]

[4.5372]

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

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

∅:D[4.5743] → [6.5035:5068]

                            x.1,

[4.5743]

[4.5776]

                            x.1.into(),

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

∅:D[4.6018] → [6.5069:5156]

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

[4.6018]

[4.6105]

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

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

∅:D[9.900] → [6.5157:5226]

    .and_then(|(opt_cmd_recv, _conn): (Option<S>, XmppConnection)| {

[9.900]

[9.969]

    .and_then(|(opt_cmd_recv, _conn): (Option<S>, MaybeXmppConnection)| {

Insertion in src/main.rs at line 22 [10.17]
[5.3921]
[5.3953]
```
use hyper::service::service_fn;
```
Replacement in src/main.rs at line 28 [10.17]
∅:D[5.4066] → [2.78:122]
```
use tokio::prelude::{Future, Sink, Stream};
```
[5.4066]
[5.4092]
```
use tokio::prelude::Sink;
```

Deletion in src/main.rs at line 38 [10.17]

∅:D[5.4260] → [2.123:481]


struct ServiceCmd {
    cmd_send: tokio_channel::mpsc::Sender<XmppCommand>,
}
impl hyper::service::Service for ServiceCmd {
    type ReqBody = Body;
    type ResBody = Body;
    type Error = Box<dyn std::error::Error + Sync + Send + 'static>;
    type Future =
        Box<dyn Future<Item = Response<Self::ResBody>, Error = Self::Error> + Send + 'static>;

Deletion in src/main.rs at line 39 [10.17]

∅:D[11.399] → [2.482:5634]

    fn call(&mut self, req: Request<Self::ReqBody>) -> Self::Future {
        let xmpp_to_opt = req.headers().get("X-XMPP-To");
        let xmpp_to_res: Result<String, std::borrow::Cow<str>> = xmpp_to_opt.map_or_else(
            || Err("none".into()),
            |xmpp_to| {
                xmpp_to.to_str().map(|x| x.to_owned()).map_err(|e| {
                    format!("\"{}\" {}", String::from_utf8_lossy(xmpp_to.as_bytes()), e).into()
                })
            },
        );
        match xmpp_to_res {
            Err(err) => {
                warn!("Unknown destination: {}", err);
                Box::new(tokio::prelude::future::result(
                    Response::builder()
                        .status(hyper::StatusCode::BAD_REQUEST)
                        .body(Body::from(format!("Unknown destination: {}", err)))
                        .map_err(|e| {
                            Box::new(e) as Box<dyn std::error::Error + Sync + Send + 'static>
                        }),
                )) as Box<Future<Item = _, Error = _> + Send + 'static>
            }
            Ok(xmpp_to) => {
                info!("Got request. Reading body...");
                let cmd_send = self.cmd_send.clone();
                Box::new(
                    req.into_body()
                        .map_err(|e| {
                            Box::new(e) as Box<dyn std::error::Error + Sync + Send + 'static>
                        })
                        .fold(String::new(), |mut acc, ch| {
                            std::str::from_utf8(&*ch).map(|s| {
                                acc.push_str(s);
                                acc
                            })
                        })
                        .and_then(move |message: String| {
                            if !message.is_empty() {
                                Box::new(
                                    cmd_send
                                        .clone()
                                        .send(XmppCommand { xmpp_to, message })
                                        .then(|r| match r {
                                            Ok(_) => tokio::prelude::future::ok(Response::new(
                                                Body::from("Accepted"),
                                            )),
                                            Err(e) => {
                                                error!("Command sent error: {}", e);
                                                tokio::prelude::future::result(
                                                    Response::builder()
                                                        .status(hyper::StatusCode::BAD_REQUEST)
                                                        .body(Body::from(format!(
                                                            "Command sent error: {}",
                                                            e
                                                        ))),
                                                )
                                            }
                                        })
                                        .map_err(|e| {
                                            Box::new(e)
                                                as Box<
                                                    dyn std::error::Error + Sync + Send + 'static,
                                                >
                                        }),
                                )
                            } else {
                                warn!("Empty message");
                                Box::new(tokio::prelude::future::result(
                                    Response::builder()
                                        .status(hyper::StatusCode::BAD_REQUEST)
                                        .body(Body::from("Empty message"))
                                        .map_err(|e| {
                                            Box::new(e)
                                                as Box<
                                                    dyn std::error::Error + Sync + Send + 'static,
                                                >
                                        }),
                                ))
                                    as Box<Future<Item = _, Error = _> + Send + 'static>
                            }
                        }),
                ) as Box<Future<Item = _, Error = _> + Send + 'static>
            }
        }
    }
}
struct MakeServiceCmd {
    cmd_send: tokio_channel::mpsc::Sender<XmppCommand>,
}
impl<Ctx> hyper::service::MakeService<Ctx> for MakeServiceCmd {
    type ReqBody = Body;
    type ResBody = Body;
    type Error = Box<dyn std::error::Error + Sync + Send + 'static>;
    type Service = ServiceCmd;
    type Future = tokio::prelude::future::FutureResult<ServiceCmd, Self::MakeError>;
    type MakeError = hyper::http::Error;
    fn make_service(&mut self, _ctx: Ctx) -> Self::Future {
        tokio::prelude::future::ok(ServiceCmd {
            cmd_send: self.cmd_send.clone(),
        })
    }
}

Replacement in src/main.rs at line 69 [10.17]

∅:D[5.5155] → [2.5635:5679]

        .serve(MakeServiceCmd { cmd_send })

[5.5155]

[5.5935]

        .serve(move || {
            let cmd_send = cmd_send.clone();
            service_fn(move |_req: Request<Body>| {
                info!("Got request");
                cmd_send.clone().send(XmppCommand {}).then(|r| match r {
                    Ok(_) => tokio::prelude::future::ok(Response::new(Body::from("Accepted"))),
                    Err(e) => {
                        error!("Command sent error: {}", e);
                        tokio::prelude::future::result(
                            Response::builder()
                                .status(hyper::StatusCode::BAD_REQUEST)
                                .body(Body::from(format!("Command sent error: {}", e))),
                        )
                    }
                })
            })
        })