Add rustdoc

[?]

Apr 21, 2019, 3:38 PM

3DSOPLCGIJFUUYWVQUQW5ESFRJRB2VLZC6VCJOJES3IJ6TQB6ZTAC

Dependencies

[2] YEMBT7TB Add support for XEP-0092: Software Version
[3] ALP2YJIU Rename XmppState to XmppProcessState
[4] S754Y5DF Refactor IQ processing Always answer to set and get requests. Use XML encoding for stanzas.
[5] BYJPYYSM Process iq ping response
[6] QDHDTOLM Starting support for commands XEP-0050: Ad-Hoc Commands (there no support in xmpp_parsers still)
[7] WDCZNZOP Fix rustdoc
[8] RQZCVDFD Implement applying timeout for expired iq await
[9] HU3NZX5Z Process self-presence via new processing code
[10] ZFBPXPAD Cleanup timeouted iq requests with ping Output elapsed time. Refactor iq handling.
[11] DYRPAV6T Update dependencies
[12] 4LRBIGVT Show info about xmpp errors
[13] GXQCDLYQ Use element processor for incoming iq get
[14] HDLI2X4H Ignore delayed XEP-0203 messages
[15] CCLGGFKR Move out XmppConnection into own file
[16] 2THKW66M Ignore .orig files
[17] VS6AHRWI Move XMPP to separate dir
[18] X6L47BHQ Use different structure for established xmpp connection
[19] Z3NQEYVI Rename IqSetHandler to IqResuestHandler as it should provide both get and set handling
[20] DCMDASHV Mention XEP-0050 and XEP-0203 support
[21] FV6BJ5K6 Send self-presence and store account info in Rc so it willbe used in some future in parallel
[22] LL3D5CXK Staring using element processor
[23] QWE26TMV update deps
[24] PFC7OJQF Query roster
[25] 4IPZTMFI Update dependencies
[26] JD62RVOJ Update dependencies
[27] PLWPCM47 Add id to initital presence
[28] 5A5UVGNM Move receiver closing logic out of xmpp processing
[29] TPVUBB3F Answer to ping requests
[30] QTCUURXN Add additional requirement for command stream
[31] FCPF2FV6 Break connection on iq error
[32] OFLAP2G2 Fix possible utf8 errors
[33] JY4F7VBC Use element processor for incoming iq set
[34] V5HDBSZM Use jid for receiver address
[35] 3GEU7TC7 Welcome to 2018!
[36] PJV5HPIF Starting to imlements timeouts for iqs
[37] RRLRZTMR Use element processor for iq
[38] 6UKCVM6E Use new iq processng for initial roster
[39] NDDQQP2P Update deps
[40] LNUU5R56 Support disco#info from XEP-0030 Service Discovery
[41] GVZ4JAR5 Process self-presence with incoming stanza processor
[42] BWDUANCV Second part of processing result is only about stop_future
[43] 37OMJ4CK Send MUC message
[44] AYQZ2UIA Update deps
[45] AEH7WP42 Make element processors static
[46] 5IKA4GO7 Rename xmpp client field from "inner" to "client"
[47] DCGEFPRC Better README
[48] OANBCLN5 Move xmpp client into XmppState
[49] FVVPKFTL Initial commit
[50] MAC6WCSX Fix pipelines
[51] 6E5IC33Z Try to test 2018 edition
[52] TDOR5XQU Accept destination
[53] RGOSS73U Convert self-presence to xmpp_parser's type
[54] OB3HA2MD Use Client::new_with_jid to parse jid only once
[55] YTN366WA Support disco#items
[56] LQXBWNFT Remove unneeded requirement
[57] ZT3YEIVX Consume connection on processing command

Change contents

replacement in src/xmpp/xmpp_connection.rs at line 0

[3.21]→[2.0:38895](∅→∅)

use tokio_xmpp::{Client, Event, Packet};

use tokio::prelude::future::{self, Either};
use tokio::prelude::stream;
use tokio::prelude::{Future, Stream};

use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant};

use super::stanzas;

use super::element_processor;

use crate::config;

#[derive(Debug)]
pub enum XmppCommand {
/// Send message to someone by jid
Chat {
xmpp_to: xmpp_parsers::Jid,
message: String,
},
/// Send message to MUC
Chatroom { muc_id: String, message: String },
/// Send ping request to the server to test connection
Ping,
/// Check iq requests if some have expired timeouts
TimeoutCleanup,
}

/// trait of processing iq
/// each function consumes handlers and
/// returns false if connection should be reset
trait IqHandler {
/// process result
fn result(
self: Box<Self>,
conn: &mut XmppConnection,
opt_element: Option<xmpp_parsers::Element>,
) -> bool;

/// process error
fn error(
self: Box<Self>,
conn: &mut XmppConnection,
error: xmpp_parsers::stanza_error::StanzaError,
) -> bool;

/// process tmeout
fn timeout(self: Box<Self>, conn: &mut XmppConnection) -> bool;
}

struct AddRosterIqHandler {
jid: xmpp_parsers::Jid,
}

impl IqHandler for AddRosterIqHandler {
fn result(
self: Box<Self>,
conn: &mut XmppConnection,
opt_element: Option<xmpp_parsers::Element>,
) -> bool {
match opt_element {
Some(element) => {
warn!(
"Wrong payload when adding {} to roster: {}",
self.jid,
String::from(&element)
);
}
None => {
if conn.state.data.roster.contains_key(&self.jid) {
info!("Jid {} updated to roster", self.jid);
} else {
info!("Jid {} added in roster", self.jid);
conn.state.data.roster.insert(
self.jid.clone(),
(
xmpp_parsers::roster::Subscription::None,
xmpp_parsers::roster::Ask::None,
),
);
}
conn.process_jid(&self.jid);
}
}
true
}

fn error(
self: Box<Self>,
_conn: &mut XmppConnection,
_error: xmpp_parsers::stanza_error::StanzaError,
) -> bool {
true
}

fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
true // ignore
}
}

struct PingIqHandler {}

impl IqHandler for PingIqHandler {
fn result(
self: Box<Self>,
_conn: &mut XmppConnection,
_opt_element: Option<xmpp_parsers::Element>,
) -> bool {
info!("ping successed");
true
}

fn error(
self: Box<Self>,
_conn: &mut XmppConnection,
_error: xmpp_parsers::stanza_error::StanzaError,
) -> bool {
false
}

fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
false
}
}

struct InitRosterIqHandler {}

impl IqHandler for InitRosterIqHandler {
fn result(
self: Box<Self>,
conn: &mut XmppConnection,
opt_element: Option<xmpp_parsers::Element>,
) -> bool {
if let Some(result) = opt_element {
use std::convert::TryInto;
match result.try_into() as Result<xmpp_parsers::roster::Roster, _> {
Ok(roster) => {
conn.state.data.roster_init = true;
conn.state.data.roster.clear();
info!("Got first roster:");
for i in roster.items {
info!(" >>> {:?}", i);
conn.state
.data
.roster
.insert(i.jid, (i.subscription, i.ask));
}
true
}
Err(e) => {
error!("Cann't parse roster: {}", e);
false
}
}
} else {
error!("No roster responded");
false
}
}

fn error(
self: Box<Self>,
_conn: &mut XmppConnection,
_error: xmpp_parsers::stanza_error::StanzaError,
) -> bool {
false
}

fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
false
}
}

#[derive(Default)]
struct XmppData {
/// known roster data
roster: HashMap<
xmpp_parsers::Jid,
(
xmpp_parsers::roster::Subscription,
xmpp_parsers::roster::Ask,
),
>,
/// if roster was initialized
/// ToDo: remove it as it is used only for initialization
roster_init: bool,
/// if self-presence accepted
/// ToDo: remove it as it is used only for initialization
self_presence: bool,
/// ids counter
counter: usize,
/// stanzas to send
send_queue: VecDeque<minidom::Element>,
/// outgoing mailbox
outgoing_mailbox: HashMap<xmpp_parsers::Jid, Vec<String>>,
/// muc id to muc jid
mucs: HashMap<String, xmpp_parsers::Jid>,
/// map from iq's id to handler of this type of iqs
pending_ids: HashMap<String, (Instant, Box<dyn IqHandler>)>,
}

struct XmppState {
client: Client,
data: XmppData,
}

pub struct XmppConnection {
account: std::rc::Rc<config::Account>,
state: XmppState,
}

trait IqRequestHandler {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq;
}

struct IqRequestUnknown {
element: xmpp_parsers::Element,
type_: &'static str,
}

impl IqRequestHandler for IqRequestUnknown {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
warn!(
"Unsupported IQ {} request from {:?}: {}",
self.type_,
from,
String::from(&self.element)
);
stanzas::make_iq_unsupported_error(id, conn.state.client.jid.clone(), from)
}
}

struct IqSetRoster {}

impl IqRequestHandler for IqSetRoster {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
info!("Got roster push {} from {:?}", id, from);
stanzas::make_roster_push_answer(id, conn.state.client.jid.clone(), from)
}
}

struct IqGetPing {}

impl IqRequestHandler for IqGetPing {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
info!("Got ping {} from {:?}", id, from);
stanzas::make_pong(id, conn.state.client.jid.clone(), from)
}
}

struct IqGetVersion {}

impl IqRequestHandler for IqGetVersion {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
info!("Got version query {} from {:?}", id, from);
stanzas::make_version(id, conn.state.client.jid.clone(), from)
}
}

struct IqGetDiscoInfo {}

impl IqRequestHandler for IqGetDiscoInfo {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
info!("Got disco query {} from {:?}", id, from);
stanzas::make_disco_info_result(id, conn.state.client.jid.clone(), from)
}
}

struct IqGetDiscoItems {}

impl IqRequestHandler for IqGetDiscoItems {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
info!("Got disco items query {} from {:?}", id, from);
stanzas::make_disco_items_result(id, conn.state.client.jid.clone(), from)
}
}

