Make xmpp command enum to allow different commands Save subscription ask status. Don't ask if already requested subscription.

[?]

Jan 23, 2019, 11:20 AM

SYH7UQP6A62HRY2K6YR23X45TMFHJD7SW6J4T2MH4C3DJXKXVDZQC

Dependencies

[2] AA2ZWGRL Enter to MUC
[3] 5A5UVGNM Move receiver closing logic out of xmpp processing
[4] PVCRPP3B Some servers don't send to in initial presence
[5] 5Y6YJ6UH Add shutdown function to make actions before offline
[6] BWDUANCV Second part of processing result is only about stop_future
[7] CBWCXUZZ Prepare adding new items to roster
[8] QWE26TMV update deps
[9] IK3YDPTY Update deps
[10] 5OBTKGDL Update deps
[11] 5IKA4GO7 Rename xmpp client field from "inner" to "client"
[12] VS6AHRWI Move XMPP to separate dir
[13] HU3NZX5Z Process self-presence via new processing code
[14] OB3HA2MD Use Client::new_with_jid to parse jid only once
[15] V5HDBSZM Use jid for receiver address
[16] UMTLHH77 Process commands in the separate function
[17] AYQZ2UIA Update deps
[18] RGOSS73U Convert self-presence to xmpp_parser's type
[19] XGP44R5H Rework stopping xmpp connection
[20] 4LRBIGVT Show info about xmpp errors
[21] 3GEU7TC7 Welcome to 2018!
[22] OGMBXBKP Move online to XmppConnection
[23] ACXUIS63 Update dependecies
[24] EOHEZXX3 Move request processing to structure
[25] QTCUURXN Add additional requirement for command stream
[26] QYY3KRGL Use failure instead Box<dyn Error>
[27] HKSQO7JZ Enable hyper http server and configuration
[28] TDOR5XQU Accept destination
[29] ALP2YJIU Rename XmppState to XmppProcessState
[30] ZI4GJ72V Add message to xmpp command
[31] WBU7UOQW Read chatroom from config
[32] OANBCLN5 Move xmpp client into XmppState
[*] FV6BJ5K6 Send self-presence and store account info in Rc so it willbe used in some future in parallel
[*] FVVPKFTL Initial commit

Change contents

edit in src/xmpp/stanzas.rs at line 52

[3.796]→[2.597:931](∅→∅)

pub fn make_muc_presence(id: &str, from: xmpp_parsers::Jid, to: xmpp_parsers::Jid) -> Element {
let mut presence = Presence::new(PresenceType::None);
presence.from = Some(from);
presence.to = Some(to);
presence.id = Some(id.to_string());
presence.add_payload(xmpp_parsers::muc::Muc::new());
presence.into()
}
replacement in src/xmpp/mod.rs at line 15

[3.1300]→[2.932:1023](∅→∅)

roster: HashMap<xmpp_parsers::Jid, (xmpp_parsers::roster::Subscription, Vec<String>)>,

[3.1300]

[3.1391]

roster: HashMap<
xmpp_parsers::Jid,
(
xmpp_parsers::roster::Subscription,
xmpp_parsers::roster::Ask,
Vec<String>,
),
>,
replacement in src/xmpp/mod.rs at line 101

[3.1591]→[2.1156:1346](∅→∅)

let client =
Client::new_with_jid(account.jid.clone(), &account.password);
info!("xmpp initialized");

[3.1591]

[3.1476]

