O01eg/sendxmppd-rs - Change PHPCAQ6ZOH64J22YO3LAHZDZ6IWRKDT4CRRC4Y2I5ADERWPBRWIQC

Use env logger. Implement command loop

Created by O01eg on October 28, 2018

PHPCAQ6ZOH64J22YO3LAHZDZ6IWRKDT4CRRC4Y2I5ADERWPBRWIQC

Dependencies

In channels

main

Change contents

Replacement in src/xmpp.rs at line 0 [3.10]

∅:D[3.10] → [3.11:39]

use tokio::prelude::future;

[3.10]

[3.39]

use tokio::prelude::future::{self, Either};
use tokio::prelude::{Future, Stream};
use tokio_channel::mpsc::Receiver;
pub struct XmppCommand;
struct XmppState<F> {
    cmd_recv: Receiver<XmppCommand>,
    signal: F,
}

Replacement in src/xmpp.rs at line 11 [3.10]

∅:D[3.40] → [3.40:134]

pub fn xmpp_process<F>(signal: F) -> impl future::Future<Item = (), Error = tokio::io::Error>

[3.40]

[3.134]

impl<F> XmppState<F> {
    fn new(cmd_recv: Receiver<XmppCommand>, signal: F) -> XmppState<F> {
        XmppState { cmd_recv, signal }
    }
}
pub fn xmpp_process<F>(
    signal: F,
    cmd_recv: Receiver<XmppCommand>,
) -> impl future::Future<Item = (), Error = tokio::io::Error>

Replacement in src/xmpp.rs at line 24 [3.10]

∅:D[3.176] → [3.176:276]

    signal.map_err(|_| tokio::io::Error::new(tokio::io::ErrorKind::Other, "Wrong shutdown signal"))

[3.176]

[3.276]

    let signal = signal
        .map_err(|_| tokio::io::Error::new(tokio::io::ErrorKind::Other, "Wrong shutdown signal"));
    future::loop_fn(XmppState::new(cmd_recv, signal), |s| {
        s.signal.select2(s.cmd_recv.into_future()).then(|r| {
            match r {
                Ok(Either::A((_x, _b))) => {
                    // got signal, breaks
                    future::ok(future::Loop::Break(()))
                }
                Ok(Either::B((x, a))) => {
                    // got cmd, continue
                    future::ok(future::Loop::Continue(XmppState::new(x.1, a)))
                }
                Err(Either::A((e, _b))) => {
                    // got signal error, breaks
                    error!("Signal error: {}", e);
                    future::ok(future::Loop::Break(()))
                }
                Err(Either::B((_e, _a))) => {
                    // got cmd error, its bad
                    future::err(tokio::io::Error::new(
                        tokio::io::ErrorKind::Other,
                        "Cmd error",
                    ))
                }
            }
        })
    })

Insertion in src/main.rs at line 1 [6.17]
[4.21]
[5.23]
```
#![deny(missing_docs)]
//! XMPP client service
```
Insertion in src/main.rs at line 13 [6.17]
[4.121]
[4.121]
```
extern crate tokio_channel;
```
Replacement in src/main.rs at line 18 [6.17]
∅:D[4.193] → [4.193:228]
∅:D[4.193] → [4.193:228]
```
use hyper::service::service_fn_ok;
```
[4.193]
[4.228]
```
use hyper::service::service_fn;
```
Insertion in src/main.rs at line 21 [6.17]
[4.275]
[4.275]
```
use tokio::runtime::current_thread;
```
Insertion in src/main.rs at line 23 [6.17]
[4.304]
[4.304]
```
use tokio::prelude::Sink;
```

Replacement in src/main.rs at line 29 [6.17]

∅:D[4.338] → [4.338:434]

fn hello_world(_req: Request<Body>) -> Response<Body> {
    Response::new(Body::from("Test"))
}

[3.280]

[4.434]

mod xmpp;
use xmpp::{xmpp_process, XmppCommand};

Insertion in src/main.rs at line 57 [6.17]

[4.1137]


    let (cmd_send, cmd_recv) = tokio_channel::mpsc::channel(10);

Replacement in src/main.rs at line 61 [6.17]

∅:D[4.1187] → [4.1187:1293]

        .serve(|| service_fn_ok(hello_world))
        .with_graceful_shutdown(ctrl_c.clone().map(|_| ()))

[4.1187]

[2.150]

        .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))),
                        )
                    }
                })
            })
        }).with_graceful_shutdown(ctrl_c.clone().map(|_| ()))

Insertion in src/main.rs at line 79 [6.17]

[2.203]

[5.176]


    let mut rt = Runtime::new().expect("Cann't start tokio");
    rt.spawn(http_server);

Replacement in src/main.rs at line 84 [6.17]

∅:D[5.177] → [2.204:237]

    hyper::rt::run(http_server);

[5.177]

[6.61]

    let xmpp_join = std::thread::spawn(move || -> Result<(), tokio::io::Error> {
        // Launch single-threaded runtime
        let mut ctrt = current_thread::Runtime::new()?;
        ctrt.block_on(xmpp_process(ctrl_c.clone().map(|_| ()), cmd_recv))
    });
    info!("Server started");
    rt.shutdown_on_idle().wait().expect("Shutdown error");
    info!("Server stopper");
    xmpp_join
        .join()
        .expect("Join xmpp thread")
        .expect("Result xmpp thread");

Insertion in Cargo.toml at line 9 [6.11995]
[4.2097]
[4.2097]
```
tokio-channel = "0.1"
```

Insertion in Cargo.lock at line 1052 [6.12201]

[4.35655]

 "tokio-channel 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",

Insertion in Cargo.lock at line 1325 [6.12201]

[4.45320]

]
[[package]]
name = "tokio-channel"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
 "futures 0.1.25 (registry+https://github.com/rust-lang/crates.io-index)",

Insertion in Cargo.lock at line 1932 [6.12201]

[4.84002]

"checksum tokio-channel 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "530d0fb87416dd531600f7cccc438bb35c5b91883065c9e6dca7cdecee991cfa"