lazy_static! {
static ref INCOMING: element_processor::Processor<XmppConnection, bool, xmpp_parsers::Element> = {
let mut incoming = element_processor::Processor::new(&|_, e| {
warn!("Unknown stanza {}", String::from(&e));
true
});
incoming.register(&XmppConnection::incoming_iq_processing);
incoming.register(&XmppConnection::incoming_presence_processing);
incoming.register(&XmppConnection::incoming_message_processing);
incoming
};
static ref INCOMING_IQ_SET: element_processor::Processor<XmppConnection, Box<dyn IqRequestHandler>, xmpp_parsers::Element> = {
let mut iq_set =
element_processor::Processor::new(&|_conn: &mut XmppConnection, element| {
Box::new(IqRequestUnknown {
element,
type_: "set",
}) as Box<dyn IqRequestHandler>
});
iq_set.register(&XmppConnection::incoming_iq_processing_set_roster);
iq_set
};
static ref INCOMING_IQ_GET: element_processor::Processor<XmppConnection, Box<dyn IqRequestHandler>, xmpp_parsers::Element> = {
let mut iq_get =
element_processor::Processor::new(&|_conn: &mut XmppConnection, element| {
Box::new(IqRequestUnknown {
element,
type_: "get",
}) as Box<dyn IqRequestHandler>
});
iq_get.register(&XmppConnection::incoming_iq_processing_get_ping);
iq_get.register(&XmppConnection::incoming_iq_processing_get_disco_info);
iq_get.register(&XmppConnection::incoming_iq_processing_get_disco_items);
iq_get.register(&XmppConnection::incoming_iq_processing_get_version);
iq_get
};
}

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

impl From<XmppConnection> for MaybeXmppConnection {
fn from(from: XmppConnection) -> MaybeXmppConnection {
MaybeXmppConnection {
account: from.account,
state: Some(from.state),
}
}
}

impl From<config::Account> for MaybeXmppConnection {
fn from(from: config::Account) -> MaybeXmppConnection {
MaybeXmppConnection {
account: std::rc::Rc::new(from),
state: None,
}
}
}

impl From<std::rc::Rc<config::Account>> for MaybeXmppConnection {
fn from(from: std::rc::Rc<config::Account>) -> MaybeXmppConnection {
MaybeXmppConnection {
account: from,
state: None,
}
}
}

impl MaybeXmppConnection {
/// connects if nothing connected
/// don't connect only if stop_future resolved
pub 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,
{
info!("xmpp connection...");
let MaybeXmppConnection { account, state } = self;

if let Some(state) = state {
Box::new(future::ok(XmppConnection { account, state }))
as Box<dyn Future<Item = _, Error = _>>
} else {
Box::new(
stop_future
.clone()
.select2(
future::loop_fn(account, move |account| {
info!("xmpp initialization...");
let client =
Client::new_with_jid(account.jid.clone(), &account.password);
info!("xmpp initialized");

let stop_future2 = stop_future.clone();
let stop_future3 = stop_future.clone();
let stop_future4 = stop_future.clone();

// future to wait for online
Box::new(
XmppConnection {
state: XmppState {
client,
data: std::default::Default::default(),
},
account,
}
.processing(XmppConnection::online, stop_future.clone())
.map_err(|(acc, _)| acc)
.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),
})
.and_then(|conn| conn.initial_roster(stop_future2))
.and_then(|conn| conn.self_presence(stop_future3))
.and_then(|conn| conn.enter_mucs(stop_future4))
.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"))
}
}),
)
}
}
}

impl XmppConnection {
/// base XMPP processing
/// Returns false on error to disconnect
fn xmpp_processing(&mut self, event: &Event) -> bool {
match event {
Event::Stanza(stanza) => INCOMING.process(self, stanza.clone()),
Event::Online => true,
e => {
warn!("Unexpected event {:?}", e);
false
}
}
}

/// Process roster push
/// see RFC 6212 2.1.6. Roster Push
fn incoming_iq_processing_set_roster(
&mut self,
roster: xmpp_parsers::roster::Roster,
) -> Box<dyn IqRequestHandler> {
for i in roster.items {
if let Some(ref mut rdata) = self.state.data.roster.get_mut(&i.jid) {
info!("Update {} in roster", i.jid);
rdata.0 = i.subscription;
rdata.1 = i.ask;
} else {
info!("Add {} to roster", i.jid);
self.state
.data
.roster
.insert(i.jid.clone(), (i.subscription, i.ask));
}
self.process_jid(&i.jid);
}
Box::new(IqSetRoster {})
}

/// Enforce to answer to IQ "set"
fn incoming_iq_processing_set(
&mut self,
id: String,
from: Option<xmpp_parsers::Jid>,
element: minidom::Element,
) -> xmpp_parsers::iq::Iq {
INCOMING_IQ_SET
.process(self, element)
.process(self, id, from)
}

/// Process ping request
/// see XEP-0199: XMPP Ping
fn incoming_iq_processing_get_ping(
&mut self,
_ping: xmpp_parsers::ping::Ping,
) -> Box<dyn IqRequestHandler> {
Box::new(IqGetPing {})
}

/// Process disco query
/// see XEP-0030: Service Discovery
fn incoming_iq_processing_get_disco_info(
&mut self,
disco: xmpp_parsers::disco::DiscoInfoQuery,
) -> Box<dyn IqRequestHandler> {
if let Some(ref node) = disco.node {
warn!("Unsupported node {}", node);
Box::new(IqRequestUnknown {
element: disco.into(),
type_: "get",
})
} else {
Box::new(IqGetDiscoInfo {})
}
}

/// Process disco items query
/// see XEP-0030: Service Discovery
fn incoming_iq_processing_get_disco_items(
&mut self,
disco: xmpp_parsers::disco::DiscoItemsQuery,
) -> Box<dyn IqRequestHandler> {
if let Some(ref node) = disco.node {
warn!("Unsupported node {}", node);
Box::new(IqRequestUnknown {
element: disco.into(),
type_: "get",
})
} else {
Box::new(IqGetDiscoItems {})
}
}

/// Process version query
/// see XEP-0092: Software Version
fn incoming_iq_processing_get_version(
&mut self,
_version: xmpp_parsers::version::VersionQuery,
) -> Box<dyn IqRequestHandler> {
Box::new(IqGetVersion {})
}

/// Enforce to answer to IQ "get"
fn incoming_iq_processing_get(
&mut self,
id: String,
from: Option<xmpp_parsers::Jid>,
element: minidom::Element,
) -> xmpp_parsers::iq::Iq {
INCOMING_IQ_GET
.process(self, element)
.process(self, id, from)
}

fn incoming_iq_processing(&mut self, iq: xmpp_parsers::iq::Iq) -> bool {
match iq.payload {
xmpp_parsers::iq::IqType::Set(element) => {
let iq_answer = self.incoming_iq_processing_set(iq.id, iq.from, element);
self.state.data.send_queue.push_back(iq_answer.into());
}
xmpp_parsers::iq::IqType::Error(e) => {
if let Some((_, handler)) = self.state.data.pending_ids.remove_entry(&iq.id) {
return handler.1.error(self, e);
}
error!("iq error: {:?}", e);
return false;
}
xmpp_parsers::iq::IqType::Get(element) => {
let iq_answer = self.incoming_iq_processing_get(iq.id, iq.from, element);
self.state.data.send_queue.push_back(iq_answer.into());
}
xmpp_parsers::iq::IqType::Result(opt_element) => {
if let Some((_, handler)) = self.state.data.pending_ids.remove_entry(&iq.id) {
return handler.1.result(self, opt_element);
}
warn!(
"Unwanted iq result id {} from {:?}: {:?}",
iq.id,
iq.from,
opt_element.map(|e| String::from(&e))
);
}
}
true
}

fn incoming_presence_processing(&mut self, presence: xmpp_parsers::presence::Presence) -> bool {
if presence.from.as_ref() == Some(&self.state.client.jid) {
info!("Self-presence accepted");
self.state.data.self_presence = true;
} else {
warn!("Incoming presence stanza: {:?}", presence);
}
true
}

fn incoming_message_processing(&mut self, message: xmpp_parsers::message::Message) -> bool {
for payload in message.payloads.iter() {
use std::convert::TryInto;
if let Some(_delay) =
payload.clone().try_into().ok() as Option<xmpp_parsers::delay::Delay>
{
return true; // ignore delayed messages
}
}
warn!("Incoming message stanza: {:?}", message);
true
}

/// process event from xmpp stream
/// returns from future when condition met
/// or stop future was resolved.
/// Return item if connection was preserved or error otherwise.
/// Second part is a state of stop_future
pub fn processing<S, F, T, E>(
self,
stop_condition: S,
stop_future: F,
) -> impl Future<
Item = (Self, Result<Either<F, T>, E>),
Error = (std::rc::Rc<config::Account>, Result<Either<F, T>, E>),
>
where
F: Future<Item = T, Error = E> + 'static,
S: FnMut(&mut Self, Event) -> Result<bool, ()> + 'static,
T: 'static,
E: 'static,
{
future::loop_fn(
(self, stop_future, stop_condition),
|(xmpp, stop_future, mut stop_condition)| {
// ToDo: check timeouts if iqs

let XmppConnection {
state: XmppState { client, mut data },
account,
} = xmpp;

if let Some(send_element) = data.send_queue.pop_front() {
use tokio::prelude::Sink;

info!("Sending {}", String::from(&send_element));
Box::new(
client
.send(Packet::Stanza(send_element))
.select2(stop_future)
.then(move |r| match r {
Ok(Either::A((client, b))) => {
Box::new(future::ok(future::Loop::Continue((
XmppConnection {
state: XmppState { client, data },
account,
},
b,
stop_condition,
))))
as Box<dyn Future<Item = _, Error = _>>
}
Ok(Either::B((t, a))) => Box::new(a.then(|r| match r {
Ok(client) => future::ok(future::Loop::Break((
XmppConnection {
state: XmppState { client, data },
account,
},
Ok(Either::B(t)),
))),
Err(se) => {
warn!("XMPP sending error: {}", se);
future::err((account, Ok(Either::B(t))))
}
})),
Err(Either::A((e, b))) => {
warn!("XMPP sending error: {}", e);
Box::new(future::err((account, Ok(Either::A(b)))))
}
Err(Either::B((e, a))) => Box::new(a.then(|r| match r {
Ok(client) => future::ok(future::Loop::Break((
XmppConnection {
state: XmppState { client, data },
account,
},
Err(e),
))),
Err(se) => {
warn!("XMPP sending error: {}", se);
future::err((account, Err(e)))
}
})),
}),
) as Box<dyn Future<Item = _, Error = _>>
} else {
Box::new(
client
.into_future()
.select2(stop_future)
.then(move |r| match r {
Ok(Either::A(((event, client), b))) => {
if let Some(event) = event {
let mut xmpp = XmppConnection {
state: XmppState { client, data },
account,
};

if 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, Ok(Either::A(b))))
}
}
} else {
future::err((xmpp.account, Ok(Either::A(b))))
}
} 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, data },
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, data },
account,
},
Err(e),
)))
} else {
future::err((account, Err(e)))
}
}
}),
)
}
},
)
}

