UMTLHH77LGABTVKULH6ONVSBTMSFGH3CJ6GPNTFWH73AWNJZV6LQC
FWJDW3G5KT66GHFS7IPPSFS3G5COH2BNN57GHTVT533FE7467QZAC
QWE26TMV6A5VUQLWG5ZMKHMH4NFZIQMVVXQJMPP42JUMM66C3VOQC
VS6AHRWIPIMQNQAFX7INQGSUB72OW2K6HM6NGOV76RBRF6COQFRQC
BWDUANCV77MCLCYMRS2UNFIFW4ZC3KB2KEGEUI77FRO7KW6TUZJQC
UIXIQHDY7E4DLLRQP4ATXF625QC4KGMHWOV24UEQ6WD7G2IPM4HAC
5IKA4GO7DIMBBYB7OUF5S7N2LUWIZ5MKCCCTZCAYPZI2357MVFHAC
HU3NZX5ZNTZB43SBAYMY2PUUMIF2ROJ4EFAJ6HTZXZL5T7ZJ3OQAC
ALP2YJIUN45LOOJU7GZWYDY7BMLCKR3LJVFPXTTLLZFB4LSYJKMAC
XGP44R5HBQAHBXEJKN4AB77OZ32B32FPUHSUILZDVNO4QLUOTW6QC
5A5UVGNMHEYI62XUENBEPSRJ7M3MBULYITRDOEYYA5ZDUMFBB6ZQC
let XmppConnection {
state: XmppState { client, roster },
account,
} = xmpp;
client.into_future().select2(stop_future).then(|r| match r {
Ok(Either::A(((event, client), b))) => {
if let Some(event) = event {
let xmpp = XmppConnection {
state: XmppState { client, roster },
account,
};
Box::new(xmpp.xmpp_processing(&event).then(|r| match r {
Ok(mut xmpp) => match stop_condition(&mut xmpp, event) {
let XmppConnection { state, account } = xmpp;
state
.client
.into_future()
.select2(stop_future)
.then(|r| match r {
Ok(Either::A(((event, client), b))) => {
if let Some(event) = event {
let mut xmpp = XmppConnection {
state: XmppState { client },
account,
};
xmpp.xmpp_processing(&event);
match stop_condition(&mut xmpp, event) {
Err(_e) => future::err((xmpp.account, Ok(Either::A(b)))),
},
Err(account) => future::err((account, Ok(Either::A(b)))),
})) as Box<dyn Future<Item = _, Error = _>>
} else {
Box::new(future::err((account, Ok(Either::A(b)))))
Err(e) => future::err((xmpp.account, Err(e))),
}
} else {
future::err((account, Ok(Either::A(b))))
}
}
Ok(Either::B((t, a))) => {
if let Some(client) = a.into_inner() {
future::ok(future::Loop::Break((
XmppConnection {
state: XmppState { client },
account,
},
Ok(Either::B(t)),
)))
} else {
future::err((account, Ok(Either::B(t))))
}
}
Err(Either::A((e, b))) => {
warn!("XMPP error: {}", e.0);
future::err((account, Ok(Either::A(b))))
}
Err(Either::B((e, a))) => {
if let Some(client) = a.into_inner() {
future::ok(future::Loop::Break((
XmppConnection {
state: XmppState { client },
account,
},
Err(e.into()),
)))
} else {
future::err((account, Err(e.into())))
}
}
Ok(Either::B((t, a))) => Box::new(if let Some(client) = a.into_inner() {
future::ok(future::Loop::Break((
XmppConnection {
state: XmppState { client, roster },
account,
},
Ok(Either::B(t)),
)))
} else {
future::err((account, Ok(Either::B(t))))
}),
Err(Either::A((e, b))) => {
warn!("XMPP error: {}", e.0);
Box::new(future::err((account, Ok(Either::A(b)))))
}
Err(Either::B((e, a))) => Box::new(if let Some(client) = a.into_inner() {
future::ok(future::Loop::Break((
XmppConnection {
state: XmppState { client, roster },
account,
},
Err(e),
)))
} else {
future::err((account, Err(e)))
}),
})
})
.map_err(|(account, _)| account)
.and_then(|(conn, r)| match r {
Ok(Either::A(_)) => future::ok(conn),
Ok(Either::B(_)) => future::err(conn.account),
Err(_e) => future::err(conn.account),
})
})
.then(|r| match r {
Err((account, e)) => {
error!(
"Cann't wait roster: {}",
e.err().map_or_else(
|| std::borrow::Cow::Borrowed("None"),
|e| e.to_string().into()
)
);
future::err(account)
}
Ok((conn, _)) => future::ok(conn),
.map_err(|(account, _)| account)
.and_then(|(conn, r)| match r {
Ok(Either::A(_)) => future::ok(conn),
Ok(Either::B(_)) => future::err(conn.account),
Err(_e) => future::err(conn.account),
})
})
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
.then(|r| match r {
Err((account, _e)) => {
error!("Cann't wait self-presence");
future::err(account)
}
Ok((conn, _)) => future::ok(conn),
impl<F, S> XmppState<F, S> {
fn new(cmd_recv: S, signal: F, conn: MaybeXmppConnection) -> XmppState<F, S> {
XmppState {
impl<F, S> XmppProcessState<F, S> {
fn new(cmd_recv: S, signal: F, conn: MaybeXmppConnection) -> XmppProcessState<F, S> {
XmppProcessState {
conn.connect(signal.clone())
.and_then(|conn| {
info!("xmpp connected!");
cmd_recv
.into_future()
.map_err(|_| {
error!("Got error on recv cmd");
format_err!("Receive cmd error")
})
.map(|(cmd, cmd_recv)| (cmd, cmd_recv, conn))
})
.then(|r| {
match r {
Ok((cmd, cmd_recv, conn)) => {
if let Some(_cmd) = cmd {
info!("Got cmd");
// got cmd, continue
future::ok(future::Loop::Continue(XmppState::new(
conn.connect(signal.clone()).and_then(|conn| {
info!("xmpp connected!");
conn.processing(|_, _| Ok(false), cmd_recv.into_future())
.then(|r| match r {
Ok((conn, r)) => match r {
Ok(Either::A(f)) => Box::new(if let Some(cmd_recv) = f.into_inner() {
future::ok(future::Loop::Continue(XmppProcessState {
future::ok(future::Loop::Break((None, conn.into())))
future::err(format_err!("Command receiver is gone"))
})
as Box<dyn Future<Item = _, Error = _>>,
Ok(Either::B((cmd, cmd_recv))) => {
if let Some(cmd) = cmd {
Box::new(conn.process_command(&cmd).then(|r| {
future::ok(future::Loop::Continue(XmppProcessState {
cmd_recv,
signal,
conn: match r {
Ok(conn) => conn.into(),
Err(account) => account.into(),
},
}))
}))
as Box<dyn Future<Item = _, Error = _>>
} else {
Box::new(future::ok(future::Loop::Break(())))
}
}
Err(_) => Box::new(future::err(format_err!("Command receiver is broken"))),
},
Err((account, r)) => Box::new(match r {
Ok(Either::A(f)) => {
if let Some(cmd_recv) = f.into_inner() {
future::ok(future::Loop::Continue(XmppProcessState {
cmd_recv,
signal,
conn: account.into(),
}))
} else {
future::err(format_err!("Command receiver is gone"))
}
}
Err(e) => {
// got cmd error, its bad
error!("Cmd error: {}", e);
future::err(e)
}
}
})
})
.and_then(|(opt_cmd_recv, _conn): (Option<S>, MaybeXmppConnection)| {
if let Some(cmd_recv) = opt_cmd_recv {
// process left commands
info!("Stop accepting commands");
Box::new(
cmd_recv
.for_each(|_cmd| future::ok(()))
.map_err(|_| format_err!("cmd receiver last error")),
) as Box<Future<Item = (), Error = failure::Error>>
} else {
Box::new(future::ok(()))
}
Ok(Either::B((cmd, cmd_recv))) => {
if let Some(_cmd) = cmd {
error!("Xmpp connection broken while get command");
future::ok(future::Loop::Continue(XmppProcessState {
cmd_recv,
signal,
conn: account.into(),
}))
} else {
future::ok(future::Loop::Break(()))
}
}
Err(_) => future::err(format_err!("Command receiver is broken")),
}),
})
})