Process self-presence via new processing code

[?]

Dec 30, 2018, 9:05 AM

HU3NZX5ZNTZB43SBAYMY2PUUMIF2ROJ4EFAJ6HTZXZL5T7ZJ3OQAC

Dependencies

[2] QYY3KRGL Use failure instead Box<dyn Error>
[3] AGIW6YR3 Use shared future for signal everywhere
[4] IK3YDPTY Update deps
[5] MOXHYSQ3 Use more common error type\n\nMove to failure::Error after https://github.com/rust-lang-nursery/failure/pull/283 landed.
[6] OGMBXBKP Move online to XmppConnection
[7] O2GM5J4F Don't split xmpp receiving and sending
[8] PBRUH4BJ Rename optional XmppConnection to MaybeXmppConnection
[9] NDDQQP2P Update deps
[10] QWE26TMV update deps
[11] FVVPKFTL Initial commit
[12] VS6AHRWI Move XMPP to separate dir
[13] BTOZT4JP Use failure
[14] X6L47BHQ Use different structure for established xmpp connection
[15] 3GEU7TC7 Welcome to 2018!
[16] 5A5UVGNM Move receiver closing logic out of xmpp processing
[17] TDOR5XQU Accept destination
[18] XGP44R5H Rework stopping xmpp connection
[19] H7R7Y3FQ Use new processing code to wait online
[20] 5OBTKGDL Update deps
[21] L77O4T7M Formatting and fixes
[22] FV6BJ5K6 Send self-presence and store account info in Rc so it willbe used in some future in parallel
[23] EOHEZXX3 Move request processing to structure
[24] ZI4GJ72V Add message to xmpp command
[25] ZJ4QKTJR Fixed body reading
[26] UCY2DO3D Try to read request body

Change contents

edit in src/xmpp/mod.rs at line 10

[3.73]

[3.73]

pub struct MaybeXmppConnection {
account: std::rc::Rc<config::Account>,
inner: Option<Client>,
}
replacement in src/xmpp/mod.rs at line 17

[3.74]→[3.0:77](∅→∅)

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

[3.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 29

[3.359]→[3.78:182](∅→∅)

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

[3.359]

[3.479]

impl MaybeXmppConnection {
fn new(account: config::Account) -> MaybeXmppConnection {
MaybeXmppConnection {
replacement in src/xmpp/mod.rs at line 38

[3.554]→[3.183:282](∅→∅)

/// Error shoud be !
fn connect<E: 'static>(self) -> impl Future<Item = Self, 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 48

[3.690]→[3.283:337](∅→∅)

let XmppConnection { account, inner } = self;

[3.690]

[3.744]

let MaybeXmppConnection { account, inner } = self;
replacement in src/xmpp/mod.rs at line 51

[3.782]→[3.338:500](∅→∅)

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

[3.782]

[3.944]

Box::new(future::ok(XmppConnection { account, inner }))
as Box<Future<Item = _, Error = _>>
replacement in src/xmpp/mod.rs at line 54

[3.961]→[3.501:1016](∅→∅)

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");
replacement in src/xmpp/mod.rs at line 67

[3.1477]→[3.1017:1167](∅→∅)

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

[3.1477]

[3.426]

let stop_future2 = stop_future.clone();

// 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(|_: ()| ()),
)
replacement in src/xmpp/mod.rs at line 92

[3.466]→[3.1168:1359](∅→∅)

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

[3.466]

[3.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"))
}
}),
)
edit in src/xmpp/mod.rs at line 102

[3.2890]

[3.2997]

}
replacement in src/xmpp/mod.rs at line 104

[3.2998]→[3.1360:2085](∅→∅)