/// get connection and wait for online status and set presence
/// returns error if something went wrong and xmpp connection is broken
fn online(&mut self, event: Event) -> Result<bool, ()> {
match event {
Event::Online => {
info!("Online!");
Ok(true)
}
Event::Stanza(s) => {
warn!("Stanza before online: {}", String::from(&s));
Ok(false)
}
_ => {
error!("Disconnected while online");
Err(())
}
}
}

fn initial_roster<F, E>(
self,
stop_future: F,
) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
where
F: Future<Error = E> + 'static,
E: 'static,
{
let XmppConnection {
account,
state: XmppState { client, mut data },
} = self;
use tokio::prelude::Sink;

data.counter += 1;
let id_init_roster = format!("id_init_roster{}", data.counter);
let get_roster = stanzas::make_get_roster(&id_init_roster);
let account2 = account.clone();
info!("Quering roster... {}", String::from(&get_roster));
data.pending_ids.insert(
id_init_roster.clone(),
(
Instant::now() + Duration::from_secs(60),
Box::new(InitRosterIqHandler {}),
),
);

client
.send(Packet::Stanza(get_roster))
.map_err(move |e| {
error!("Error on querying roster: {}", e);
account2
})
.and_then(move |client| {
XmppConnection {
state: XmppState { client, data },
account,
}
.processing(move |conn, _| Ok(conn.state.data.roster_init), stop_future)
.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),
})
})
}

fn self_presence<F, E>(
self,
stop_future: F,
) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
where
F: Future<Error = E> + 'static,
E: Into<failure::Error> + 'static,
{
let XmppConnection {
account,
state: XmppState { client, data },
} = self;
use tokio::prelude::Sink;

let presence = stanzas::make_presence(&account);
let account2 = account.clone();
info!("Sending presence... {}", String::from(&presence));

client
.send(Packet::Stanza(presence))
.map_err(|e| {
error!("Error on send self-presence: {}", e);
account2
})
.and_then(move |client| {
XmppConnection {
state: XmppState { client, data },
account,
}
.processing(
move |conn, _| Ok(conn.state.data.self_presence),
stop_future,
)
.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),
})
})
}

fn process_jid(&mut self, xmpp_to: &xmpp_parsers::Jid) {
if let Some(ref mut mailbox) = self.state.data.outgoing_mailbox.get_mut(xmpp_to) {
if !mailbox.is_empty() {
if let Some(ref mut rdata) = self.state.data.roster.get_mut(xmpp_to) {
info!("Jid {} in roster", xmpp_to);
let sub_to = match rdata.0 {
xmpp_parsers::roster::Subscription::To => true,
xmpp_parsers::roster::Subscription::Both => true,
_ => false,
};
if sub_to {
info!("Subscribed to {}", xmpp_to);
self.state.data.send_queue.extend(
mailbox.drain(..).map(|message| {
stanzas::make_chat_message(xmpp_to.clone(), message)
}),
);
} else if rdata.1 == xmpp_parsers::roster::Ask::None {
info!("Not subscribed to {}", xmpp_to);
self.state
.data
.send_queue
.push_back(stanzas::make_ask_subscribe(xmpp_to.clone()));
}
let sub_from = match rdata.0 {
xmpp_parsers::roster::Subscription::From => true,
xmpp_parsers::roster::Subscription::Both => true,
_ => false,
};
if !sub_from {
info!("Not subscription from {}", xmpp_to);
self.state
.data
.send_queue
.push_back(stanzas::make_allow_subscribe(xmpp_to.clone()));
}
} else {
info!("Jid {} not in roster", xmpp_to);

self.state.data.counter += 1;
let id_add_roster = format!("id_add_roster{}", self.state.data.counter);
let add_roster = stanzas::make_add_roster(&id_add_roster, xmpp_to.clone());
info!("Adding jid {} to roster id {}", xmpp_to, id_add_roster);
self.state.data.pending_ids.insert(
id_add_roster,
(
Instant::now() + Duration::from_secs(60),
Box::new(AddRosterIqHandler {
jid: xmpp_to.clone(),
}),
),
);
self.state.data.send_queue.push_back(add_roster);
}
}
}
}

pub fn process_command(&mut self, cmd: XmppCommand) {
info!("Got command");
match cmd {
XmppCommand::Chat { xmpp_to, message } => {
self.state
.data
.outgoing_mailbox
.entry(xmpp_to.clone())
.or_default()
.push(message);
self.process_jid(&xmpp_to);
}
XmppCommand::Chatroom { muc_id, message } => {
if let Some(muc) = self.state.data.mucs.get(&muc_id) {
self.state
.data
.send_queue
.push_back(stanzas::make_muc_message(muc.clone(), message));
} else {
error!("Not found MUC {}", muc_id);
}
}
XmppCommand::Ping => {
self.state.data.counter += 1;
let id_ping = format!("id_ping{}", self.state.data.counter);
let ping = stanzas::make_ping(&id_ping, self.state.client.jid.clone());
self.state.data.send_queue.push_back(ping);
self.state.data.pending_ids.insert(
id_ping,
(
Instant::now() + Duration::from_secs(30),
Box::new(PingIqHandler {}),
),
);
}
XmppCommand::TimeoutCleanup => {
let now = Instant::now();
let timeouted: Vec<String> = self
.state
.data
.pending_ids
.iter()
.filter_map(|(id, (timeout, _))| {
if now >= *timeout {
Some(id.to_string())
} else {
None
}
})
.collect();
let mut correct = true;
timeouted.into_iter().for_each(|id| {
if let Some((_, handler)) = self.state.data.pending_ids.remove(&id) {
correct &= handler.timeout(&mut self);
}
})
}
}
}

pub fn shutdown(self) -> impl Future<Item = (), Error = failure::Error> {
info!("Shutdown connection");
let XmppConnection { account, state } = self;
stream::iter_ok(
state
.data
.mucs
.values()
.map(std::clone::Clone::clone)
.collect::<Vec<_>>(),
)
.fold(state, move |XmppState { client, data }, muc_jid| {
let muc_presence =
stanzas::make_muc_presence_leave(account.jid.clone(), muc_jid.clone());

info!(
"Sending muc leave presence... {}",
String::from(&muc_presence)
);
use tokio::prelude::Sink;
client
.send(Packet::Stanza(muc_presence))
.map_err(|e| {
error!("Error on send muc presence: {}", e);
e
})
.and_then(|client| future::ok(XmppState { client, data }))
})
.map(|_| ())
}

fn enter_mucs<F, E>(
self,
_stop_future: F,
) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
where
F: Future<Error = E> + 'static,
E: Into<failure::Error> + 'static,
{
let XmppConnection { account, state } = self;

let account2 = account.clone();
let account3 = account.clone();

stream::iter_ok(account.chatrooms.clone())
.fold(state, move |XmppState { client, mut data }, muc_jid| {
data.counter += 1;
let id_muc_presence = format!("id_muc_presence{}", data.counter);
let muc_presence = stanzas::make_muc_presence(
&id_muc_presence,
account2.jid.clone(),
muc_jid.1.clone(),
);

info!("Sending muc presence... {}", String::from(&muc_presence));

let account4 = account2.clone();

use tokio::prelude::Sink;
client
.send(Packet::Stanza(muc_presence))
.map_err(|e| {
error!("Error on send muc presence: {}", e);
account4
})
.and_then(|client| {
data.mucs.insert(muc_jid.0, muc_jid.1);
future::ok(XmppState { client, data })
})
})
.map(|state| XmppConnection {
account: account3,
state,
})
}
}

[3.21]

use tokio_xmpp::{Client, Event, Packet};

use tokio::prelude::future::{self, Either};
use tokio::prelude::stream;
use tokio::prelude::{Future, Stream};

use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant};

use super::stanzas;

use super::element_processor;

use crate::config;

#[derive(Debug)]
pub enum XmppCommand {
/// Send message to someone by jid
Chat {
xmpp_to: xmpp_parsers::Jid,
message: String,
},
/// Send message to MUC
Chatroom { muc_id: String, message: String },
/// Send ping request to the server to test connection
Ping,
/// Check iq requests if some have expired timeouts
TimeoutCleanup,
}

/// trait of processing iq
/// each function consumes handlers and
/// returns false if connection should be reset
trait IqHandler {
/// process result
fn result(
self: Box<Self>,
conn: &mut XmppConnection,
opt_element: Option<xmpp_parsers::Element>,
) -> bool;

/// process error
fn error(
self: Box<Self>,
conn: &mut XmppConnection,
error: xmpp_parsers::stanza_error::StanzaError,
) -> bool;

/// process tmeout
fn timeout(self: Box<Self>, conn: &mut XmppConnection) -> bool;
}

struct AddRosterIqHandler {
jid: xmpp_parsers::Jid,
}

