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

Rustfmt pass, and rustfmt --check in CI"

Browse source 
Signed-off-by: Maxime “pep” Buquet <pep@bouah.net>
This commit is contained in:
Maxime “pep” Buquet 2019-10-23 01:32:41 +02:00
commit a104ebc3f6
No known key found for this signature in database
GPG key ID: DEDA74AEECA9D0F2
79 changed files with 1344 additions and 957 deletions
Download patch file Download diff file Expand all files Collapse all files

39
tokio-xmpp/examples/contact_addr.rs
View file

@ -3,20 +3,13 @@ use std::convert::TryFrom;
use std::env::args;
use std::process::exit;
use tokio::runtime::current_thread::Runtime;
use tokio_xmpp::{Client, xmpp_codec::Packet};
use tokio_xmpp::{xmpp_codec::Packet, Client};
use xmpp_parsers::{
Element,
Jid,
disco::{DiscoInfoQuery, DiscoInfoResult},
iq::{Iq, IqType},
ns,
iq::{
Iq,
IqType,
},
disco::{
DiscoInfoResult,
DiscoInfoQuery,
},
server_info::ServerInfo,
Element, Jid,
};
fn main() {
@ -95,17 +88,19 @@ fn make_disco_iq(target: Jid) -> Element {
}
fn convert_field(field: Vec<String>) -> String {
field.iter()
.fold((field.len(), String::new()), |(l, mut acc), s| {
acc.push('<');
acc.push_str(&s);
acc.push('>');
if l > 1 {
acc.push(',');
acc.push(' ');
}
(0, acc)
}).1
field
.iter()
.fold((field.len(), String::new()), |(l, mut acc), s| {
acc.push('<');
acc.push_str(&s);
acc.push('>');
if l > 1 {
acc.push(',');
acc.push(' ');
}
(0, acc)
})
.1
}
fn print_server_info(server_info: ServerInfo) {

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

@ -21,7 +21,7 @@ use xmpp_parsers::{
pubsub::{Items, PubSub},
NodeName,
},
stanza_error::{StanzaError, ErrorType, DefinedCondition},
stanza_error::{DefinedCondition, ErrorType, StanzaError},
Jid,
};
@ -46,16 +46,14 @@ fn main() {
// Create outgoing pipe
let (mut tx, rx) = futures::unsync::mpsc::unbounded();
rt.spawn(
rx.forward(
sink.sink_map_err(|_| panic!("Pipe"))
)
rx.forward(sink.sink_map_err(|_| panic!("Pipe")))
.map(|(rx, mut sink)| {
drop(rx);
let _ = sink.close();
})
.map_err(|e| {
panic!("Send error: {:?}", e);
})
}),
);
let disco_info = make_disco();
@ -66,8 +64,7 @@ fn main() {
// Helper function to send an iq error.
let mut send_error = |to, id, type_, condition, text: &str| {
let error = StanzaError::new(type_, condition, "en", text);
let iq = Iq::from_error(id, error)
.with_to(to);
let iq = Iq::from_error(id, error).with_to(to);
tx.start_send(Packet::Stanza(iq.into())).unwrap();
};
@ -89,28 +86,45 @@ fn main() {
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());
let iq =
Iq::from_result(iq.id, Some(disco)).with_to(iq.from.unwrap());
tx.start_send(Packet::Stanza(iq.into())).unwrap();
},
}
Err(err) => {
send_error(iq.from.unwrap(), iq.id, ErrorType::Modify, DefinedCondition::BadRequest, &format!("{}", err));
},
send_error(
iq.from.unwrap(),
iq.id,
ErrorType::Modify,
DefinedCondition::BadRequest,
&format!("{}", err),
);
}
}
} else {
// We MUST answer unhandled get iqs with a service-unavailable error.
send_error(iq.from.unwrap(), iq.id, ErrorType::Cancel, DefinedCondition::ServiceUnavailable, "No handler defined for this kind of iq.");
send_error(
iq.from.unwrap(),
iq.id,
ErrorType::Cancel,
DefinedCondition::ServiceUnavailable,
"No handler defined for this kind of iq.",
);
}
} 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::from_str(jid).unwrap());
let from = iq.from.clone().unwrap_or(Jid::from_str(jid).unwrap());
handle_iq_result(pubsub, &from);
}
} else if let IqType::Set(_) = iq.payload {
// We MUST answer unhandled set iqs with a service-unavailable error.
send_error(iq.from.unwrap(), iq.id, ErrorType::Cancel, DefinedCondition::ServiceUnavailable, "No handler defined for this kind of iq.");
send_error(
iq.from.unwrap(),
iq.id,
ErrorType::Cancel,
DefinedCondition::ServiceUnavailable,
"No handler defined for this kind of iq.",
);
}
} else if stanza.is("message", "jabber:client") {
let message = Message::try_from(stanza).unwrap();
@ -186,20 +200,22 @@ fn get_disco_caps(disco: &DiscoInfoResult, node: &str) -> Caps {
// Construct a <presence/>
fn make_presence(caps: Caps) -> Presence {
let mut presence = Presence::new(PresenceType::None)
.with_priority(-1);
let mut presence = Presence::new(PresenceType::None).with_priority(-1);
presence.set_status("en", "Downloading avatars.");
presence.add_payload(caps);
presence
}
fn download_avatar(from: Jid) -> Iq {
Iq::from_get("coucou", PubSub::Items(Items {
max_items: None,
node: NodeName(String::from(ns::AVATAR_DATA)),
subid: None,
items: Vec::new(),
}))
Iq::from_get(
"coucou",
PubSub::Items(Items {
max_items: None,
node: NodeName(String::from(ns::AVATAR_DATA)),
subid: None,
items: Vec::new(),
}),
)
.with_to(from)
}
@ -222,10 +238,7 @@ fn handle_iq_result(pubsub: PubSub, from: &Jid) {
fn save_avatar(from: &Jid, id: String, data: &[u8]) -> io::Result<()> {
let directory = format!("data/{}", from);
let filename = format!("data/{}/{}", from, id);
println!(
"Saving avatar from {} to {}.",
from, filename
);
println!("Saving avatar from {} to {}.", from, filename);
create_dir_all(directory)?;
let mut file = File::create(filename)?;
file.write_all(data)

11
tokio-xmpp/examples/echo_bot.rs
View file

@ -4,9 +4,9 @@ use std::env::args;
use std::process::exit;
use tokio::runtime::current_thread::Runtime;
use tokio_xmpp::{Client, Packet};
use xmpp_parsers::{Jid, Element};
use xmpp_parsers::message::{Body, Message, MessageType};
use xmpp_parsers::presence::{Presence, Show as PresenceShow, Type as PresenceType};
use xmpp_parsers::{Element, Jid};
fn main() {
let args: Vec<String> = args().collect();
@ -29,16 +29,14 @@ fn main() {
// Create outgoing pipe
let (mut tx, rx) = futures::unsync::mpsc::unbounded();
rt.spawn(
rx.forward(
sink.sink_map_err(|_| panic!("Pipe"))
)
rx.forward(sink.sink_map_err(|_| panic!("Pipe")))
.map(|(rx, mut sink)| {
drop(rx);
let _ = sink.close();
})
.map_err(|e| {
panic!("Send error: {:?}", e);
})
}),
);
// Main loop, processes events
@ -47,7 +45,8 @@ fn main() {
if wait_for_stream_end {
/* Do nothing */
} else if event.is_online() {
let jid = event.get_jid()
let jid = event
.get_jid()
.map(|jid| format!("{}", jid))
.unwrap_or("unknown".to_owned());
println!("Online at {}", jid);

2
tokio-xmpp/examples/echo_component.rs
View file

@ -5,9 +5,9 @@ use std::process::exit;
use std::str::FromStr;
use tokio::runtime::current_thread::Runtime;
use tokio_xmpp::Component;
use xmpp_parsers::{Jid, Element};
use xmpp_parsers::message::{Body, Message, MessageType};
use xmpp_parsers::presence::{Presence, Show as PresenceShow, Type as PresenceType};
use xmpp_parsers::{Element, Jid};
fn main() {
let args: Vec<String> = args().collect();

108
tokio-xmpp/src/client/auth.rs
View file

@ -1,11 +1,14 @@
use std::str::FromStr;
use std::collections::HashSet;
use std::convert::TryFrom;
use futures::{Future, Poll, Stream, future::{ok, err, IntoFuture}};
use futures::{
future::{err, ok, IntoFuture},
Future, Poll, Stream,
};
use sasl::client::mechanisms::{Anonymous, Plain, Scram};
use sasl::client::Mechanism;
use sasl::common::scram::{Sha1, Sha256};
use sasl::common::Credentials;
use std::collections::HashSet;
use std::convert::TryFrom;
use std::str::FromStr;
use tokio_io::{AsyncRead, AsyncWrite};
use xmpp_parsers::sasl::{Auth, Challenge, Failure, Mechanism as XMPPMechanism, Response, Success};
@ -41,67 +44,74 @@ impl<S: AsyncRead + AsyncWrite + 'static> ClientAuth<S> {
let mut mechanism = local_mech();
if remote_mechs.contains(mechanism.name()) {
let initial = mechanism.initial().map_err(AuthError::Sasl)?;
let mechanism_name = XMPPMechanism::from_str(mechanism.name()).map_err(ProtocolError::Parsers)?;
let mechanism_name =
XMPPMechanism::from_str(mechanism.name()).map_err(ProtocolError::Parsers)?;
let send_initial = Box::new(stream.send_stanza(Auth {
mechanism: mechanism_name,
data: initial,
}))
.map_err(Error::Io);
let future = Box::new(send_initial.and_then(
|stream| Self::handle_challenge(stream, mechanism)
).and_then(
|stream| stream.restart()
));
return Ok(ClientAuth {
future,
});
.map_err(Error::Io);
let future = Box::new(
send_initial
.and_then(|stream| Self::handle_challenge(stream, mechanism))
.and_then(|stream| stream.restart()),
);
return Ok(ClientAuth { future });
}
}
Err(AuthError::NoMechanism)?
}
fn handle_challenge(stream: XMPPStream<S>, mut mechanism: Box<dyn Mechanism>) -> Box<dyn Future<Item = XMPPStream<S>, Error = Error>> {
fn handle_challenge(
stream: XMPPStream<S>,
mut mechanism: Box<dyn Mechanism>,
) -> Box<dyn Future<Item = XMPPStream<S>, Error = Error>> {
Box::new(
stream.into_future()
.map_err(|(e, _stream)| e.into())
.and_then(|(stanza, stream)| {
match stanza {
Some(Packet::Stanza(stanza)) => {
if let Ok(challenge) = Challenge::try_from(stanza.clone()) {
let response = mechanism
.response(&challenge.data);
Box::new(
response
.map_err(|e| AuthError::Sasl(e).into())
.into_future()
.and_then(|response| {
// Send response and loop
stream.send_stanza(Response { data: response })
.map_err(Error::Io)
.and_then(|stream| Self::handle_challenge(stream, mechanism))
})
)
} else if let Ok(_) = Success::try_from(stanza.clone()) {
Box::new(ok(stream))
} else if let Ok(failure) = Failure::try_from(stanza.clone()) {
Box::new(err(Error::Auth(AuthError::Fail(failure.defined_condition))))
} else if stanza.name() == "failure" {
// Workaround for https://gitlab.com/xmpp-rs/xmpp-parsers/merge_requests/1
Box::new(err(Error::Auth(AuthError::Sasl("failure".to_string()))))
} else {
stream
.into_future()
.map_err(|(e, _stream)| e.into())
.and_then(|(stanza, stream)| {
match stanza {
Some(Packet::Stanza(stanza)) => {
if let Ok(challenge) = Challenge::try_from(stanza.clone()) {
let response = mechanism.response(&challenge.data);
Box::new(
response
.map_err(|e| AuthError::Sasl(e).into())
.into_future()
.and_then(|response| {
// Send response and loop
stream
.send_stanza(Response { data: response })
.map_err(Error::Io)
.and_then(|stream| {
Self::handle_challenge(stream, mechanism)
})
}),
)
} else if let Ok(_) = Success::try_from(stanza.clone()) {
Box::new(ok(stream))
} else if let Ok(failure) = Failure::try_from(stanza.clone()) {
Box::new(err(Error::Auth(AuthError::Fail(
failure.defined_condition,
))))
} else if stanza.name() == "failure" {
// Workaround for https://gitlab.com/xmpp-rs/xmpp-parsers/merge_requests/1
Box::new(err(Error::Auth(AuthError::Sasl("failure".to_string()))))
} else {
// ignore and loop
Self::handle_challenge(stream, mechanism)
}
}
Some(_) => {
// ignore and loop
Self::handle_challenge(stream, mechanism)
}
None => Box::new(err(Error::Disconnected)),
}
Some(_) => {
// ignore and loop
Self::handle_challenge(stream, mechanism)
}
None => Box::new(err(Error::Disconnected))
}
})
}),
)
}
}

2
tokio-xmpp/src/client/bind.rs
View file

@ -2,9 +2,9 @@ use futures::{sink, Async, Future, Poll, Stream};
use std::convert::TryFrom;
use std::mem::replace;
use tokio_io::{AsyncRead, AsyncWrite};
use xmpp_parsers::Jid;
use xmpp_parsers::bind::{BindQuery, BindResponse};
use xmpp_parsers::iq::{Iq, IqType};
use xmpp_parsers::Jid;
use crate::xmpp_codec::Packet;
use crate::xmpp_stream::XMPPStream;

15
tokio-xmpp/src/client/mod.rs
View file

@ -1,12 +1,12 @@
use futures::{done, Async, AsyncSink, Future, Poll, Sink, StartSend, Stream};
use idna;
use xmpp_parsers::{Jid, JidParseError};
use sasl::common::{ChannelBinding, Credentials};
use std::mem::replace;
use std::str::FromStr;
use tokio::net::TcpStream;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_tls::TlsStream;
use xmpp_parsers::{Jid, JidParseError};
use super::event::Event;
use super::happy_eyeballs::Connecter;
@ -205,10 +205,8 @@ impl Sink for Client {
fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
match self.state {
ClientState::Connected(ref mut stream) =>
Ok(stream.start_send(item)?),
_ =>
Ok(AsyncSink::NotReady(item)),
ClientState::Connected(ref mut stream) => Ok(stream.start_send(item)?),
_ => Ok(AsyncSink::NotReady(item)),
}
}
@ -226,11 +224,8 @@ impl Sink for Client {
/// incoming stream before closing the connection.
fn close(&mut self) -> Poll<(), Self::SinkError> {
match self.state {
ClientState::Connected(ref mut stream) =>
stream.close()
.map_err(|e| e.into()),
_ =>
Ok(Async::Ready(())),
ClientState::Connected(ref mut stream) => stream.close().map_err(|e| e.into()),
_ => Ok(Async::Ready(())),
}
}
}

2
tokio-xmpp/src/component/mod.rs
View file

@ -2,11 +2,11 @@
//! XMPP server under a JID consisting of just a domain name. They are
//! allowed to use any user and resource identifiers in their stanzas.
use futures::{done, Async, AsyncSink, Future, Poll, Sink, StartSend, Stream};
use xmpp_parsers::{Jid, JidParseError, Element};
use std::mem::replace;
use std::str::FromStr;
use tokio::net::TcpStream;
use tokio_io::{AsyncRead, AsyncWrite};
use xmpp_parsers::{Element, Jid, JidParseError};
use super::event::Event;
use super::happy_eyeballs::Connecter;

10
tokio-xmpp/src/error.rs
View file

@ -7,8 +7,8 @@ use std::str::Utf8Error;
use trust_dns_proto::error::ProtoError;
use trust_dns_resolver::error::ResolveError;
use xmpp_parsers::Error as ParsersError;
use xmpp_parsers::sasl::DefinedCondition as SaslDefinedCondition;
use xmpp_parsers::Error as ParsersError;
/// Top-level error type
#[derive(Debug)]
@ -159,8 +159,12 @@ impl fmt::Display for ProtocolError {
ProtocolError::Parser(e) => write!(fmt, "XML parser error: {}", e),
ProtocolError::Parsers(e) => write!(fmt, "error with expected stanza schema: {}", e),
ProtocolError::NoTls => write!(fmt, "no TLS available"),
ProtocolError::InvalidBindResponse => write!(fmt, "invalid response to resource binding"),
ProtocolError::NoStreamNamespace => write!(fmt, "no xmlns attribute in <stream:stream>"),
ProtocolError::InvalidBindResponse => {
write!(fmt, "invalid response to resource binding")
}
ProtocolError::NoStreamNamespace => {
write!(fmt, "no xmlns attribute in <stream:stream>")
}
ProtocolError::NoStreamId => write!(fmt, "no id attribute in <stream:stream>"),
ProtocolError::InvalidToken => write!(fmt, "encountered an unexpected XML token"),
ProtocolError::InvalidStreamStart => write!(fmt, "unexpected <stream:stream>"),

3
tokio-xmpp/src/happy_eyeballs.rs
View file

@ -8,11 +8,10 @@ use std::mem;
use std::net::SocketAddr;
use tokio::net::tcp::ConnectFuture;
use tokio::net::TcpStream;
use trust_dns_resolver::{AsyncResolver, Name, IntoName, Background, BackgroundLookup};
use trust_dns_resolver::config::LookupIpStrategy;
use trust_dns_resolver::lookup::SrvLookupFuture;
use trust_dns_resolver::lookup_ip::LookupIpFuture;
use trust_dns_resolver::{AsyncResolver, Background, BackgroundLookup, IntoName, Name};
enum State {
ResolveSrv(AsyncResolver, BackgroundLookup<SrvLookupFuture>),

2
tokio-xmpp/src/starttls.rs
View file

@ -1,11 +1,11 @@
use futures::sink;
use futures::stream::Stream;
use futures::{Async, Future, Poll, Sink};
use xmpp_parsers::{Jid, Element};
use native_tls::TlsConnector as NativeTlsConnector;
use std::mem::replace;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_tls::{Connect, TlsConnector, TlsStream};
use xmpp_parsers::{Element, Jid};
use crate::xmpp_codec::Packet;
use crate::xmpp_stream::XMPPStream;

2
tokio-xmpp/src/stream_start.rs
View file

@ -1,8 +1,8 @@
use futures::{sink, Async, Future, Poll, Sink, Stream};
use xmpp_parsers::{Jid, Element};
use std::mem::replace;
use tokio_codec::Framed;
use tokio_io::{AsyncRead, AsyncWrite};
use xmpp_parsers::{Element, Jid};
use crate::xmpp_codec::{Packet, XMPPCodec};
use crate::xmpp_stream::XMPPStream;

30
tokio-xmpp/src/xmpp_codec.rs
View file

@ -2,9 +2,9 @@
use crate::{ParseError, ParserError};
use bytes::{BufMut, BytesMut};
use xmpp_parsers::Element;
use quick_xml::Writer as EventWriter;
use std;
use std::borrow::Cow;
use std::cell::RefCell;
use std::collections::vec_deque::VecDeque;
use std::collections::HashMap;
@ -14,11 +14,11 @@ use std::io;
use std::iter::FromIterator;
use std::rc::Rc;
use std::str::from_utf8;
use std::borrow::Cow;
use tokio_codec::{Decoder, Encoder};
use xml5ever::buffer_queue::BufferQueue;
use xml5ever::interface::Attribute;
use xml5ever::tokenizer::{Tag, TagKind, Token, TokenSink, XmlTokenizer};
use xml5ever::buffer_queue::BufferQueue;
use xmpp_parsers::Element;
/// Anything that can be sent or received on an XMPP/XML stream
#[derive(Debug, Clone, PartialEq, Eq)]
@ -288,21 +288,17 @@ impl Encoder for XMPPCodec {
match item {
Packet::StreamStart(start_attrs) => {
let mut buf = String::new();
write!(buf, "<stream:stream")
.map_err(to_io_err)?;
write!(buf, "<stream:stream").map_err(to_io_err)?;
for (name, value) in start_attrs {
write!(buf, " {}=\"{}\"", escape(&name), escape(&value))
.map_err(to_io_err)?;
write!(buf, " {}=\"{}\"", escape(&name), escape(&value)).map_err(to_io_err)?;
if name == "xmlns" {
self.ns = Some(value);
}
}
write!(buf, ">\n")
.map_err(to_io_err)?;
write!(buf, ">\n").map_err(to_io_err)?;
// print!(">> {}", buf);
write!(dst, "{}", buf)
.map_err(to_io_err)
write!(dst, "{}", buf).map_err(to_io_err)
}
Packet::Stanza(stanza) => {
stanza
@ -321,10 +317,7 @@ impl Encoder for XMPPCodec {
})
.map_err(to_io_err)
}
Packet::StreamEnd => {
write!(dst, "</stream:stream>\n")
.map_err(to_io_err)
}
Packet::StreamEnd => write!(dst, "</stream:stream>\n").map_err(to_io_err),
}
}
}
@ -483,10 +476,13 @@ mod tests {
b.put(r"<status xml:lang='en'>Test status</status>");
let r = c.decode(&mut b);
assert!(match r {
Ok(Some(Packet::Stanza(ref el))) if el.name() == "status" && el.text() == "Test status" && el.attr("xml:lang").map_or(false, |a| a == "en") => true,
Ok(Some(Packet::Stanza(ref el)))
if el.name() == "status"
&& el.text() == "Test status"
&& el.attr("xml:lang").map_or(false, |a| a == "en") =>
true,
_ => false,
});
}
/// By default, encode() only get's a BytesMut that has 8kb space reserved.

2
tokio-xmpp/src/xmpp_stream.rs
View file

@ -2,9 +2,9 @@
use futures::sink::Send;
use futures::{Poll, Sink, StartSend, Stream};
use xmpp_parsers::{Jid, Element};
use tokio_codec::Framed;
use tokio_io::{AsyncRead, AsyncWrite};
use xmpp_parsers::{Element, Jid};
use crate::stream_start::StreamStart;
use crate::xmpp_codec::{Packet, XMPPCodec};