xmppftw/xmpp-rs
Watch
1
0
Fork
You've already forked xmpp-rs
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

parsers: port Iq to use the derive macros

Browse source
This commit is contained in:
Jonas Schäfer 2025-04-30 18:07:47 +02:00 • committed by Link Mauve
commit dc8c3eac4c
8 changed files with 486 additions and 312 deletions
Download patch file Download diff file Expand all files Collapse all files

110
tokio-xmpp/examples/download_avatars.rs
View file

@ -11,7 +11,7 @@ use xmpp_parsers::{
caps::{compute_disco, hash_caps, Caps},
disco::{DiscoInfoQuery, DiscoInfoResult, Feature, Identity},
hashes::Algo,
iq::{Iq, IqType},
iq::Iq,
jid::{BareJid, Jid},
ns,
presence::{Presence, Type as PresenceType},
@ -54,63 +54,41 @@ async fn main() {
client.send_stanza(presence.into()).await.unwrap();
} else if let Some(stanza) = event.into_stanza() {
match stanza {
Stanza::Iq(iq) => {
if let IqType::Get(payload) = iq.payload {
if payload.is("query", ns::DISCO_INFO) {
let query = DiscoInfoQuery::try_from(payload);
match query {
Ok(query) => {
let mut disco = disco_info.clone();
disco.node = query.node;
let iq = Iq::from_result(iq.id, Some(disco))
.with_to(iq.from.unwrap());
client.send_stanza(iq.into()).await.unwrap();
}
Err(err) => {
client
.send_stanza(
make_error(
iq.from.unwrap(),
iq.id,
ErrorType::Modify,
DefinedCondition::BadRequest,
&format!("{}", err),
)
.into(),
)
.await
.unwrap();
}
Stanza::Iq(Iq::Get {
payload, id, from, ..
}) => {
if payload.is("query", ns::DISCO_INFO) {
let query = DiscoInfoQuery::try_from(payload);
match query {
Ok(query) => {
let mut disco = disco_info.clone();
disco.node = query.node;
let iq = Iq::from_result(id, Some(disco)).with_to(from.unwrap());
client.send_stanza(iq.into()).await.unwrap();
}
} else {
// We MUST answer unhandled get iqs with a service-unavailable error.
client
.send_stanza(
make_error(
iq.from.unwrap(),
iq.id,
ErrorType::Cancel,
DefinedCondition::ServiceUnavailable,
"No handler defined for this kind of iq.",
Err(err) => {
client
.send_stanza(
make_error(
from.unwrap(),
id,
ErrorType::Modify,
DefinedCondition::BadRequest,
&format!("{}", err),
)
.into(),
)
.into(),
)
.await
.unwrap();
.await
.unwrap();
}
}
} else if let IqType::Result(Some(payload)) = iq.payload {
if payload.is("pubsub", ns::PUBSUB) {
let pubsub = PubSub::try_from(payload).unwrap();
let from = iq.from.clone().unwrap_or(jid.clone().into());
handle_iq_result(pubsub, &from);
}
} else if let IqType::Set(_) = iq.payload {
// We MUST answer unhandled set iqs with a service-unavailable error.
} else {
// We MUST answer unhandled get iqs with a service-unavailable error.
client
.send_stanza(
make_error(
iq.from.unwrap(),
iq.id,
from.unwrap(),
id,
ErrorType::Cancel,
DefinedCondition::ServiceUnavailable,
"No handler defined for this kind of iq.",
@ -121,6 +99,34 @@ async fn main() {
.unwrap();
}
}
Stanza::Iq(Iq::Result {
payload: Some(payload),
from,
..
}) => {
if payload.is("pubsub", ns::PUBSUB) {
let pubsub = PubSub::try_from(payload).unwrap();
let from = from.unwrap_or(jid.clone().into());
handle_iq_result(pubsub, &from);
}
}
Stanza::Iq(Iq::Set { from, id, .. }) => {
// We MUST answer unhandled set iqs with a service-unavailable error.
client
.send_stanza(
make_error(
from.unwrap(),
id,
ErrorType::Cancel,
DefinedCondition::ServiceUnavailable,
"No handler defined for this kind of iq.",
)
.into(),
)
.await
.unwrap();
}
Stanza::Iq(Iq::Error { .. }) | Stanza::Iq(Iq::Result { payload: None, .. }) => (),
Stanza::Message(message) => {
let from = message.from.clone().unwrap();
if let Some(body) = message.get_best_body(vec!["en"]) {

3
tokio-xmpp/examples/keep_connection.rs
View file

@ -88,8 +88,7 @@ async fn main() {
_ = ping_timer.tick() => {
log::info!("sending ping for fun & profit");
ping_ctr = ping_ctr.wrapping_add(1);
let mut iq = Iq::from_get(format!("ping-{}", ping_ctr), ping::Ping);
iq.to = Some(domain.clone());
let iq = Iq::from_get(format!("ping-{}", ping_ctr), ping::Ping).with_to(domain.clone());
stream.send(Box::new(iq.into())).await;
}
ev = stream.next() => match ev {

82
tokio-xmpp/src/client/iq.rs
View file

@ -18,10 +18,7 @@ use std::sync::Mutex;
use futures::Stream;
use tokio::sync::oneshot;
use xmpp_parsers::{
iq::{Iq, IqType},
stanza_error::StanzaError,
};
use xmpp_parsers::{iq::Iq, stanza_error::StanzaError};
use crate::{
event::make_id,
@ -40,11 +37,21 @@ pub enum IqRequest {
Set(Element),
}
impl From<IqRequest> for IqType {
fn from(other: IqRequest) -> IqType {
match other {
IqRequest::Get(v) => Self::Get(v),
IqRequest::Set(v) => Self::Set(v),
impl IqRequest {
fn into_iq(self, from: Option<Jid>, to: Option<Jid>, id: String) -> Iq {
match self {
Self::Get(payload) => Iq::Get {
from,
to,
id,
payload,
},
Self::Set(payload) => Iq::Set {
from,
to,
id,
payload,
},
}
}
}
@ -59,11 +66,22 @@ pub enum IqResponse {
Error(StanzaError),
}
impl From<IqResponse> for IqType {
fn from(other: IqResponse) -> IqType {
match other {
IqResponse::Result(v) => Self::Result(v),
IqResponse::Error(v) => Self::Error(v),
impl IqResponse {
fn into_iq(self, from: Option<Jid>, to: Option<Jid>, id: String) -> Iq {
match self {
Self::Error(error) => Iq::Error {
from,
to,
id,
error,
payload: None,
},
Self::Result(payload) => Iq::Result {
from,
to,
id,
payload,
},
}
}
}
@ -251,22 +269,28 @@ impl IqResponseTracker {
/// Returns the IQ stanza unharmed if it is not an IQ response matching
/// any request which is still being tracked.
pub fn handle_iq(&self, iq: Iq) -> ControlFlow<(), Iq> {
let payload = match iq.payload {
IqType::Error(error) => IqResponse::Error(error),
IqType::Result(result) => IqResponse::Result(result),
let (from, to, id, payload) = match iq {
Iq::Error {
from,
to,
id,
error,
payload: _,
} => (from, to, id, IqResponse::Error(error)),
Iq::Result {
from,
to,
id,
payload,
} => (from, to, id, IqResponse::Result(payload)),
_ => return ControlFlow::Continue(iq),
};
let key = (iq.from, iq.id);
let key = (from, id);
let mut map = self.map.lock().unwrap();
match map.remove(&key) {
None => {
log::trace!("not handling IQ response from {:?} with id {:?}: no active tracker for this tuple", key.0, key.1);
ControlFlow::Continue(Iq {
from: key.0,
id: key.1,
to: iq.to,
payload: payload.into(),
})
ControlFlow::Continue(payload.into_iq(key.0, to, key.1))
}
Some(sink) => {
sink.complete(payload);
@ -298,14 +322,6 @@ impl IqResponseTracker {
inner: rx,
};
map.insert(key.clone(), sink);
(
Iq {
from,
to: key.0,
id: key.1,
payload: req.into(),
},
token,
)
(req.into_iq(from, key.0, key.1), token)
}
}

9
tokio-xmpp/src/event.rs
View file

@ -43,10 +43,11 @@ impl Stanza {
pub fn ensure_id(&mut self) -> &str {
match self {
Self::Iq(iq) => {
if iq.id.is_empty() {
iq.id = make_id();
let id = iq.id_mut();
if id.is_empty() {
*id = make_id();
}
&iq.id
id
}
Self::Message(message) => message.id.get_or_insert_with(|| Id(make_id())).0.as_ref(),
Self::Presence(presence) => presence.id.get_or_insert_with(make_id),
@ -97,7 +98,7 @@ impl TryFrom<Stanza> for Presence {
impl TryFrom<Stanza> for Iq {
type Error = Stanza;
fn try_from(other: Stanza) -> Result<Self, Self::Error> {
fn try_from(other: Stanza) -> Result<Self, Stanza> {
match other {
Stanza::Iq(st) => Ok(st),
other => Err(other),

47
tokio-xmpp/src/stanzastream/negotiation.rs
View file

@ -13,7 +13,7 @@ use futures::{ready, Sink, Stream};
use xmpp_parsers::{
bind::{BindQuery, BindResponse},
iq::{Iq, IqType},
iq::Iq,
jid::{FullJid, Jid},
sm,
stream_error::{DefinedCondition, StreamError},
@ -200,31 +200,34 @@ impl NegotiationState {
match item {
Ok(XmppStreamElement::Stanza(data)) => match data {
Stanza::Iq(iq) if iq.id == BIND_REQ_ID => {
let error = match iq.payload {
IqType::Result(Some(payload)) => {
match BindResponse::try_from(payload) {
Ok(v) => {
let bound_jid = v.into();
if *sm_supported {
*self = Self::SendSmRequest {
Stanza::Iq(iq) if iq.id() == BIND_REQ_ID => {
let error = match iq {
Iq::Result {
payload: Some(payload),
..
} => match BindResponse::try_from(payload) {
Ok(v) => {
let bound_jid = v.into();
if *sm_supported {
*self = Self::SendSmRequest {
sm_state: None,
bound_jid: Some(bound_jid),
};
return Poll::Ready(Continue(None));
} else {
return Poll::Ready(Break(
NegotiationResult::StreamReset {
sm_state: None,
bound_jid: Some(bound_jid),
};
return Poll::Ready(Continue(None));
} else {
return Poll::Ready(Break(
NegotiationResult::StreamReset {
sm_state: None,
bound_jid: Jid::from(bound_jid),
},
));
}
bound_jid: Jid::from(bound_jid),
},
));
}
Err(e) => e.to_string(),
}
Err(e) => e.to_string(),
},
Iq::Result { payload: None, .. } => {
"Bind response has no payload".to_owned()
}
IqType::Result(None) => "Bind response has no payload".to_owned(),
_ => "Unexpected IQ type in response to bind request".to_owned(),
};
log::warn!("Received IQ matching the bind request, but parsing failed ({error})! Emitting stream error.");