impl IqHandler for AddRosterIqHandler {
fn result(
self: Box<Self>,
conn: &mut XmppConnection,
opt_element: Option<xmpp_parsers::Element>,
) -> bool {
match opt_element {
Some(element) => {
warn!(
"Wrong payload when adding {} to roster: {}",
self.jid,
String::from(&element)
);
}
None => {
if conn.state.data.roster.contains_key(&self.jid) {
info!("Jid {} updated to roster", self.jid);
} else {
info!("Jid {} added in roster", self.jid);
conn.state.data.roster.insert(
self.jid.clone(),
(
xmpp_parsers::roster::Subscription::None,
xmpp_parsers::roster::Ask::None,
),
);
}
conn.process_jid(&self.jid);
}
}
true
}

fn error(
self: Box<Self>,
_conn: &mut XmppConnection,
_error: xmpp_parsers::stanza_error::StanzaError,
) -> bool {
true
}

fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
true // ignore
}
}

struct PingIqHandler {}

impl IqHandler for PingIqHandler {
fn result(
self: Box<Self>,
_conn: &mut XmppConnection,
_opt_element: Option<xmpp_parsers::Element>,
) -> bool {
info!("ping successed");
true
}

fn error(
self: Box<Self>,
_conn: &mut XmppConnection,
_error: xmpp_parsers::stanza_error::StanzaError,
) -> bool {
false
}

fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
false
}
}

struct InitRosterIqHandler {}

impl IqHandler for InitRosterIqHandler {
fn result(
self: Box<Self>,
conn: &mut XmppConnection,
opt_element: Option<xmpp_parsers::Element>,
) -> bool {
if let Some(result) = opt_element {
use std::convert::TryInto;
match result.try_into() as Result<xmpp_parsers::roster::Roster, _> {
Ok(roster) => {
conn.state.data.roster_init = true;
conn.state.data.roster.clear();
info!("Got first roster:");
for i in roster.items {
info!(" >>> {:?}", i);
conn.state
.data
.roster
.insert(i.jid, (i.subscription, i.ask));
}
true
}
Err(e) => {
error!("Cann't parse roster: {}", e);
false
}
}
} else {
error!("No roster responded");
false
}
}

fn error(
self: Box<Self>,
_conn: &mut XmppConnection,
_error: xmpp_parsers::stanza_error::StanzaError,
) -> bool {
false
}

fn timeout(self: Box<Self>, _conn: &mut XmppConnection) -> bool {
false
}
}

#[derive(Default)]
struct XmppData {
/// known roster data
roster: HashMap<
xmpp_parsers::Jid,
(
xmpp_parsers::roster::Subscription,
xmpp_parsers::roster::Ask,
),
>,
/// if roster was initialized
/// ToDo: remove it as it is used only for initialization
roster_init: bool,
/// if self-presence accepted
/// ToDo: remove it as it is used only for initialization
self_presence: bool,
/// ids counter
counter: usize,
/// stanzas to send
send_queue: VecDeque<minidom::Element>,
/// outgoing mailbox
outgoing_mailbox: HashMap<xmpp_parsers::Jid, Vec<String>>,
/// muc id to muc jid
mucs: HashMap<String, xmpp_parsers::Jid>,
/// map from iq's id to handler of this type of iqs
pending_ids: HashMap<String, (Instant, Box<dyn IqHandler>)>,
}

struct XmppState {
client: Client,
data: XmppData,
}

pub struct XmppConnection {
account: std::rc::Rc<config::Account>,
state: XmppState,
}

trait IqRequestHandler {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq;
}

struct IqRequestUnknown {
element: xmpp_parsers::Element,
type_: &'static str,
}

impl IqRequestHandler for IqRequestUnknown {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
warn!(
"Unsupported IQ {} request from {:?}: {}",
self.type_,
from,
String::from(&self.element)
);
stanzas::make_iq_unsupported_error(id, conn.state.client.jid.clone(), from)
}
}

struct IqSetRoster {}

impl IqRequestHandler for IqSetRoster {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
info!("Got roster push {} from {:?}", id, from);
stanzas::make_roster_push_answer(id, conn.state.client.jid.clone(), from)
}
}

struct IqGetPing {}

impl IqRequestHandler for IqGetPing {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
info!("Got ping {} from {:?}", id, from);
stanzas::make_pong(id, conn.state.client.jid.clone(), from)
}
}

struct IqGetVersion {}

impl IqRequestHandler for IqGetVersion {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
info!("Got version query {} from {:?}", id, from);
stanzas::make_version(id, conn.state.client.jid.clone(), from)
}
}

struct IqGetDiscoInfo {}

impl IqRequestHandler for IqGetDiscoInfo {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
info!("Got disco query {} from {:?}", id, from);
stanzas::make_disco_info_result(id, conn.state.client.jid.clone(), from)
}
}

struct IqGetDiscoItems {}

impl IqRequestHandler for IqGetDiscoItems {
fn process(
self: Box<Self>,
conn: &mut XmppConnection,
id: String,
from: Option<xmpp_parsers::Jid>,
) -> xmpp_parsers::iq::Iq {
info!("Got disco items query {} from {:?}", id, from);
stanzas::make_disco_items_result(id, conn.state.client.jid.clone(), from)
}
}

lazy_static! {
static ref INCOMING: element_processor::Processor<XmppConnection, bool, xmpp_parsers::Element> = {
let mut incoming = element_processor::Processor::new(&|_, e| {
warn!("Unknown stanza {}", String::from(&e));
true
});
incoming.register(&XmppConnection::incoming_iq_processing);
incoming.register(&XmppConnection::incoming_presence_processing);
incoming.register(&XmppConnection::incoming_message_processing);
incoming
};
static ref INCOMING_IQ_SET: element_processor::Processor<XmppConnection, Box<dyn IqRequestHandler>, xmpp_parsers::Element> = {
let mut iq_set =
element_processor::Processor::new(&|_conn: &mut XmppConnection, element| {
Box::new(IqRequestUnknown {
element,
type_: "set",
}) as Box<dyn IqRequestHandler>
});
iq_set.register(&XmppConnection::incoming_iq_processing_set_roster);
iq_set
};
static ref INCOMING_IQ_GET: element_processor::Processor<XmppConnection, Box<dyn IqRequestHandler>, xmpp_parsers::Element> = {
let mut iq_get =
element_processor::Processor::new(&|_conn: &mut XmppConnection, element| {
Box::new(IqRequestUnknown {
element,
type_: "get",
}) as Box<dyn IqRequestHandler>
});
iq_get.register(&XmppConnection::incoming_iq_processing_get_ping);
iq_get.register(&XmppConnection::incoming_iq_processing_get_disco_info);
iq_get.register(&XmppConnection::incoming_iq_processing_get_disco_items);
iq_get.register(&XmppConnection::incoming_iq_processing_get_version);
iq_get
};
}

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

impl From<XmppConnection> for MaybeXmppConnection {
fn from(from: XmppConnection) -> MaybeXmppConnection {
MaybeXmppConnection {
account: from.account,
state: Some(from.state),
}
}
}

impl From<config::Account> for MaybeXmppConnection {
fn from(from: config::Account) -> MaybeXmppConnection {
MaybeXmppConnection {
account: std::rc::Rc::new(from),
state: None,
}
}
}

impl From<std::rc::Rc<config::Account>> for MaybeXmppConnection {
fn from(from: std::rc::Rc<config::Account>) -> MaybeXmppConnection {
MaybeXmppConnection {
account: from,
state: None,
}
}
}

impl MaybeXmppConnection {
/// connects if nothing connected
/// don't connect only if stop_future resolved
pub 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,
{
info!("xmpp connection...");
let MaybeXmppConnection { account, state } = self;

if let Some(state) = state {
Box::new(future::ok(XmppConnection { account, state }))
as Box<dyn Future<Item = _, Error = _>>
} else {
Box::new(
stop_future
.clone()
.select2(
future::loop_fn(account, move |account| {
info!("xmpp initialization...");
let client =
Client::new_with_jid(account.jid.clone(), &account.password);
info!("xmpp initialized");

let stop_future2 = stop_future.clone();
let stop_future3 = stop_future.clone();
let stop_future4 = stop_future.clone();

// future to wait for online
Box::new(
XmppConnection {
state: XmppState {
client,
data: std::default::Default::default(),
},
account,
}
.processing(XmppConnection::online, stop_future.clone())
.map_err(|(acc, _)| acc)
.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),
})
.and_then(|conn| conn.initial_roster(stop_future2))
.and_then(|conn| conn.self_presence(stop_future3))
.and_then(|conn| conn.enter_mucs(stop_future4))
.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"))
}
}),
)
}
}
}