/// get connection and wait for online status and set presence
/// returns error if something went wrong
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>> {
Box::new(future::loop_fn(
(sink, stream, account),
|(sink, stream, account)| {
stream.into_future().then(|r| match r {
Ok((event, stream)) => match event {
Some(Event::Online) => {
info!("Online");
future::ok(future::Loop::Break(XmppConnection {

[3.2998]

[3.3955]

impl XmppConnection {
/// base XMPP processing
fn xmpp_processing(&mut self, event: &Event) {
info!("Incoming xmpp 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>, failure::Error>),
Error = (
std::rc::Rc<config::Account>,
Result<Either<F, T>, failure::Error>,
),
>
where
F: Future<Item = T, Error = E>,
E: Into<failure::Error>,
S: FnMut(&mut Self, &Event) -> Result<bool, failure::Error>,
{
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,
replacement in src/xmpp/mod.rs at line 139

[3.3996]→[3.2086:2179](∅→∅)

inner: Some((sink, stream)),
}))

[3.3996]

[3.4505]

};
xmpp.xmpp_processing(&event);
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))),
}
} else {
future::err((account, Ok(Either::A(b))))
replacement in src/xmpp/mod.rs at line 153

[3.4531]→[3.2180:2379](∅→∅)

Some(Event::Stanza(s)) => {
info!("xmpp stanza: {:?}", s);
future::ok(future::Loop::Continue((sink, stream, account)))

[3.4531]

[3.4974]

}
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))))
replacement in src/xmpp/mod.rs at line 163

[3.5000]→[3.2380:2511](∅→∅)

_ => {
warn!("Disconnected");
future::err(account)

[3.5000]

[3.5513]

}
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.into()),
)))
} else {
future::err((account, Err(e.into())))
edit in src/xmpp/mod.rs at line 174

[3.5539]→[3.2512:2678](∅→∅)

},
Err((e, _)) => {
error!("xmpp receive error: {}", e);
future::err(account)
replacement in src/xmpp/mod.rs at line 177

[3.5595]→[3.2679:2690](∅→∅)

))

[3.5595]

[3.3349]

)
replacement in src/xmpp/mod.rs at line 180

[3.887]→[3.650:745](∅→∅),[3.3356]→[3.650:745](∅→∅)

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

[3.3356]

[3.6101]

/// get connection and wait for online status and set presence
/// returns error if something went wrong
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"))
}
}
}

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 208

[3.6155]→[3.2691:2775](∅→∅)

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

[3.6155]

[3.784]

let client = inner;
use tokio::prelude::Sink;
replacement in src/xmpp/mod.rs at line 211

[3.785]→[3.2776:4695](∅→∅),[3.7755]→[3.2608:2657](∅→∅),[3.2471]→[3.2608:2657](∅→∅),[3.3187]→[3.2608:2657](∅→∅),[3.4695]→[3.2608:2657](∅→∅),[3.3348]→[3.2608:2657](∅→∅),[3.3847]→[3.2608:2657](∅→∅),[3.1498]→[3.2608:2657](∅→∅),[3.2608]→[3.2608:2657](∅→∅),[3.2657]→[3.4696:4782](∅→∅),[3.7854]→[3.2743:2785](∅→∅),[3.2570]→[3.2743:2785](∅→∅),[3.3286]→[3.2743:2785](∅→∅),[3.4782]→[3.2743:2785](∅→∅),[3.3435]→[3.2743:2785](∅→∅),[3.3934]→[3.2743:2785](∅→∅),[3.1597]→[3.2743:2785](∅→∅),[3.2743]→[3.2743:2785](∅→∅),[3.7982]→[3.2785:2823](∅→∅),[3.2698]→[3.2785:2823](∅→∅),[3.3414]→[3.2785:2823](∅→∅),[3.2785]→[3.2785:2823](∅→∅),[3.2823]→[3.4783:5234](∅→∅)

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

[3.785]

[3.5234]

