Some servers don't send to in initial presence

[?]

Dec 31, 2018, 3:29 PM

PVCRPP3BXTLRRT2VK2BCKFCRJXUOX3AEF2A6UCPOSZYEKQEMBWDQC

Dependencies

[2] 4LRBIGVT Show info about xmpp errors
[3] HOAZX2PB Reorganize roster processing. Output roster
[4] TDOR5XQU Accept destination
[5] X6L47BHQ Use different structure for established xmpp connection
[6] ZI4GJ72V Add message to xmpp command
[7] PBRUH4BJ Rename optional XmppConnection to MaybeXmppConnection
[8] AGIW6YR3 Use shared future for signal everywhere
[9] FVVPKFTL Initial commit
[10] VS6AHRWI Move XMPP to separate dir
[11] 3GEU7TC7 Welcome to 2018!
[12] QYY3KRGL Use failure instead Box<dyn Error>
[13] BTOZT4JP Use failure
[14] HU3NZX5Z Process self-presence via new processing code
[15] OGMBXBKP Move online to XmppConnection
[16] EOHEZXX3 Move request processing to structure
[17] PFC7OJQF Query roster
[18] QWE26TMV update deps
[19] 5OBTKGDL Update deps
[20] L77O4T7M Formatting and fixes
[21] 5A5UVGNM Move receiver closing logic out of xmpp processing
[22] FV6BJ5K6 Send self-presence and store account info in Rc so it willbe used in some future in parallel
[23] IK3YDPTY Update deps
[24] XGP44R5H Rework stopping xmpp connection
[25] NDDQQP2P Update deps

Change contents

replacement in src/xmpp/mod.rs at line 38

[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.910]→[2.110:162](∅→∅)

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

[3.910]

[3.944]

as Box<Future<Item = _, Error = _>>
replacement in src/xmpp/mod.rs at line 54

[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.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.3427]→[2.1138:1190](∅→∅)

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

[3.3427]

[3.3535]

fn xmpp_processing(&mut self, event: &Event) {
info!("Incoming xmpp event: {:?}", event);
}
replacement in src/xmpp/mod.rs at line 118

[3.3776]→[2.1191:1312](∅→∅)

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

[3.3776]

[3.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.4014]→[2.1313:1358](∅→∅)

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

[3.4014]

[3.4116]

E: Into<failure::Error>,
S: FnMut(&mut Self, &Event) -> Result<bool, failure::Error>,
replacement in src/xmpp/mod.rs at line 141

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

[3.4701]

[3.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.5802]→[2.1648:1837](∅→∅)

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

[3.5802]

[3.5891]

Err(Either::A((_e, b))) => future::err((account, Ok(Either::A(b)))),
replacement in src/xmpp/mod.rs at line 169

[3.6129]→[2.1838:1878](∅→∅)

Err(e),

[3.6129]

[3.6176]

Err(e.into()),
replacement in src/xmpp/mod.rs at line 172

[3.6241]→[2.1879:1938](∅→∅)

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

[3.6241]

[3.5513]

future::err((account, Err(e.into())))
replacement in src/xmpp/mod.rs at line 182

[3.6432]→[2.1939:2924](∅→∅),[2.2924]→[3.1664:1690](∅→∅),[3.1664]→[3.1664:1690](∅→∅),[3.1690]→[2.2925:3091](∅→∅),[2.3091]→[3.1784:1806](∅→∅),[3.1784]→[3.1784:1806](∅→∅),[3.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)
}
})
},
))

[3.6432]

[3.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.3694]→[2.3138:3233](∅→∅)

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

[3.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.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 }),
}),
)

[3.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.4311]→[2.5693:5737](∅→∅)

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,
edit in src/xmpp/mod.rs at line 281

[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.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.675]→[2.5885:5941](∅→∅)

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

[3.675]

[3.731]

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

[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

[2.6261]→[2.6261:6986](∅→∅),[2.6986]→[3.7045:7067](∅→∅),[3.9763]→[3.7045:7067](∅→∅),[3.7045]→[3.7045:7067](∅→∅),[3.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.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.1354]→[2.7174:7242](∅→∅)

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

[3.1354]

[3.1422]

Box::new(future::ok(()))
replacement in src/main.rs at line 63

[3.4954]→[2.7243:7309](∅→∅)

.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 100

[3.6389]→[2.7310:7457](∅→∅)

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