impl XmppConnection {
/// base XMPP processing
/// Returns false on error to disconnect
fn xmpp_processing(&mut self, event: &Event) -> bool {
match event {
Event::Stanza(stanza) => INCOMING.process(self, stanza.clone()),
Event::Online => true,
e => {
warn!("Unexpected event {:?}", e);
false
}
}
}

/// Process roster push
/// see RFC 6212 2.1.6. Roster Push
fn incoming_iq_processing_set_roster(
&mut self,
roster: xmpp_parsers::roster::Roster,
) -> Box<dyn IqRequestHandler> {
for i in roster.items {
if let Some(ref mut rdata) = self.state.data.roster.get_mut(&i.jid) {
info!("Update {} in roster", i.jid);
rdata.0 = i.subscription;
rdata.1 = i.ask;
} else {
info!("Add {} to roster", i.jid);
self.state
.data
.roster
.insert(i.jid.clone(), (i.subscription, i.ask));
}
self.process_jid(&i.jid);
}
Box::new(IqSetRoster {})
}

/// Enforce to answer to IQ "set"
fn incoming_iq_processing_set(
&mut self,
id: String,
from: Option<xmpp_parsers::Jid>,
element: minidom::Element,
) -> xmpp_parsers::iq::Iq {
INCOMING_IQ_SET
.process(self, element)
.process(self, id, from)
}

/// Process ping request
/// see XEP-0199: XMPP Ping
fn incoming_iq_processing_get_ping(
&mut self,
_ping: xmpp_parsers::ping::Ping,
) -> Box<dyn IqRequestHandler> {
Box::new(IqGetPing {})
}

/// Process disco query
/// see XEP-0030: Service Discovery
fn incoming_iq_processing_get_disco_info(
&mut self,
disco: xmpp_parsers::disco::DiscoInfoQuery,
) -> Box<dyn IqRequestHandler> {
if let Some(ref node) = disco.node {
warn!("Unsupported node {}", node);
Box::new(IqRequestUnknown {
element: disco.into(),
type_: "get",
})
} else {
Box::new(IqGetDiscoInfo {})
}
}

/// Process disco items query
/// see XEP-0030: Service Discovery
fn incoming_iq_processing_get_disco_items(
&mut self,
disco: xmpp_parsers::disco::DiscoItemsQuery,
) -> Box<dyn IqRequestHandler> {
if let Some(ref node) = disco.node {
warn!("Unsupported node {}", node);
Box::new(IqRequestUnknown {
element: disco.into(),
type_: "get",
})
} else {
Box::new(IqGetDiscoItems {})
}
}

/// Process version query
/// see XEP-0092: Software Version
fn incoming_iq_processing_get_version(
&mut self,
_version: xmpp_parsers::version::VersionQuery,
) -> Box<dyn IqRequestHandler> {
Box::new(IqGetVersion {})
}

/// Enforce to answer to IQ "get"
fn incoming_iq_processing_get(
&mut self,
id: String,
from: Option<xmpp_parsers::Jid>,
element: minidom::Element,
) -> xmpp_parsers::iq::Iq {
INCOMING_IQ_GET
.process(self, element)
.process(self, id, from)
}

fn incoming_iq_processing(&mut self, iq: xmpp_parsers::iq::Iq) -> bool {
match iq.payload {
xmpp_parsers::iq::IqType::Set(element) => {
let iq_answer = self.incoming_iq_processing_set(iq.id, iq.from, element);
self.state.data.send_queue.push_back(iq_answer.into());
}
xmpp_parsers::iq::IqType::Error(e) => {
if let Some((_, handler)) = self.state.data.pending_ids.remove_entry(&iq.id) {
return handler.1.error(self, e);
}
error!("iq error: {:?}", e);
return false;
}
xmpp_parsers::iq::IqType::Get(element) => {
let iq_answer = self.incoming_iq_processing_get(iq.id, iq.from, element);
self.state.data.send_queue.push_back(iq_answer.into());
}
xmpp_parsers::iq::IqType::Result(opt_element) => {
if let Some((_, handler)) = self.state.data.pending_ids.remove_entry(&iq.id) {
return handler.1.result(self, opt_element);
}
warn!(
"Unwanted iq result id {} from {:?}: {:?}",
iq.id,
iq.from,
opt_element.map(|e| String::from(&e))
);
}
}
true
}

fn incoming_presence_processing(&mut self, presence: xmpp_parsers::presence::Presence) -> bool {
if presence.from.as_ref() == Some(&self.state.client.jid) {
info!("Self-presence accepted");
self.state.data.self_presence = true;
} else {
warn!("Incoming presence stanza: {:?}", presence);
}
true
}

fn incoming_message_processing(&mut self, message: xmpp_parsers::message::Message) -> bool {
for payload in message.payloads.iter() {
use std::convert::TryInto;
if let Some(_delay) =
payload.clone().try_into().ok() as Option<xmpp_parsers::delay::Delay>
{
return true; // ignore delayed messages
}
}
warn!("Incoming message stanza: {:?}", message);
true
}

/// process event from xmpp stream
/// returns from future when condition met
/// or stop future was resolved.
/// Return item if connection was preserved or error otherwise.
/// Second part is a state of stop_future
pub fn processing<S, F, T, E>(
self,
stop_condition: S,
stop_future: F,
) -> impl Future<
Item = (Self, Result<Either<F, T>, E>),
Error = (std::rc::Rc<config::Account>, Result<Either<F, T>, E>),
>
where
F: Future<Item = T, Error = E> + 'static,
S: FnMut(&mut Self, Event) -> Result<bool, ()> + 'static,
T: 'static,
E: 'static,
{
future::loop_fn(
(self, stop_future, stop_condition),
|(xmpp, stop_future, mut stop_condition)| {
// ToDo: check timeouts if iqs

let XmppConnection {
state: XmppState { client, mut data },
account,
} = xmpp;

if let Some(send_element) = data.send_queue.pop_front() {
use tokio::prelude::Sink;

info!("Sending {}", String::from(&send_element));
Box::new(
client
.send(Packet::Stanza(send_element))
.select2(stop_future)
.then(move |r| match r {
Ok(Either::A((client, b))) => {
Box::new(future::ok(future::Loop::Continue((
XmppConnection {
state: XmppState { client, data },
account,
},
b,
stop_condition,
))))
as Box<dyn Future<Item = _, Error = _>>
}
Ok(Either::B((t, a))) => Box::new(a.then(|r| match r {
Ok(client) => future::ok(future::Loop::Break((
XmppConnection {
state: XmppState { client, data },
account,
},
Ok(Either::B(t)),
))),
Err(se) => {
warn!("XMPP sending error: {}", se);
future::err((account, Ok(Either::B(t))))
}
})),
Err(Either::A((e, b))) => {
warn!("XMPP sending error: {}", e);
Box::new(future::err((account, Ok(Either::A(b)))))
}
Err(Either::B((e, a))) => Box::new(a.then(|r| match r {
Ok(client) => future::ok(future::Loop::Break((
XmppConnection {
state: XmppState { client, data },
account,
},
Err(e),
))),
Err(se) => {
warn!("XMPP sending error: {}", se);
future::err((account, Err(e)))
}
})),
}),
) as Box<dyn Future<Item = _, Error = _>>
} else {
Box::new(
client
.into_future()
.select2(stop_future)
.then(move |r| match r {
Ok(Either::A(((event, client), b))) => {
if let Some(event) = event {
let mut xmpp = XmppConnection {
state: XmppState { client, data },
account,
};

if 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, Ok(Either::A(b))))
}
}
} else {
future::err((xmpp.account, Ok(Either::A(b))))
}
} 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, data },
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, data },
account,
},
Err(e),
)))
} else {
future::err((account, Err(e)))
}
}
}),
)
}
},
)
}

/// get connection and wait for online status and set presence
/// returns error if something went wrong and xmpp connection is broken
fn online(&mut self, event: Event) -> Result<bool, ()> {
match event {
Event::Online => {
info!("Online!");
Ok(true)
}
Event::Stanza(s) => {
warn!("Stanza before online: {}", String::from(&s));
Ok(false)
}
_ => {
error!("Disconnected while online");
Err(())
}
}
}

fn initial_roster<F, E>(
self,
stop_future: F,
) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
where
F: Future<Error = E> + 'static,
E: 'static,
{
let XmppConnection {
account,
state: XmppState { client, mut data },
} = self;
use tokio::prelude::Sink;

data.counter += 1;
let id_init_roster = format!("id_init_roster{}", data.counter);
let get_roster = stanzas::make_get_roster(&id_init_roster);
let account2 = account.clone();
info!("Quering roster... {}", String::from(&get_roster));
data.pending_ids.insert(
id_init_roster.clone(),
(
Instant::now() + Duration::from_secs(60),
Box::new(InitRosterIqHandler {}),
),
);

client
.send(Packet::Stanza(get_roster))
.map_err(move |e| {
error!("Error on querying roster: {}", e);
account2
})
.and_then(move |client| {
XmppConnection {
state: XmppState { client, data },
account,
}
.processing(move |conn, _| Ok(conn.state.data.roster_init), stop_future)
.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),
})
})
}

fn self_presence<F, E>(
self,
stop_future: F,
) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
where
F: Future<Error = E> + 'static,
E: Into<failure::Error> + 'static,
{
let XmppConnection {
account,
state: XmppState { client, data },
} = self;
use tokio::prelude::Sink;

let presence = stanzas::make_presence(&account);
let account2 = account.clone();
info!("Sending presence... {}", String::from(&presence));

client
.send(Packet::Stanza(presence))
.map_err(|e| {
error!("Error on send self-presence: {}", e);
account2
})
.and_then(move |client| {
XmppConnection {
state: XmppState { client, data },
account,
}
.processing(
move |conn, _| Ok(conn.state.data.self_presence),
stop_future,
)
.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),
})
})
}

fn process_jid(&mut self, xmpp_to: &xmpp_parsers::Jid) {
if let Some(ref mut mailbox) = self.state.data.outgoing_mailbox.get_mut(xmpp_to) {
if !mailbox.is_empty() {
if let Some(ref mut rdata) = self.state.data.roster.get_mut(xmpp_to) {
info!("Jid {} in roster", xmpp_to);
let sub_to = match rdata.0 {
xmpp_parsers::roster::Subscription::To => true,
xmpp_parsers::roster::Subscription::Both => true,
_ => false,
};
if sub_to {
info!("Subscribed to {}", xmpp_to);
self.state.data.send_queue.extend(
mailbox.drain(..).map(|message| {
stanzas::make_chat_message(xmpp_to.clone(), message)
}),
);
} else if rdata.1 == xmpp_parsers::roster::Ask::None {
info!("Not subscribed to {}", xmpp_to);
self.state
.data
.send_queue
.push_back(stanzas::make_ask_subscribe(xmpp_to.clone()));
}
let sub_from = match rdata.0 {
xmpp_parsers::roster::Subscription::From => true,
xmpp_parsers::roster::Subscription::Both => true,
_ => false,
};
if !sub_from {
info!("Not subscription from {}", xmpp_to);
self.state
.data
.send_queue
.push_back(stanzas::make_allow_subscribe(xmpp_to.clone()));
}
} else {
info!("Jid {} not in roster", xmpp_to);

self.state.data.counter += 1;
let id_add_roster = format!("id_add_roster{}", self.state.data.counter);
let add_roster = stanzas::make_add_roster(&id_add_roster, xmpp_to.clone());
info!("Adding jid {} to roster id {}", xmpp_to, id_add_roster);
self.state.data.pending_ids.insert(
id_add_roster,
(
Instant::now() + Duration::from_secs(60),
Box::new(AddRosterIqHandler {
jid: xmpp_to.clone(),
}),
),
);
self.state.data.send_queue.push_back(add_roster);
}
}
}
}