let presence = stanzas::make_presence(&account);
info!("Sending presence...");
let account2 = account.clone();
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)
&& s.attr("to").map_or(false, |f| f == conn.account.jid)
{
Ok(true)
} else {
Ok(false)
}
} else {
Err(format_err!("Wrong event while waiting self-presence"))
replacement in src/xmpp/mod.rs at line 239

[3.5260]→[3.5260:5642](∅→∅)

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

[3.3783]

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

[3.60]→[2.0:52](∅→∅)

pub struct XmppCommand {
pub xmpp_to: String,
}

[3.60]

[3.3816]

pub struct XmppCommand;
replacement in src/xmpp/mod.rs at line 259

[3.3891]→[3.5643:5669](∅→∅)

conn: XmppConnection,

[3.3891]

[3.3917]

conn: MaybeXmppConnection,
replacement in src/xmpp/mod.rs at line 263

[3.133]→[3.5670:5748](∅→∅)

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

[3.133]

[3.4038]

fn new(cmd_recv: S, signal: F, conn: MaybeXmppConnection) -> XmppState<F, S> {
replacement in src/xmpp/mod.rs at line 278

[3.4311]→[3.5749:5793](∅→∅)

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

[3.4311]

[3.465]

F: future::Future + Clone + 'static,
<F as hyper::rt::Future>::Error: std::fmt::Display + Into<failure::Error> + Send,
replacement in src/xmpp/mod.rs at line 282

[3.4357]→[3.5794:5915](∅→∅)

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

let conn = XmppConnection::new(account);

[3.4357]

[3.4526]

let conn = MaybeXmppConnection::new(account);
replacement in src/xmpp/mod.rs at line 290

[3.4692]→[3.5916:5985](∅→∅)

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 299

[3.675]→[3.5986:6042](∅→∅)

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

[3.675]

[3.731]

.map(|(cmd, cmd_recv)| (cmd, cmd_recv, conn))
})
replacement in src/xmpp/mod.rs at line 303

[3.5146]→[3.6043:6333](∅→∅),[3.9376]→[3.6169:6191](∅→∅),[3.6333]→[3.6169:6191](∅→∅),[3.6169]→[3.6169:6191](∅→∅),[3.6191]→[3.6334:7045](∅→∅)

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 = _>>

[3.5146]

[3.7045]

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 316

[3.7067]→[3.7067:7116](∅→∅)

Err(Either::B((e, _a))) => {

[3.7067]

[3.6240]

Err(e) => {
replacement in src/xmpp/mod.rs at line 319

[3.6342]→[3.7117:7253](∅→∅)

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 324

[3.900]→[3.7254:7323](∅→∅)

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

[3.900]

[3.969]

.and_then(|(opt_cmd_recv, _conn): (Option<S>, MaybeXmppConnection)| {
replacement in src/xmpp/mod.rs at line 334

[3.1354]→[3.7324:7392](∅→∅)

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

[3.1354]

[3.1422]

Box::new(future::ok(()))
edit in src/main.rs at line 22

[3.3921]

[3.3953]

use hyper::service::service_fn;
replacement in src/main.rs at line 28

[3.4066]→[2.53:97](∅→∅)

use tokio::prelude::{Future, Sink, Stream};

[3.4066]

[3.4092]

use tokio::prelude::Sink;
edit in src/main.rs at line 38

[3.4260]→[2.98:908](∅→∅),[3.219]→[3.398:399](∅→∅),[2.908]→[3.398:399](∅→∅),[3.481]→[3.398:399](∅→∅),[3.113]→[3.398:399](∅→∅),[3.398]→[3.398:399](∅→∅),[3.399]→[2.909:4701](∅→∅)

struct ServiceCmd {
cmd_send: tokio_channel::mpsc::Sender<XmppCommand>,
}

impl hyper::service::Service for ServiceCmd {
type ReqBody = Body;
type ResBody = Body;
type Error = failure::Error;
type Future =
Box<dyn Future<Item = Response<Self::ResBody>, Error = Self::Error> + Send + 'static>;

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| e.into()),
)) 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| e.into())
.fold(String::new(), |mut acc, ch| {
std::str::from_utf8(&*ch).map(|s| {
acc.push_str(s);
acc
})
})
.and_then(move |msg: String| {
if !msg.is_empty() {
Box::new(
cmd_send
.clone()
.send(XmppCommand { xmpp_to })
.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| e.into()),
)
} 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| e.into()),
))
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 = failure::Error;
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 63

[3.4954]→[3.7393:7459](∅→∅)

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

[3.4954]

[3.5020]

.map_err(|e| {
error!("Cann't get CTRL+C signal: {}", e.0);
e.0
})
replacement in src/main.rs at line 72

[3.5155]→[2.4703:4747](∅→∅)

.serve(MakeServiceCmd { cmd_send })

[3.5155]

[3.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))),
)
}
})
})
})
replacement in src/main.rs at line 100

[3.6389]→[3.7460:7607](∅→∅)

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

[3.6389]

[3.6536]

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