let res_client = Client::new(&account.jid, &account.password);
match res_client {
Err(_e) => Box::new(future::ok(future::Loop::Continue(account)))
as Box<dyn Future<Item = _, Error = _>>,
Ok(client) => {
info!("xmpp initialized");
replacement in src/xmpp/mod.rs at line 108

[3.1477]→[2.1347:1551](∅→∅)

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

[3.1477]

[3.1065]

let stop_future2 = stop_future.clone();
let stop_future3 = stop_future.clone();
replacement in src/xmpp/mod.rs at line 111

[3.1066]→[2.1552:1963](∅→∅)

// future to wait for online
Box::new(
XmppConnection {
state: XmppState {
client,
data: std::default::Default::default(),
},
account,

[3.1066]

[3.3128]

// 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))
.then(
|r| match r {
Ok(conn) => future::ok(future::Loop::Break(conn)),
Err(acc) => future::ok(future::Loop::Continue(acc)),
},
),
)
replacement in src/xmpp/mod.rs at line 137

[3.3162]→[2.1964:2981](∅→∅)

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

[3.3162]

[3.3192]

}
edit in src/xmpp/mod.rs at line 172

[3.5147]→[3.5147:5217](∅→∅),[3.5147]→[3.5147:5217](∅→∅),[3.5147]→[3.5147:5217](∅→∅)

info!("Jid {} added to roster", jid);
replacement in src/xmpp/mod.rs at line 173

[2.3142]→[2.3142:3201](∅→∅)

rdata.1.push(message);

[2.3142]

[2.3201]

info!("Jid {} updated to roster", jid);
rdata.2.push(message);
edit in src/xmpp/mod.rs at line 176

[2.3242]

[2.3242]

info!("Jid {} added in roster", jid);
replacement in src/xmpp/mod.rs at line 179

[2.3354]→[2.3354:3453](∅→∅)

(xmpp_parsers::roster::Subscription::None, vec![message]),

[2.3354]

[2.3453]

(
xmpp_parsers::roster::Subscription::None,
xmpp_parsers::roster::Ask::None,
vec![message],
),
replacement in src/xmpp/mod.rs at line 203

[2.4095]→[2.4095:4156](∅→∅)

if !rdata.1.is_empty() {

[2.4095]

[2.4156]

rdata.1 = i.ask;
if !rdata.2.is_empty() {
replacement in src/xmpp/mod.rs at line 214

[2.4788]→[2.4788:4870](∅→∅)

rdata.1.drain(..).map(|message| {

[2.4788]

[2.4870]

rdata.2.drain(..).map(|message| {
replacement in src/xmpp/mod.rs at line 218

[2.5069]→[2.5069:5118](∅→∅)

} else {

[2.5069]

[2.5118]

} else if rdata.1 == xmpp_parsers::roster::Ask::None {
replacement in src/xmpp/mod.rs at line 231

[2.5728]→[2.5728:5810](∅→∅)

.insert(i.jid, (i.subscription, vec![]));

[2.5728]

[2.5810]

.insert(i.jid, (i.subscription, i.ask, vec![]));
replacement in src/xmpp/mod.rs at line 444

[2.6052]→[2.6052:6146](∅→∅)

.insert(i.jid, (i.subscription, vec![]));

[2.6052]

[3.18005]

.insert(i.jid, (i.subscription, i.ask, vec![]));
replacement in src/xmpp/mod.rs at line 575

[3.21716]→[2.6147:6231](∅→∅),[2.6231]→[3.21800:21852](∅→∅),[3.2982]→[3.21800:21852](∅→∅),[3.21800]→[3.21800:21852](∅→∅),[3.21852]→[2.6232:6994](∅→∅),[2.6994]→[3.22614:22687](∅→∅),[3.22614]→[3.22614:22687](∅→∅)

if let Some(ref mut rdata) = self.state.data.roster.get_mut(&cmd.xmpp_to) {
info!("Jid {} in roster", cmd.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 {}", cmd.xmpp_to);
self.state
.data
.send_queue
.push_back(stanzas::make_chat_message(cmd.xmpp_to, cmd.message));
} else {
info!("Not subscribed to {}", cmd.xmpp_to);
rdata.1.push(cmd.message);
self.state
.data
.send_queue
.push_back(stanzas::make_ask_subscribe(cmd.xmpp_to));
}
} else {
info!("Jid {} not in roster", cmd.xmpp_to);

[3.21716]

[3.22687]

match cmd {
XmppCommand::Chat { xmpp_to, message } => {
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
.push_back(stanzas::make_chat_message(xmpp_to, message));
} else if rdata.1 == xmpp_parsers::roster::Ask::None {
info!("Not subscribed to {}", xmpp_to);
rdata.2.push(message);
self.state
.data
.send_queue
.push_back(stanzas::make_ask_subscribe(xmpp_to));
}
} else {
info!("Jid {} not in roster", xmpp_to);
replacement in src/xmpp/mod.rs at line 601

[3.22688]→[3.22688:23123](∅→∅),[3.22688]→[3.22688:23123](∅→∅)

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, cmd.xmpp_to.clone());
self.state
.data
.pending_add_roster_ids
.insert(id_add_roster, (cmd.xmpp_to, cmd.message));
info!("Adding jid to roster... {:?}", add_roster);

[3.22688]

[3.23123]

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, message));
info!("Adding jid to roster... {:?}", add_roster);
replacement in src/xmpp/mod.rs at line 610

[3.23124]→[3.23124:23186](∅→∅),[3.23124]→[3.23124:23186](∅→∅)

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

[3.23124]

[3.23186]

self.state.data.send_queue.push_back(add_roster);
}
}
edit in src/xmpp/mod.rs at line 615

[3.12136]→[3.2983:2984](∅→∅),[3.2984]→[2.6995:8397](∅→∅),[2.8397]→[3.3119:3125](∅→∅),[3.3119]→[3.3119:3125](∅→∅)

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
.values()
.map(std::clone::Clone::clone)
.collect::<Vec<_>>(),
)
.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);

info!("Sending muc presence... {:?}", muc_presence);

let account4 = account2.clone();

use tokio::prelude::Sink;
client
.send(muc_presence)
.map_err(|e| {
error!("Error on send muc presence: {}", e);
account4
})
.and_then(|client| future::ok(XmppState { client, data }))
})
.map(|state| XmppConnection {
account: account3,
state,
})
}
replacement in src/xmpp/mod.rs at line 618

[3.60]→[2.8398:8484](∅→∅)

pub struct XmppCommand {
pub xmpp_to: xmpp_parsers::Jid,
pub message: String,

[3.60]

[2.8484]

pub enum XmppCommand {
Chat {
xmpp_to: xmpp_parsers::Jid,
message: String,
},
replacement in src/main.rs at line 90

[2.11076]→[2.11076:11156](∅→∅)

.send(XmppCommand { xmpp_to, message })

[2.11076]

[2.11156]

.send(XmppCommand::Chat { xmpp_to, message })
replacement in src/config.rs at line 6

[3.1502]→[2.13690:13773](∅→∅)

#[serde(deserialize_with = "deserialize_jid")]
pub jid: xmpp_parsers::Jid,

[3.1502]

[3.1523]

pub jid: String,
edit in src/config.rs at line 25

[3.1943]→[3.31464:31467](∅→∅),[3.31467]→[2.13774:13857](∅→∅),[3.780]→[3.31550:31619](∅→∅),[2.13857]→[3.31550:31619](∅→∅),[3.31550]→[3.31550:31619](∅→∅),[3.31619]→[2.13858:13976](∅→∅)

}

fn deserialize_jid<'de, D>(deserializer: D) -> Result<xmpp_parsers::Jid, D::Error>
where
D: serde::Deserializer<'de>,
{
use serde::Deserialize;
let s = String::deserialize(deserializer)?;
std::str::FromStr::from_str(&s).map_err(serde::de::Error::custom)