pub fn process_command(&mut self, cmd: XmppCommand) {
info!("Got command");
match cmd {
XmppCommand::Chat { xmpp_to, message } => {
self.state
.data
.outgoing_mailbox
.entry(xmpp_to.clone())
.or_default()
.push(message);
self.process_jid(&xmpp_to);
}
XmppCommand::Chatroom { muc_id, message } => {
if let Some(muc) = self.state.data.mucs.get(&muc_id) {
self.state
.data
.send_queue
.push_back(stanzas::make_muc_message(muc.clone(), message));
} else {
error!("Not found MUC {}", muc_id);
}
}
XmppCommand::Ping => {
self.state.data.counter += 1;
let id_ping = format!("id_ping{}", self.state.data.counter);
let ping = stanzas::make_ping(&id_ping, self.state.client.jid.clone());
self.state.data.send_queue.push_back(ping);
self.state.data.pending_ids.insert(
id_ping,
(
Instant::now() + Duration::from_secs(30),
Box::new(PingIqHandler {}),
),
);
}
XmppCommand::TimeoutCleanup => {
let now = Instant::now();
let timeouted: Vec<String> = self
.state
.data
.pending_ids
.iter()
.filter_map(|(id, (timeout, _))| {
if now >= *timeout {
Some(id.to_string())
} else {
None
}
})
.collect();
let mut correct = true;
timeouted.into_iter().for_each(|id| {
if let Some((_, handler)) = self.state.data.pending_ids.remove(&id) {
correct &= handler.timeout(&mut self);
}
})
}
}
}

pub fn shutdown(self) -> impl Future<Item = (), Error = failure::Error> {
info!("Shutdown connection");
let XmppConnection { account, state } = self;
stream::iter_ok(
state
.data
.mucs
.values()
.map(std::clone::Clone::clone)
.collect::<Vec<_>>(),
)
.fold(state, move |XmppState { client, data }, muc_jid| {
let muc_presence =
stanzas::make_muc_presence_leave(account.jid.clone(), muc_jid.clone());

info!(
"Sending muc leave presence... {}",
String::from(&muc_presence)
);
use tokio::prelude::Sink;
client
.send(Packet::Stanza(muc_presence))
.map_err(|e| {
error!("Error on send muc presence: {}", e);
e
})
.and_then(|client| future::ok(XmppState { client, data }))
})
.map(|_| ())
}

fn enter_mucs<F, E>(
self,
_stop_future: F,
) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
where
F: Future<Error = E> + 'static,
E: Into<failure::Error> + 'static,
{
let XmppConnection { account, state } = self;

let account2 = account.clone();
let account3 = account.clone();

stream::iter_ok(account.chatrooms.clone())
.fold(state, move |XmppState { client, mut data }, muc_jid| {
data.counter += 1;
let id_muc_presence = format!("id_muc_presence{}", data.counter);
let muc_presence = stanzas::make_muc_presence(
&id_muc_presence,
account2.jid.clone(),
muc_jid.1.clone(),
);

info!("Sending muc presence... {}", String::from(&muc_presence));

let account4 = account2.clone();

use tokio::prelude::Sink;
client
.send(Packet::Stanza(muc_presence))
.map_err(|e| {
error!("Error on send muc presence: {}", e);
account4
})
.and_then(|client| {
data.mucs.insert(muc_jid.0, muc_jid.1);
future::ok(XmppState { client, data })
})
})
.map(|state| XmppConnection {
account: account3,
state,
})
}
}
edit in src/xmpp/stanzas.rs at line 2

[3.57]→[2.38896:38977](∅→∅)

use xmpp_parsers::disco::{DiscoInfoResult, DiscoItemsResult, Feature, Identity};
edit in src/xmpp/stanzas.rs at line 7

[3.35465]→[2.39009:39127](∅→∅)

use xmpp_parsers::stanza_error::{DefinedCondition, ErrorType, StanzaError};
use xmpp_parsers::version::VersionResult;
edit in src/xmpp/stanzas.rs at line 43

[3.39288]→[3.36163:36166](∅→∅),[2.39608]→[3.36163:36166](∅→∅),[3.36163]→[3.36163:36166](∅→∅),[3.36166]→[2.39609:39858](∅→∅)

}

pub fn make_roster_push_answer(
id: String,
from: xmpp_parsers::Jid,
to: Option<xmpp_parsers::Jid>,
) -> Iq {
let mut answer = Iq::from_result(id, None as Option<Roster>);
answer.from = Some(from);
answer.to = to;
answer
edit in src/xmpp/stanzas.rs at line 51

[2.39914]→[2.39914:40043](∅→∅),[2.40043]→[3.36544:36596](∅→∅),[3.36544]→[3.36544:36596](∅→∅)

pub fn make_allow_subscribe(jid: xmpp_parsers::Jid) -> Element {
let mut presence = Presence::new(PresenceType::Subscribed);
presence.to = Some(jid);
presence.into()
}
replacement in src/xmpp/stanzas.rs at line 88

[2.40494]→[2.40494:40587](∅→∅)

pub fn make_pong(id: String, from: xmpp_parsers::Jid, to: Option<xmpp_parsers::Jid>) -> Iq {

[2.40494]

[2.40587]

pub fn make_pong(id: &str, from: xmpp_parsers::Jid, to: Option<xmpp_parsers::Jid>) -> Element {
replacement in src/xmpp/stanzas.rs at line 92

[2.40697]→[2.40697:41574](∅→∅)

pong
}

pub fn make_version(id: String, from: xmpp_parsers::Jid, to: Option<xmpp_parsers::Jid>) -> Iq {
let mut version = Iq::from_result(
id,
Some(VersionResult {
name: "SendXmppDRust".to_string(),
version: "0.1.0".to_string(),
os: None,
}),
);
version.from = Some(from);
version.to = to;
version
}

pub fn make_iq_unsupported_error(
id: String,
from: xmpp_parsers::Jid,
to: Option<xmpp_parsers::Jid>,
) -> Iq {
let mut error = Iq::from_error(
id,
StanzaError {
type_: ErrorType::Cancel,
by: Some(from.clone()),
defined_condition: DefinedCondition::ServiceUnavailable,
texts: std::collections::BTreeMap::new(),
other: None,
},
);
error.from = Some(from);
error.to = to;
error

[2.40697]

[2.41574]

pong.into()
edit in src/xmpp/stanzas.rs at line 94

[2.41576]→[2.41576:42720](∅→∅),[2.42720]→[3.37821:37823](∅→∅),[3.37821]→[3.37821:37823](∅→∅)

pub fn make_disco_info_result(
id: String,
from: xmpp_parsers::Jid,
to: Option<xmpp_parsers::Jid>,
) -> Iq {
let mut result = Iq::from_result(
id,
Some(DiscoInfoResult {
node: None,
identities: vec![Identity {
category: "client".to_string(),
type_: "bot".to_string(),
name: None,
lang: None,
}],
features: vec![
Feature {
var: "http://jabber.org/protocol/disco#info".to_string(),
},
Feature {
var: "urn:xmpp:ping".to_string(),
},
],
extensions: vec![],
}),
);
result.from = Some(from);
result.to = to;
result
}

pub fn make_disco_items_result(
id: String,
from: xmpp_parsers::Jid,
to: Option<xmpp_parsers::Jid>,
) -> Iq {
let mut result = Iq::from_result(
id,
Some(DiscoItemsResult {
node: None,
items: vec![],
}),
);
result.from = Some(from);
result.to = to;
result
}
replacement in src/xmpp/mod.rs at line 0

[3.17]→[2.42721:42773](∅→∅)

use tokio::prelude::future::{self, Either, Future};

[3.17]

[3.62]

use tokio::prelude::future::{self, Either};
edit in src/xmpp/mod.rs at line 2

[3.90]

[3.197]

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

use tokio_xmpp::{Client, Event, Packet};
edit in src/xmpp/mod.rs at line 7

[3.38811]

[3.296]

use std::collections::{HashMap, VecDeque};

mod stanzas;
edit in src/xmpp/mod.rs at line 13

[2.42797]

[3.3355]

#[derive(Default)]
struct XmppData {
/// known roster data
roster: HashMap<
xmpp_parsers::Jid,
(
xmpp_parsers::roster::Subscription,
xmpp_parsers::roster::Ask,
),
>,
/// ids counter
counter: usize,
/// map from id of adding item to roster and jid of item
pending_add_roster_ids: HashMap<String, xmpp_parsers::Jid>,
/// stanzas to send
send_queue: VecDeque<minidom::Element>,
/// outgoing mailbox
outgoing_mailbox: HashMap<xmpp_parsers::Jid, Vec<String>>,
/// muc id to muc jid
mucs: HashMap<String, xmpp_parsers::Jid>,
}

struct XmppState {
client: Client,
data: XmppData,
}

struct XmppElementProcessor {
incoming: element_processor::Processor<XmppConnection, bool, xmpp_parsers::Element>,
}

impl XmppElementProcessor {
fn new() -> XmppElementProcessor {
let mut incoming = element_processor::Processor::new(&|_, e| {
warn!("Unknown stanza {:#?}", e);
true
});
incoming.register(&XmppConnection::incoming_iq_processing);

XmppElementProcessor { incoming }
}
}

struct MaybeXmppConnection {
account: std::rc::Rc<config::Account>,
state: Option<XmppState>,
}

struct XmppConnection {
account: std::rc::Rc<config::Account>,
state: XmppState,
}

impl From<XmppConnection> for MaybeXmppConnection {
fn from(from: XmppConnection) -> MaybeXmppConnection {
MaybeXmppConnection {
account: from.account,
state: Some(from.state),
}
}
}

impl From<config::Account> for MaybeXmppConnection {
fn from(from: config::Account) -> MaybeXmppConnection {
MaybeXmppConnection {
account: std::rc::Rc::new(from),
state: None,
}
}
}

impl From<std::rc::Rc<config::Account>> for MaybeXmppConnection {
fn from(from: std::rc::Rc<config::Account>) -> MaybeXmppConnection {
MaybeXmppConnection {
account: from,
state: None,
}
}
}

impl MaybeXmppConnection {
/// connects if nothing connected
/// don't connect only 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,
{
info!("xmpp connection...");
let MaybeXmppConnection { account, state } = self;

if let Some(state) = state {
Box::new(future::ok(XmppConnection { account, state }))
as Box<dyn Future<Item = _, Error = _>>
} else {
Box::new(
stop_future
.clone()
.select2(
future::loop_fn(account, move |account| {
info!("xmpp initialization...");
let client =
Client::new_with_jid(account.jid.clone(), &account.password);
info!("xmpp initialized");

let stop_future2 = stop_future.clone();
let stop_future3 = stop_future.clone();
let stop_future4 = stop_future.clone();

// future to wait for online
Box::new(
XmppConnection {
state: XmppState {
client,
data: std::default::Default::default(),
},
account,
}
.processing(XmppConnection::online, stop_future.clone())
.map_err(|(acc, _)| acc)
.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),
})
.and_then(|conn| conn.initial_roster(stop_future2))
.and_then(|conn| conn.self_presence(stop_future3))
.and_then(|conn| conn.enter_mucs(stop_future4))
.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"))
}
}),
)
}
}
}

impl XmppConnection {
/// base XMPP processing
/// Returns false on error to disconnect
fn xmpp_processing(&mut self, event: &Event) -> bool {
match event {
Event::Stanza(stanza) => {
let processors = XmppElementProcessor::new();
processors.incoming.process(self, stanza.clone())
}
Event::Online => true,
e => {
warn!("Unexpected event {:?}", e);
false
}
}
}

fn incoming_iq_processing(&mut self, iq: xmpp_parsers::iq::Iq) -> bool {
use std::convert::TryInto;
if let Some((_, jid)) = self.state.data.pending_add_roster_ids.remove_entry(&iq.id) {
if let xmpp_parsers::iq::IqType::Result(None) = iq.payload {
if self.state.data.roster.contains_key(&jid) {
info!("Jid {} updated to roster", jid);
} else {
info!("Jid {} added in roster", jid);
self.state.data.roster.insert(
jid.clone(),
(
xmpp_parsers::roster::Subscription::None,
xmpp_parsers::roster::Ask::None,
),
);
}
self.process_jid(&jid);
} else {
warn!(
"Wrong payload when adding {} to roster: {:?}",
jid, iq.payload
);
}
}
match iq.payload {
xmpp_parsers::iq::IqType::Set(element) => {
if let Some(roster) =
element.try_into().ok() as Option<xmpp_parsers::roster::Roster>
{
for i in roster.items {
if let Some(ref mut rdata) = self.state.data.roster.get_mut(&i.jid) {
info!("Update {} in roster", i.jid);
rdata.0 = i.subscription;
rdata.1 = i.ask;
} else {
info!("Add {} to roster", i.jid);
self.state
.data
.roster
.insert(i.jid.clone(), (i.subscription, i.ask));
}
self.process_jid(&i.jid);
}
}
}
xmpp_parsers::iq::IqType::Error(e) => {
error!("iq error: {:?}", e);
return false;
}
xmpp_parsers::iq::IqType::Get(element) => {
if let Some(_ping) = element.try_into().ok() as Option<xmpp_parsers::ping::Ping> {
let pong = stanzas::make_pong(&iq.id, self.state.client.jid.clone(), iq.from);
self.state.data.send_queue.push_back(pong);
}
}
_ => (), // ignore
}
true
}

/// process event from xmpp stream
/// returns from future when condition met
/// or stop future was resolved.
/// Return item if connection was preserved or error otherwise.
/// Second part is a state of stop_future
fn processing<S, F, T, E>(
self,
stop_condition: S,
stop_future: F,
) -> impl Future<
Item = (Self, Result<Either<F, T>, E>),
Error = (std::rc::Rc<config::Account>, Result<Either<F, T>, E>),
>
where
F: Future<Item = T, Error = E> + 'static,
S: FnMut(&mut Self, Event) -> Result<bool, ()> + 'static,
T: 'static,
E: 'static,
{
future::loop_fn(
(self, stop_future, stop_condition),
|(xmpp, stop_future, mut stop_condition)| {
let XmppConnection {
state: XmppState { client, mut data },
account,
} = xmpp;

if let Some(send_element) = data.send_queue.pop_front() {
use tokio::prelude::Sink;

info!("Sending {:?}", send_element);
Box::new(
client
.send(Packet::Stanza(send_element))
.select2(stop_future)
.then(move |r| match r {
Ok(Either::A((client, b))) => {
Box::new(future::ok(future::Loop::Continue((
XmppConnection {
state: XmppState { client, data },
account,
},
b,
stop_condition,
))))
as Box<dyn Future<Item = _, Error = _>>
}
Ok(Either::B((t, a))) => Box::new(a.then(|r| match r {
Ok(client) => future::ok(future::Loop::Break((
XmppConnection {
state: XmppState { client, data },
account,
},
Ok(Either::B(t)),
))),
Err(se) => {
warn!("XMPP sending error: {}", se);
future::err((account, Ok(Either::B(t))))
}
})),
Err(Either::A((e, b))) => {
warn!("XMPP sending error: {}", e);
Box::new(future::err((account, Ok(Either::A(b)))))
}
Err(Either::B((e, a))) => Box::new(a.then(|r| match r {
Ok(client) => future::ok(future::Loop::Break((
XmppConnection {
state: XmppState { client, data },
account,
},
Err(e),
))),
Err(se) => {
warn!("XMPP sending error: {}", se);
future::err((account, Err(e)))
}
})),
}),
) as Box<dyn Future<Item = _, Error = _>>
} else {
Box::new(
client
.into_future()
.select2(stop_future)
.then(move |r| match r {
Ok(Either::A(((event, client), b))) => {
if let Some(event) = event {
let mut xmpp = XmppConnection {
state: XmppState { client, data },
account,
};
replacement in src/xmpp/mod.rs at line 335

[3.3356]→[3.29447:29460](∅→∅)

mod stanzas;

[3.3356]

[3.40647]

if 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, Ok(Either::A(b))))
}
}
} else {
future::err((xmpp.account, Ok(Either::A(b))))
}
} 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, data },
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, data },
account,
},
Err(e),
)))
} else {
future::err((account, Err(e)))
}
}
}),
)
}
},
)
}

/// get connection and wait for online status and set presence
/// returns error if something went wrong and xmpp connection is broken
fn online(&mut self, event: Event) -> Result<bool, ()> {
match event {
Event::Online => {
info!("Online!");
Ok(true)
}
Event::Stanza(s) => {
warn!("Stanza before online: {:?}", s);
Ok(false)
}
_ => {
error!("Disconnected while online");
Err(())
}
}
}

fn process_initial_roster(&mut self, event: Event, id_init_roster: &str) -> Result<bool, ()> {
if let Event::Stanza(s) = event {
use std::convert::TryInto;
match s.try_into() as Result<xmpp_parsers::iq::Iq, _> {
Ok(iq) => {
if iq.id == id_init_roster {
match iq.payload {
xmpp_parsers::iq::IqType::Error(_e) => {
error!("Get error instead of roster");
Err(())
}
xmpp_parsers::iq::IqType::Result(Some(result)) => {
match result.try_into() as Result<xmpp_parsers::roster::Roster, _> {
Ok(roster) => {
self.state.data.roster.clear();
info!("Got first roster:");
for i in roster.items {
info!(" >>> {:?}", i);
self.state
.data
.roster
.insert(i.jid, (i.subscription, i.ask));
}
Ok(true)
}
Err(e) => {
error!("Cann't parse roster: {}", e);
Err(())
}
}
}
_ => {
error!("Unknown result of roster");
Err(())
}
}
} else {
Ok(false)
}
}
Err(_e) => Ok(false),
}
} else {
error!("Wrong event while waiting roster");
Err(())
}
}

fn initial_roster<F, E>(
self,
stop_future: F,
) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
where
F: Future<Error = E> + 'static,
E: 'static,
{
let XmppConnection {
account,
state: XmppState { client, mut data },
} = self;
use tokio::prelude::Sink;

data.counter += 1;
let id_init_roster = format!("id_init_roster{}", data.counter);
let get_roster = stanzas::make_get_roster(&id_init_roster);
let account2 = account.clone();
info!("Quering roster... {:?}", get_roster);

client
.send(Packet::Stanza(get_roster))
.map_err(move |e| {
error!("Error on querying roster: {}", e);
account2
})
.and_then(move |client| {
XmppConnection {
state: XmppState { client, data },
account,
}
.processing(
move |conn, event| conn.process_initial_roster(event, &id_init_roster),
stop_future,
)
.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),
})
})
}

fn self_presence<F, E>(
self,
stop_future: F,
) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
where
F: Future<Error = E> + 'static,
E: Into<failure::Error> + 'static,
{
let XmppConnection {
account,
state: XmppState { client, data },
} = self;
use tokio::prelude::Sink;

let presence = stanzas::make_presence(&account);
let account2 = account.clone();
info!("Sending presence... {:?}", presence);

client
.send(Packet::Stanza(presence))
.map_err(|e| {
error!("Error on send self-presence: {}", e);
account2
})
.and_then(move |client| {
XmppConnection {
state: XmppState { client, data },
account,
}
.processing(
move |conn, event| {
if let Event::Stanza(s) = event {
use std::convert::TryInto;
match s.try_into() as Result<xmpp_parsers::presence::Presence, _> {
Ok(presence) => {
Ok(presence.from.as_ref() == Some(&conn.state.client.jid))
}
Err(e) => {
warn!("Not a self-presence: {}", e);
Ok(false)
}
}
} else {
error!("Wrong event while waiting self-presence");
Err(())
}
},
stop_future,
)
.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),
})
})
}

fn process_jid(&mut self, xmpp_to: &xmpp_parsers::Jid) {
if let Some(ref mut mailbox) = self.state.data.outgoing_mailbox.get_mut(xmpp_to) {
if !mailbox.is_empty() {
if let Some(ref mut rdata) = self.state.data.roster.get_mut(xmpp_to) {
info!("Jid {} in roster", xmpp_to);
let sub_to = match rdata.0 {
xmpp_parsers::roster::Subscription::To => true,
xmpp_parsers::roster::Subscription::Both => true,
_ => false,
};
if sub_to {
info!("Subscribed to {}", xmpp_to);
self.state.data.send_queue.extend(
mailbox.drain(..).map(|message| {
stanzas::make_chat_message(xmpp_to.clone(), message)
}),
);
} else if rdata.1 == xmpp_parsers::roster::Ask::None {
info!("Not subscribed to {}", xmpp_to);
self.state
.data
.send_queue
.push_back(stanzas::make_ask_subscribe(xmpp_to.clone()));
}
} else {
info!("Jid {} not in roster", xmpp_to);

self.state.data.counter += 1;
let id_add_roster = format!("id_add_roster{}", self.state.data.counter);
let add_roster = stanzas::make_add_roster(&id_add_roster, xmpp_to.clone());
self.state
.data
.pending_add_roster_ids
.insert(id_add_roster, xmpp_to.clone());
info!("Adding jid to roster... {:?}", add_roster);

self.state.data.send_queue.push_back(add_roster);
}
}
}
}

fn process_command(&mut self, cmd: XmppCommand) {
info!("Got command");
match cmd {
XmppCommand::Chat { xmpp_to, message } => {
self.state
.data
.outgoing_mailbox
.entry(xmpp_to.clone())
.or_default()
.push(message);
self.process_jid(&xmpp_to);
}
XmppCommand::Chatroom { muc_id, message } => {
if let Some(muc) = self.state.data.mucs.get(&muc_id) {
self.state
.data
.send_queue
.push_back(stanzas::make_muc_message(muc.clone(), message));
} else {
error!("Not found MUC {}", muc_id);
}
}
XmppCommand::Ping => {
self.state.data.counter += 1;
let id_ping = format!("id_ping{}", self.state.data.counter);
let ping = stanzas::make_ping(&id_ping, self.state.client.jid.clone());
self.state.data.send_queue.push_back(ping);
}
}
}

fn shutdown(self) -> impl Future<Item = (), Error = failure::Error> {
info!("Shutdown connection");
let XmppConnection { account, state } = self;
stream::iter_ok(
state
.data
.mucs
.values()
.map(std::clone::Clone::clone)
.collect::<Vec<_>>(),
)
.fold(state, move |XmppState { client, data }, muc_jid| {
let muc_presence =
stanzas::make_muc_presence_leave(account.jid.clone(), muc_jid.clone());

info!("Sending muc leave presence... {:?}", muc_presence);
use tokio::prelude::Sink;
client
.send(Packet::Stanza(muc_presence))
.map_err(|e| {
error!("Error on send muc presence: {}", e);
e
})
.and_then(|client| future::ok(XmppState { client, data }))
})
.map(|_| ())
}

fn enter_mucs<F, E>(
self,
_stop_future: F,
) -> impl Future<Item = Self, Error = std::rc::Rc<config::Account>>
where
F: Future<Error = E> + 'static,
E: Into<failure::Error> + 'static,
{
let XmppConnection { account, state } = self;

let account2 = account.clone();
let account3 = account.clone();

stream::iter_ok(account.chatrooms.clone())
.fold(state, move |XmppState { client, mut data }, muc_jid| {
data.counter += 1;
let id_muc_presence = format!("id_muc_presence{}", data.counter);
let muc_presence = stanzas::make_muc_presence(
&id_muc_presence,
account2.jid.clone(),
muc_jid.1.clone(),
);
replacement in src/xmpp/mod.rs at line 687

[3.40648]→[2.42798:42861](∅→∅)

mod xmpp_connection;
use xmpp_connection::MaybeXmppConnection;

[3.40648]

[3.56501]

info!("Sending muc presence... {:?}", muc_presence);
replacement in src/xmpp/mod.rs at line 689

[3.56502]→[2.42862:42900](∅→∅)

pub use xmpp_connection::XmppCommand;

[3.56502]

[3.60584]

let account4 = account2.clone();

use tokio::prelude::Sink;
client
.send(Packet::Stanza(muc_presence))
.map_err(|e| {
error!("Error on send muc presence: {}", e);
account4
})
.and_then(|client| {
data.mucs.insert(muc_jid.0, muc_jid.1);
future::ok(XmppState { client, data })
})
})
.map(|state| XmppConnection {
account: account3,
state,
})
}
}

#[derive(Debug)]
pub enum XmppCommand {
Chat {
xmpp_to: xmpp_parsers::Jid,
message: String,
},
Chatroom {
muc_id: String,
message: String,
},
Ping,
}
replacement in src/xmpp/element_processor.rs at line 0

[3.11301]→[2.43612:44709](∅→∅)

type Func<S, T, E> = dyn Fn(&mut S, E) -> T + Sync;

pub struct Processor<S: 'static, T: 'static, E: Clone + 'static> {
processors: Vec<Box<Func<S, Option<T>, E>>>,
default: &'static Func<S, T, E>,
}

impl<S: 'static, T: 'static, E: Clone + 'static> Processor<S, T, E> {
pub fn new<F>(f: &'static F) -> Processor<S, T, E>
where
F: Fn(&mut S, E) -> T + Sync + 'static,
{
Processor {
processors: vec![],
default: f,
}
}

pub fn register<F, A>(&mut self, f: &'static F)
where
F: Fn(&mut S, A) -> T + Sync + 'static,
A: std::convert::TryFrom<E>,
{
self.processors.push(Box::new(move |s, e: E| {
use std::convert::TryInto;
(e.try_into().ok() as Option<A>).map(|a| f(s, a))
}));
}

pub fn process(&self, s: &mut S, e: E) -> T {
for processor in self.processors.iter() {
match processor(s, e.clone()) {
Some(t) => return t,
None => continue,
}
}
(*self.default)(s, e)
}
}

[3.11301]

type Func<S, T, E> = dyn Fn(&mut S, E) -> T;

/// TryFrom based visitor
pub struct Processor<S: 'static, T: 'static, E: Clone + 'static> {
processors: Vec<Box<Func<S, Option<T>, E>>>,
default: &'static Func<S, T, E>,
}

impl<S: 'static, T: 'static, E: Clone + 'static> Processor<S, T, E> {
pub fn new<F>(f: &'static F) -> Processor<S, T, E>
where
F: Fn(&mut S, E) -> T + 'static,
{
Processor {
processors: vec![],
default: f,
}
}

pub fn register<F, A>(&mut self, f: &'static F)
where
F: Fn(&mut S, A) -> T + 'static,
A: std::convert::TryFrom<E>,
{
self.processors.push(Box::new(move |s, e: E| {
use std::convert::TryInto;
(e.try_into().ok() as Option<A>).map(|a| f(s, a))
}));
}

pub fn process(&self, s: &mut S, e: E) -> T {
for processor in self.processors.iter() {
match processor(s, e.clone()) {
Some(t) => return t,
None => continue,
}
}
(*self.default)(s, e)
}
}
edit in src/main.rs at line 21

[3.41232]→[2.44710:44749](∅→∅)

#[macro_use]
extern crate lazy_static;
replacement in README.md at line 0

[3.76]→[3.39248:39360](∅→∅),[3.39360]→[2.47247:47399](∅→∅)

# XMPP client daemon

## XMPP

### XMPP Core

Ensures granted subscription before delivering messages

### XEPs

#### XEP-0030: Service Discovery

Answers to `http://jabber.org/protocol/disco#info` query.
Answers to `http://jabber.org/protocol/disco#items` query.

[3.76]

[3.39360]

### XMPP client daemon
replacement in README.md at line 2

[3.39361]→[3.69301:69332](∅→∅)

#### XEP-0045: Multi-User Chat

[3.39361]

[3.39394]

#### Sending messages
edit in README.md at line 4

[3.39395]→[3.69333:69384](∅→∅),[3.39453]→[3.54279:54280](∅→∅),[3.69384]→[3.54279:54280](∅→∅),[3.39759]→[3.54279:54280](∅→∅),[3.54279]→[3.54279:54280](∅→∅),[3.54280]→[3.69385:69416](∅→∅),[3.69416]→[3.39485:39486](∅→∅),[3.39485]→[3.39485:39486](∅→∅),[3.39486]→[2.47400:47526](∅→∅),[3.69491]→[3.39537:39619](∅→∅),[2.47526]→[3.39537:39619](∅→∅),[3.39537]→[3.39537:39619](∅→∅),[3.39619]→[3.54302:54303](∅→∅),[3.54302]→[3.54302:54303](∅→∅),[3.54303]→[3.39620:39647](∅→∅),[3.39647]→[3.69492:69525](∅→∅),[3.69525]→[3.39647:39648](∅→∅),[3.39647]→[3.39647:39648](∅→∅),[3.39648]→[3.69526:69553](∅→∅),[3.69553]→[3.39648:39683](∅→∅),[3.39648]→[3.39648:39683](∅→∅)

Enters to all MUC in `account.chatrooms` at start.

#### XEP-0050: Ad-Hoc Commands

Answers empty commands list. (Awaiting for implementation in xmpp_parser).

#### XEP-0092: Software Version

Answers version.

#### XEP-0199: XMPP Ping

Sends ping each `account.ping` seconds to XMPP server.

Answers to incoming pings.

#### XEP-0203: Delayed Delivery

Ignores delayed messages.

## HTTP API

### Sending messages
replacement in README.md at line 8

[3.54382]→[3.39684:39712](∅→∅)

### Sending messages to MUC

[3.54382]

[3.54411]

#### Sending messages to MUC
edit in Cargo.toml at line 23

[3.54630]→[2.47564:47641](∅→∅)

lazy_static = "1.3.0" # ToDo: remove after const fn will be powerfull enough
edit in Cargo.lock at line 1138

[2.56213]→[2.56213:56291](∅→∅)

"lazy_static 1.3.0 (registry+https://github.com/rust-lang/crates.io-index)",