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

Merge pull request #7 from linkmauve/edition

Browse source 
Update to edition 2018, and run cargo fmt
This commit is contained in:
Astro 2018-12-20 02:04:33 +01:00 • committed by GitHub
commit ecd3b72f75
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 609 additions and 657 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.toml
View file

@ -9,6 +9,7 @@ repository = "https://github.com/astro/tokio-xmpp"
documentation = "https://docs.rs/tokio-xmpp" documentation = "https://docs.rs/tokio-xmpp"
categories = ["asynchronous", "network-programming"] categories = ["asynchronous", "network-programming"]
keywords = ["xmpp", "tokio"] keywords = ["xmpp", "tokio"]
edition = "2018"
[dependencies] [dependencies]
futures = "0.1" futures = "0.1"

32
examples/echo_bot.rs
View file

@ -1,21 +1,13 @@
extern crate futures; use futures::{future, Sink, Stream};
extern crate tokio; use jid::Jid;
extern crate tokio_xmpp; use minidom::Element;
extern crate jid;
extern crate minidom;
extern crate xmpp_parsers;
extern crate try_from;
use std::env::args; use std::env::args;
use std::process::exit; use std::process::exit;
use try_from::TryFrom;
use futures::{Stream, Sink, future};
use tokio::runtime::current_thread::Runtime; use tokio::runtime::current_thread::Runtime;
use tokio_xmpp::Client; use tokio_xmpp::Client;
use minidom::Element; use try_from::TryFrom;
use xmpp_parsers::presence::{Presence, Type as PresenceType, Show as PresenceShow}; use xmpp_parsers::message::{Body, Message, MessageType};
use xmpp_parsers::message::{Message, MessageType, Body}; use xmpp_parsers::presence::{Presence, Show as PresenceShow, Type as PresenceType};
use jid::Jid;
fn main() { fn main() {
let args: Vec<String> = args().collect(); let args: Vec<String> = args().collect();
@ -46,16 +38,18 @@ fn main() {
let presence = make_presence(); let presence = make_presence();
send(presence); send(presence);
} else if let Some(message) = event.into_stanza() } else if let Some(message) = event
.into_stanza()
.and_then(|stanza| Message::try_from(stanza).ok()) .and_then(|stanza| Message::try_from(stanza).ok())
{ {
// This is a message we'll echo // This is a message we'll echo
match (message.from, message.bodies.get("")) { match (message.from, message.bodies.get("")) {
(Some(from), Some(body)) => (Some(from), Some(body)) => {
if message.type_ != MessageType::Error { if message.type_ != MessageType::Error {
let reply = make_reply(from, &body.0); let reply = make_reply(from, &body.0);
send(reply); send(reply);
}, }
}
_ => (), _ => (),
} }
} }
@ -77,7 +71,9 @@ fn main() {
fn make_presence() -> Element { fn make_presence() -> Element {
let mut presence = Presence::new(PresenceType::None); let mut presence = Presence::new(PresenceType::None);
presence.show = PresenceShow::Chat; presence.show = PresenceShow::Chat;
presence.statuses.insert(String::from("en"), String::from("Echoing messages.")); presence
.statuses
.insert(String::from("en"), String::from("Echoing messages."));
presence.into() presence.into()
} }

43
examples/echo_component.rs
View file

@ -1,22 +1,14 @@
extern crate futures; use futures::{future, Sink, Stream};
extern crate tokio; use jid::Jid;
extern crate tokio_xmpp; use minidom::Element;
extern crate jid;
extern crate minidom;
extern crate xmpp_parsers;
extern crate try_from;
use std::env::args; use std::env::args;
use std::process::exit; use std::process::exit;
use std::str::FromStr; use std::str::FromStr;
use try_from::TryFrom;
use tokio::runtime::current_thread::Runtime; use tokio::runtime::current_thread::Runtime;
use futures::{Stream, Sink, future};
use tokio_xmpp::Component; use tokio_xmpp::Component;
use minidom::Element; use try_from::TryFrom;
use xmpp_parsers::presence::{Presence, Type as PresenceType, Show as PresenceShow}; use xmpp_parsers::message::{Body, Message, MessageType};
use xmpp_parsers::message::{Message, MessageType, Body}; use xmpp_parsers::presence::{Presence, Show as PresenceShow, Type as PresenceType};
use jid::Jid;
fn main() { fn main() {
let args: Vec<String> = args().collect(); let args: Vec<String> = args().collect();
@ -26,7 +18,11 @@ fn main() {
} }
let jid = &args[1]; let jid = &args[1];
let password = &args[2]; let password = &args[2];
let server = &args.get(3).unwrap().parse().unwrap_or("127.0.0.1".to_owned()); let server = &args
.get(3)
.unwrap()
.parse()
.unwrap_or("127.0.0.1".to_owned());
let port: u16 = args.get(4).unwrap().parse().unwrap_or(5347u16); let port: u16 = args.get(4).unwrap().parse().unwrap_or(5347u16);
// tokio_core context // tokio_core context
@ -50,18 +46,23 @@ fn main() {
println!("Online!"); println!("Online!");
// TODO: replace these hardcoded JIDs // TODO: replace these hardcoded JIDs
let presence = make_presence(Jid::from_str("test@component.linkmauve.fr/coucou").unwrap(), Jid::from_str("linkmauve@linkmauve.fr").unwrap()); let presence = make_presence(
Jid::from_str("test@component.linkmauve.fr/coucou").unwrap(),
Jid::from_str("linkmauve@linkmauve.fr").unwrap(),
);
send(presence); send(presence);
} else if let Some(message) = event.into_stanza() } else if let Some(message) = event
.into_stanza()
.and_then(|stanza| Message::try_from(stanza).ok()) .and_then(|stanza| Message::try_from(stanza).ok())
{ {
// This is a message we'll echo // This is a message we'll echo
match (message.from, message.bodies.get("")) { match (message.from, message.bodies.get("")) {
(Some(from), Some(body)) => (Some(from), Some(body)) => {
if message.type_ != MessageType::Error { if message.type_ != MessageType::Error {
let reply = make_reply(from, &body.0); let reply = make_reply(from, &body.0);
send(reply); send(reply);
}, }
}
_ => (), _ => (),
} }
} }
@ -85,7 +86,9 @@ fn make_presence(from: Jid, to: Jid) -> Element {
presence.from = Some(from); presence.from = Some(from);
presence.to = Some(to); presence.to = Some(to);
presence.show = PresenceShow::Chat; presence.show = PresenceShow::Chat;
presence.statuses.insert(String::from("en"), String::from("Echoing messages.")); presence
.statuses
.insert(String::from("en"), String::from("Echoing messages."));
presence.into() presence.into()
} }

134
src/client/auth.rs
View file

@ -1,23 +1,22 @@
use futures::{sink, Async, Future, Poll, Stream};
use minidom::Element;
use sasl::client::mechanisms::{Anonymous, Plain, Scram};
use sasl::client::Mechanism;
use sasl::common::scram::{Sha1, Sha256};
use sasl::common::Credentials;
use std::mem::replace; use std::mem::replace;
use std::str::FromStr; use std::str::FromStr;
use futures::{Future, Poll, Async, sink, Stream};
use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::{AsyncRead, AsyncWrite};
use sasl::common::Credentials;
use sasl::common::scram::{Sha1, Sha256};
use sasl::client::Mechanism;
use sasl::client::mechanisms::{Scram, Plain, Anonymous};
use minidom::Element;
use xmpp_parsers::sasl::{Auth, Challenge, Response, Success, Failure, Mechanism as XMPPMechanism};
use try_from::TryFrom; use try_from::TryFrom;
use xmpp_parsers::sasl::{Auth, Challenge, Failure, Mechanism as XMPPMechanism, Response, Success};
use xmpp_codec::Packet; use crate::stream_start::StreamStart;
use xmpp_stream::XMPPStream; use crate::xmpp_codec::Packet;
use stream_start::StreamStart; use crate::xmpp_stream::XMPPStream;
use {Error, AuthError, ProtocolError}; use crate::{AuthError, Error, ProtocolError};
const NS_XMPP_SASL: &str = "urn:ietf:params:xml:ns:xmpp-sasl"; const NS_XMPP_SASL: &str = "urn:ietf:params:xml:ns:xmpp-sasl";
pub struct ClientAuth<S: AsyncWrite> { pub struct ClientAuth<S: AsyncWrite> {
state: ClientAuthState<S>, state: ClientAuthState<S>,
mechanism: Box<Mechanism>, mechanism: Box<Mechanism>,
@ -39,8 +38,9 @@ impl<S: AsyncWrite> ClientAuth<S> {
Box::new(Anonymous::new()), Box::new(Anonymous::new()),
]; ];
let mech_names: Vec<String> = let mech_names: Vec<String> = stream
stream.stream_features.get_child("mechanisms", NS_XMPP_SASL) .stream_features
.get_child("mechanisms", NS_XMPP_SASL)
.ok_or(AuthError::NoMechanism)? .ok_or(AuthError::NoMechanism)?
.children() .children()
.filter(|child| child.is("mechanism", NS_XMPP_SASL)) .filter(|child| child.is("mechanism", NS_XMPP_SASL))
@ -52,20 +52,18 @@ impl<S: AsyncWrite> ClientAuth<S> {
let name = mech.name().to_owned(); let name = mech.name().to_owned();
if mech_names.iter().any(|name1| *name1 == name) { if mech_names.iter().any(|name1| *name1 == name) {
// println!("SASL mechanism selected: {:?}", name); // println!("SASL mechanism selected: {:?}", name);
let initial = mech.initial() let initial = mech.initial().map_err(AuthError::Sasl)?;
.map_err(AuthError::Sasl)?;
let mut this = ClientAuth { let mut this = ClientAuth {
state: ClientAuthState::Invalid, state: ClientAuthState::Invalid,
mechanism: mech, mechanism: mech,
}; };
let mechanism = XMPPMechanism::from_str(&name) let mechanism = XMPPMechanism::from_str(&name).map_err(ProtocolError::Parsers)?;
.map_err(ProtocolError::Parsers)?;
this.send( this.send(
stream, stream,
Auth { Auth {
mechanism, mechanism,
data: initial, data: initial,
} },
); );
return Ok(this); return Ok(this);
} }
@ -89,61 +87,55 @@ impl<S: AsyncRead + AsyncWrite> Future for ClientAuth<S> {
let state = replace(&mut self.state, ClientAuthState::Invalid); let state = replace(&mut self.state, ClientAuthState::Invalid);
match state { match state {
ClientAuthState::WaitSend(mut send) => ClientAuthState::WaitSend(mut send) => match send.poll() {
match send.poll() { Ok(Async::Ready(stream)) => {
Ok(Async::Ready(stream)) => { self.state = ClientAuthState::WaitRecv(stream);
self.state = ClientAuthState::WaitRecv(stream); self.poll()
}
Ok(Async::NotReady) => {
self.state = ClientAuthState::WaitSend(send);
Ok(Async::NotReady)
}
Err(e) => Err(e)?,
},
ClientAuthState::WaitRecv(mut stream) => match stream.poll() {
Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
if let Ok(challenge) = Challenge::try_from(stanza.clone()) {
let response = self
.mechanism
.response(&challenge.data)
.map_err(AuthError::Sasl)?;
self.send(stream, Response { data: response });
self.poll() self.poll()
}, } else if let Ok(_) = Success::try_from(stanza.clone()) {
Ok(Async::NotReady) => { let start = stream.restart();
self.state = ClientAuthState::WaitSend(send);
Ok(Async::NotReady)
},
Err(e) =>
Err(e)?,
},
ClientAuthState::WaitRecv(mut stream) =>
match stream.poll() {
Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
if let Ok(challenge) = Challenge::try_from(stanza.clone()) {
let response = self.mechanism.response(&challenge.data)
.map_err(AuthError::Sasl)?;
self.send(stream, Response { data: response });
self.poll()
} else if let Ok(_) = Success::try_from(stanza.clone()) {
let start = stream.restart();
self.state = ClientAuthState::Start(start);
self.poll()
} else if let Ok(failure) = Failure::try_from(stanza) {
Err(AuthError::Fail(failure.defined_condition))?
} else {
Ok(Async::NotReady)
}
}
Ok(Async::Ready(_event)) => {
// println!("ClientAuth ignore {:?}", _event);
Ok(Async::NotReady)
},
Ok(_) => {
self.state = ClientAuthState::WaitRecv(stream);
Ok(Async::NotReady)
},
Err(e) =>
Err(ProtocolError::Parser(e))?
},
ClientAuthState::Start(mut start) =>
match start.poll() {
Ok(Async::Ready(stream)) =>
Ok(Async::Ready(stream)),
Ok(Async::NotReady) => {
self.state = ClientAuthState::Start(start); self.state = ClientAuthState::Start(start);
self.poll()
} else if let Ok(failure) = Failure::try_from(stanza) {
Err(AuthError::Fail(failure.defined_condition))?
} else {
Ok(Async::NotReady) Ok(Async::NotReady)
}, }
Err(e) => }
Err(e) Ok(Async::Ready(_event)) => {
}, // println!("ClientAuth ignore {:?}", _event);
ClientAuthState::Invalid => Ok(Async::NotReady)
unreachable!(), }
Ok(_) => {
self.state = ClientAuthState::WaitRecv(stream);
Ok(Async::NotReady)
}
Err(e) => Err(ProtocolError::Parser(e))?,
},
ClientAuthState::Start(mut start) => match start.poll() {
Ok(Async::Ready(stream)) => Ok(Async::Ready(stream)),
Ok(Async::NotReady) => {
self.state = ClientAuthState::Start(start);
Ok(Async::NotReady)
}
Err(e) => Err(e),
},
ClientAuthState::Invalid => unreachable!(),
} }
} }
} }

113
src/client/bind.rs
View file

@ -1,13 +1,13 @@
use futures::{sink, Async, Future, Poll, Stream};
use std::mem::replace; use std::mem::replace;
use futures::{Future, Poll, Async, sink, Stream};
use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::{AsyncRead, AsyncWrite};
use xmpp_parsers::iq::{Iq, IqType};
use xmpp_parsers::bind::Bind;
use try_from::TryFrom; use try_from::TryFrom;
use xmpp_parsers::bind::Bind;
use xmpp_parsers::iq::{Iq, IqType};
use xmpp_codec::Packet; use crate::xmpp_codec::Packet;
use xmpp_stream::XMPPStream; use crate::xmpp_stream::XMPPStream;
use {Error, ProtocolError}; use crate::{Error, ProtocolError};
const NS_XMPP_BIND: &str = "urn:ietf:params:xml:ns:xmpp-bind"; const NS_XMPP_BIND: &str = "urn:ietf:params:xml:ns:xmpp-bind";
const BIND_REQ_ID: &str = "resource-bind"; const BIND_REQ_ID: &str = "resource-bind";
@ -26,16 +26,17 @@ impl<S: AsyncWrite> ClientBind<S> {
pub fn new(stream: XMPPStream<S>) -> Self { pub fn new(stream: XMPPStream<S>) -> Self {
match stream.stream_features.get_child("bind", NS_XMPP_BIND) { match stream.stream_features.get_child("bind", NS_XMPP_BIND) {
None => None =>
// No resource binding available, // No resource binding available,
// return the (probably // usable) stream immediately // return the (probably // usable) stream immediately
ClientBind::Unsupported(stream), {
ClientBind::Unsupported(stream)
}
Some(_) => { Some(_) => {
let resource = stream.jid.resource.clone(); let resource = stream.jid.resource.clone();
let iq = Iq::from_set(Bind::new(resource)) let iq = Iq::from_set(Bind::new(resource)).with_id(BIND_REQ_ID.to_string());
.with_id(BIND_REQ_ID.to_string());
let send = stream.send_stanza(iq); let send = stream.send_stanza(iq);
ClientBind::WaitSend(send) ClientBind::WaitSend(send)
}, }
} }
} }
} }
@ -48,59 +49,51 @@ impl<S: AsyncRead + AsyncWrite> Future for ClientBind<S> {
let state = replace(self, ClientBind::Invalid); let state = replace(self, ClientBind::Invalid);
match state { match state {
ClientBind::Unsupported(stream) => ClientBind::Unsupported(stream) => Ok(Async::Ready(stream)),
Ok(Async::Ready(stream)), ClientBind::WaitSend(mut send) => match send.poll() {
ClientBind::WaitSend(mut send) => { Ok(Async::Ready(stream)) => {
match send.poll() { replace(self, ClientBind::WaitRecv(stream));
Ok(Async::Ready(stream)) => { self.poll()
replace(self, ClientBind::WaitRecv(stream));
self.poll()
},
Ok(Async::NotReady) => {
replace(self, ClientBind::WaitSend(send));
Ok(Async::NotReady)
},
Err(e) =>
Err(e)?
} }
Ok(Async::NotReady) => {
replace(self, ClientBind::WaitSend(send));
Ok(Async::NotReady)
}
Err(e) => Err(e)?,
}, },
ClientBind::WaitRecv(mut stream) => { ClientBind::WaitRecv(mut stream) => match stream.poll() {
match stream.poll() { Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => match Iq::try_from(stanza) {
Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => Ok(iq) => {
match Iq::try_from(stanza) { if iq.id == Some(BIND_REQ_ID.to_string()) {
Ok(iq) => if iq.id == Some(BIND_REQ_ID.to_string()) { match iq.payload {
match iq.payload { IqType::Result(payload) => {
IqType::Result(payload) => { payload
payload .and_then(|payload| Bind::try_from(payload).ok())
.and_then(|payload| Bind::try_from(payload).ok()) .map(|bind| match bind {
.map(|bind| match bind { Bind::Jid(jid) => stream.jid = jid,
Bind::Jid(jid) => stream.jid = jid, _ => {}
_ => {} });
}); Ok(Async::Ready(stream))
Ok(Async::Ready(stream))
},
_ =>
Err(ProtocolError::InvalidBindResponse)?,
} }
} else { _ => Err(ProtocolError::InvalidBindResponse)?,
Ok(Async::NotReady) }
}, } else {
_ => Ok(Async::NotReady), Ok(Async::NotReady)
}, }
Ok(Async::Ready(_)) => { }
replace(self, ClientBind::WaitRecv(stream)); _ => Ok(Async::NotReady),
self.poll() },
}, Ok(Async::Ready(_)) => {
Ok(Async::NotReady) => { replace(self, ClientBind::WaitRecv(stream));
replace(self, ClientBind::WaitRecv(stream)); self.poll()
Ok(Async::NotReady)
},
Err(e) =>
Err(e)?,
} }
Ok(Async::NotReady) => {
replace(self, ClientBind::WaitRecv(stream));
Ok(Async::NotReady)
}
Err(e) => Err(e)?,
}, },
ClientBind::Invalid => ClientBind::Invalid => unreachable!(),
unreachable!(),
} }
} }
} }

142
src/client/mod.rs
View file

@ -1,19 +1,19 @@
use futures::{done, Async, AsyncSink, Future, Poll, Sink, StartSend, Stream};
use idna;
use jid::{Jid, JidParseError};
use minidom::Element;
use sasl::common::{ChannelBinding, Credentials};
use std::mem::replace; use std::mem::replace;
use std::str::FromStr; use std::str::FromStr;
use tokio::net::TcpStream; use tokio::net::TcpStream;
use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::{AsyncRead, AsyncWrite};
use tokio_tls::TlsStream; use tokio_tls::TlsStream;
use futures::{Future, Stream, Poll, Async, Sink, StartSend, AsyncSink, done};
use minidom::Element;
use jid::{Jid, JidParseError};
use sasl::common::{Credentials, ChannelBinding};
use idna;
use super::event::Event;
use super::happy_eyeballs::Connecter;
use super::starttls::{StartTlsClient, NS_XMPP_TLS};
use super::xmpp_codec::Packet; use super::xmpp_codec::Packet;
use super::xmpp_stream; use super::xmpp_stream;
use super::starttls::{NS_XMPP_TLS, StartTlsClient};
use super::happy_eyeballs::Connecter;
use super::event::Event;
use super::{Error, ProtocolError}; use super::{Error, ProtocolError};
mod auth; mod auth;
@ -34,7 +34,7 @@ const NS_JABBER_CLIENT: &str = "jabber:client";
enum ClientState { enum ClientState {
Invalid, Invalid,
Disconnected, Disconnected,
Connecting(Box<Future<Item=XMPPStream, Error=Error>>), Connecting(Box<Future<Item = XMPPStream, Error = Error>>),
Connected(XMPPStream), Connected(XMPPStream),
} }
@ -53,51 +53,62 @@ impl Client {
}) })
} }
fn make_connect(jid: Jid, password: String) -> impl Future<Item=XMPPStream, Error=Error> { fn make_connect(jid: Jid, password: String) -> impl Future<Item = XMPPStream, Error = Error> {
let username = jid.node.as_ref().unwrap().to_owned(); let username = jid.node.as_ref().unwrap().to_owned();
let jid1 = jid.clone(); let jid1 = jid.clone();
let jid2 = jid.clone(); let jid2 = jid.clone();
let password = password; let password = password;
done(idna::domain_to_ascii(&jid.domain)) done(idna::domain_to_ascii(&jid.domain))
.map_err(|_| Error::Idna) .map_err(|_| Error::Idna)
.and_then(|domain| .and_then(|domain| {
done(Connecter::from_lookup(&domain, Some("_xmpp-client._tcp"), 5222)) done(Connecter::from_lookup(
) &domain,
Some("_xmpp-client._tcp"),
5222,
))
})
.flatten() .flatten()
.and_then(move |tcp_stream| .and_then(move |tcp_stream| {
xmpp_stream::XMPPStream::start(tcp_stream, jid1, NS_JABBER_CLIENT.to_owned()) xmpp_stream::XMPPStream::start(tcp_stream, jid1, NS_JABBER_CLIENT.to_owned())
).and_then(|xmpp_stream| { })
.and_then(|xmpp_stream| {
if Self::can_starttls(&xmpp_stream) { if Self::can_starttls(&xmpp_stream) {
Ok(Self::starttls(xmpp_stream)) Ok(Self::starttls(xmpp_stream))
} else { } else {
Err(Error::Protocol(ProtocolError::NoTls)) Err(Error::Protocol(ProtocolError::NoTls))
} }
}).flatten() })
.and_then(|tls_stream| .flatten()
XMPPStream::start(tls_stream, jid2, NS_JABBER_CLIENT.to_owned()) .and_then(|tls_stream| XMPPStream::start(tls_stream, jid2, NS_JABBER_CLIENT.to_owned()))
).and_then(move |xmpp_stream| .and_then(
done(Self::auth(xmpp_stream, username, password)) move |xmpp_stream| done(Self::auth(xmpp_stream, username, password)), // TODO: flatten?
// TODO: flatten? )
).and_then(|auth| auth) .and_then(|auth| auth)
.and_then(|xmpp_stream| Self::bind(xmpp_stream))
.and_then(|xmpp_stream| { .and_then(|xmpp_stream| {
Self::bind(xmpp_stream)
}).and_then(|xmpp_stream| {
// println!("Bound to {}", xmpp_stream.jid); // println!("Bound to {}", xmpp_stream.jid);
Ok(xmpp_stream) Ok(xmpp_stream)
}) })
} }
fn can_starttls<S>(stream: &xmpp_stream::XMPPStream<S>) -> bool { fn can_starttls<S>(stream: &xmpp_stream::XMPPStream<S>) -> bool {
stream.stream_features stream
.stream_features
.get_child("starttls", NS_XMPP_TLS) .get_child("starttls", NS_XMPP_TLS)
.is_some() .is_some()
} }
fn starttls<S: AsyncRead + AsyncWrite>(stream: xmpp_stream::XMPPStream<S>) -> StartTlsClient<S> { fn starttls<S: AsyncRead + AsyncWrite>(
stream: xmpp_stream::XMPPStream<S>,
) -> StartTlsClient<S> {
StartTlsClient::from_stream(stream) StartTlsClient::from_stream(stream)
} }
fn auth<S: AsyncRead + AsyncWrite>(stream: xmpp_stream::XMPPStream<S>, username: String, password: String) -> Result<ClientAuth<S>, Error> { fn auth<S: AsyncRead + AsyncWrite>(
stream: xmpp_stream::XMPPStream<S>,
username: String,
password: String,
) -> Result<ClientAuth<S>, Error> {
let creds = Credentials::default() let creds = Credentials::default()
.with_username(username) .with_username(username)
.with_password(password) .with_password(password)
@ -118,31 +129,25 @@ impl Stream for Client {
let state = replace(&mut self.state, ClientState::Invalid); let state = replace(&mut self.state, ClientState::Invalid);
match state { match state {
ClientState::Invalid => ClientState::Invalid => Err(Error::InvalidState),
Err(Error::InvalidState), ClientState::Disconnected => Ok(Async::Ready(None)),
ClientState::Disconnected => ClientState::Connecting(mut connect) => match connect.poll() {
Ok(Async::Ready(None)), Ok(Async::Ready(stream)) => {
ClientState::Connecting(mut connect) => { self.state = ClientState::Connected(stream);
match connect.poll() { Ok(Async::Ready(Some(Event::Online)))
Ok(Async::Ready(stream)) => {
self.state = ClientState::Connected(stream);
Ok(Async::Ready(Some(Event::Online)))
},
Ok(Async::NotReady) => {
self.state = ClientState::Connecting(connect);
Ok(Async::NotReady)
},
Err(e) =>
Err(e),
} }
Ok(Async::NotReady) => {
self.state = ClientState::Connecting(connect);
Ok(Async::NotReady)
}
Err(e) => Err(e),
}, },
ClientState::Connected(mut stream) => { ClientState::Connected(mut stream) => {
// Poll sink // Poll sink
match stream.poll_complete() { match stream.poll_complete() {
Ok(Async::NotReady) => (), Ok(Async::NotReady) => (),
Ok(Async::Ready(())) => (), Ok(Async::Ready(())) => (),
Err(e) => Err(e) => return Err(e)?,
return Err(e)?,
}; };
// Poll stream // Poll stream
@ -151,20 +156,18 @@ impl Stream for Client {
// EOF // EOF
self.state = ClientState::Disconnected; self.state = ClientState::Disconnected;
Ok(Async::Ready(Some(Event::Disconnected))) Ok(Async::Ready(Some(Event::Disconnected)))
}, }
Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => { Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
self.state = ClientState::Connected(stream); self.state = ClientState::Connected(stream);
Ok(Async::Ready(Some(Event::Stanza(stanza)))) Ok(Async::Ready(Some(Event::Stanza(stanza))))
}, }
Ok(Async::NotReady) | Ok(Async::NotReady) | Ok(Async::Ready(_)) => {
Ok(Async::Ready(_)) => {
self.state = ClientState::Connected(stream); self.state = ClientState::Connected(stream);
Ok(Async::NotReady) Ok(Async::NotReady)
}, }
Err(e) => Err(e) => Err(e)?,
Err(e)?,
} }
}, }
} }
} }
} }
@ -175,30 +178,23 @@ impl Sink for Client {
fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> { fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
match self.state { match self.state {
ClientState::Connected(ref mut stream) => ClientState::Connected(ref mut stream) => match stream.start_send(Packet::Stanza(item))
match stream.start_send(Packet::Stanza(item)) { {
Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) => Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) => Ok(AsyncSink::NotReady(stanza)),
Ok(AsyncSink::NotReady(stanza)), Ok(AsyncSink::NotReady(_)) => {
Ok(AsyncSink::NotReady(_)) => panic!("Client.start_send with stanza but got something else back")
panic!("Client.start_send with stanza but got something else back"), }
Ok(AsyncSink::Ready) => { Ok(AsyncSink::Ready) => Ok(AsyncSink::Ready),
Ok(AsyncSink::Ready) Err(e) => Err(e)?,
}, },
Err(e) => _ => Ok(AsyncSink::NotReady(item)),
Err(e)?,
},
_ =>
Ok(AsyncSink::NotReady(item)),
} }
} }
fn poll_complete(&mut self) -> Poll<(), Self::SinkError> { fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
match self.state { match self.state {
ClientState::Connected(ref mut stream) => ClientState::Connected(ref mut stream) => stream.poll_complete().map_err(|e| e.into()),
stream.poll_complete() _ => Ok(Async::Ready(())),
.map_err(|e| e.into()),
_ =>
Ok(Async::Ready(())),
} }
} }
} }

83
src/component/auth.rs
View file

@ -1,11 +1,11 @@
use futures::{sink, Async, Future, Poll, Stream};
use std::mem::replace; use std::mem::replace;
use futures::{Future, Poll, Async, sink, Stream};
use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::{AsyncRead, AsyncWrite};
use xmpp_parsers::component::Handshake; use xmpp_parsers::component::Handshake;
use xmpp_codec::Packet; use crate::xmpp_codec::Packet;
use xmpp_stream::XMPPStream; use crate::xmpp_stream::XMPPStream;
use {Error, AuthError}; use crate::{AuthError, Error};
const NS_JABBER_COMPONENT_ACCEPT: &str = "jabber:component:accept"; const NS_JABBER_COMPONENT_ACCEPT: &str = "jabber:component:accept";
@ -29,7 +29,7 @@ impl<S: AsyncWrite> ComponentAuth<S> {
}; };
this.send( this.send(
stream, stream,
Handshake::from_password_and_stream_id(&password, &sid) Handshake::from_password_and_stream_id(&password, &sid),
); );
Ok(this) Ok(this)
} }
@ -50,45 +50,40 @@ impl<S: AsyncRead + AsyncWrite> Future for ComponentAuth<S> {
let state = replace(&mut self.state, ComponentAuthState::Invalid); let state = replace(&mut self.state, ComponentAuthState::Invalid);
match state { match state {
ComponentAuthState::WaitSend(mut send) => ComponentAuthState::WaitSend(mut send) => match send.poll() {
match send.poll() { Ok(Async::Ready(stream)) => {
Ok(Async::Ready(stream)) => { self.state = ComponentAuthState::WaitRecv(stream);
self.state = ComponentAuthState::WaitRecv(stream); self.poll()
self.poll() }
}, Ok(Async::NotReady) => {
Ok(Async::NotReady) => { self.state = ComponentAuthState::WaitSend(send);
self.state = ComponentAuthState::WaitSend(send); Ok(Async::NotReady)
Ok(Async::NotReady) }
}, Err(e) => Err(e)?,
Err(e) => },
Err(e)? ComponentAuthState::WaitRecv(mut stream) => match stream.poll() {
}, Ok(Async::Ready(Some(Packet::Stanza(ref stanza))))
ComponentAuthState::WaitRecv(mut stream) => if stanza.is("handshake", NS_JABBER_COMPONENT_ACCEPT) =>
match stream.poll() { {
Ok(Async::Ready(Some(Packet::Stanza(ref stanza)))) self.state = ComponentAuthState::Invalid;
if stanza.is("handshake", NS_JABBER_COMPONENT_ACCEPT) => Ok(Async::Ready(stream))
{ }
self.state = ComponentAuthState::Invalid; Ok(Async::Ready(Some(Packet::Stanza(ref stanza))))
Ok(Async::Ready(stream)) if stanza.is("error", "http://etherx.jabber.org/streams") =>
}, {
Ok(Async::Ready(Some(Packet::Stanza(ref stanza)))) Err(AuthError::ComponentFail.into())
if stanza.is("error", "http://etherx.jabber.org/streams") => }
{ Ok(Async::Ready(event)) => {
Err(AuthError::ComponentFail.into()) println!("ComponentAuth ignore {:?}", event);
}, Ok(Async::NotReady)
Ok(Async::Ready(event)) => { }
println!("ComponentAuth ignore {:?}", event); Ok(_) => {
Ok(Async::NotReady) self.state = ComponentAuthState::WaitRecv(stream);
}, Ok(Async::NotReady)
Ok(_) => { }
self.state = ComponentAuthState::WaitRecv(stream); Err(e) => Err(e)?,
Ok(Async::NotReady) },
}, ComponentAuthState::Invalid => unreachable!(),
Err(e) =>
Err(e)?
},
ComponentAuthState::Invalid =>
unreachable!(),
} }
} }
} }

110
src/component/mod.rs
View file

@ -1,18 +1,18 @@
//! Components in XMPP are services/gateways that are logged into an //! Components in XMPP are services/gateways that are logged into an
//! XMPP server under a JID consisting of just a domain name. They are //! XMPP server under a JID consisting of just a domain name. They are
//! allowed to use any user and resource identifiers in their stanzas. //! allowed to use any user and resource identifiers in their stanzas.
use futures::{done, Async, AsyncSink, Future, Poll, Sink, StartSend, Stream};
use jid::{Jid, JidParseError};
use minidom::Element;
use std::mem::replace; use std::mem::replace;
use std::str::FromStr; use std::str::FromStr;
use tokio::net::TcpStream; use tokio::net::TcpStream;
use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::{AsyncRead, AsyncWrite};
use futures::{Future, Stream, Poll, Async, Sink, StartSend, AsyncSink, done};
use minidom::Element;
use jid::{Jid, JidParseError};
use super::event::Event;
use super::happy_eyeballs::Connecter;
use super::xmpp_codec::Packet; use super::xmpp_codec::Packet;
use super::xmpp_stream; use super::xmpp_stream;
use super::happy_eyeballs::Connecter;
use super::event::Event;
use super::Error; use super::Error;
mod auth; mod auth;
@ -31,7 +31,7 @@ const NS_JABBER_COMPONENT_ACCEPT: &str = "jabber:component:accept";
enum ComponentState { enum ComponentState {
Invalid, Invalid,
Disconnected, Disconnected,
Connecting(Box<Future<Item=XMPPStream, Error=Error>>), Connecting(Box<Future<Item = XMPPStream, Error = Error>>),
Connected(XMPPStream), Connected(XMPPStream),
} }
@ -50,19 +50,30 @@ impl Component {
}) })
} }
fn make_connect(jid: Jid, password: String, server: &str, port: u16) -> impl Future<Item=XMPPStream, Error=Error> { fn make_connect(
jid: Jid,
password: String,
server: &str,
port: u16,
) -> impl Future<Item = XMPPStream, Error = Error> {
let jid1 = jid.clone(); let jid1 = jid.clone();
let password = password; let password = password;
done(Connecter::from_lookup(server, None, port)) done(Connecter::from_lookup(server, None, port))
.flatten() .flatten()
.and_then(move |tcp_stream| { .and_then(move |tcp_stream| {
xmpp_stream::XMPPStream::start(tcp_stream, jid1, NS_JABBER_COMPONENT_ACCEPT.to_owned()) xmpp_stream::XMPPStream::start(
}).and_then(move |xmpp_stream| { tcp_stream,
Self::auth(xmpp_stream, password).expect("auth") jid1,
NS_JABBER_COMPONENT_ACCEPT.to_owned(),
)
}) })
.and_then(move |xmpp_stream| Self::auth(xmpp_stream, password).expect("auth"))
} }
fn auth<S: AsyncRead + AsyncWrite>(stream: xmpp_stream::XMPPStream<S>, password: String) -> Result<ComponentAuth<S>, Error> { fn auth<S: AsyncRead + AsyncWrite>(
stream: xmpp_stream::XMPPStream<S>,
password: String,
) -> Result<ComponentAuth<S>, Error> {
ComponentAuth::new(stream, password) ComponentAuth::new(stream, password)
} }
} }
@ -75,31 +86,25 @@ impl Stream for Component {
let state = replace(&mut self.state, ComponentState::Invalid); let state = replace(&mut self.state, ComponentState::Invalid);
match state { match state {
ComponentState::Invalid => ComponentState::Invalid => Err(Error::InvalidState),
Err(Error::InvalidState), ComponentState::Disconnected => Ok(Async::Ready(None)),
ComponentState::Disconnected => ComponentState::Connecting(mut connect) => match connect.poll() {
Ok(Async::Ready(None)), Ok(Async::Ready(stream)) => {
ComponentState::Connecting(mut connect) => { self.state = ComponentState::Connected(stream);
match connect.poll() { Ok(Async::Ready(Some(Event::Online)))
Ok(Async::Ready(stream)) => {
self.state = ComponentState::Connected(stream);
Ok(Async::Ready(Some(Event::Online)))
},
Ok(Async::NotReady) => {
self.state = ComponentState::Connecting(connect);
Ok(Async::NotReady)
},
Err(e) =>
Err(e),
} }
Ok(Async::NotReady) => {
self.state = ComponentState::Connecting(connect);
Ok(Async::NotReady)
}
Err(e) => Err(e),
}, },
ComponentState::Connected(mut stream) => { ComponentState::Connected(mut stream) => {
// Poll sink // Poll sink
match stream.poll_complete() { match stream.poll_complete() {
Ok(Async::NotReady) => (), Ok(Async::NotReady) => (),
Ok(Async::Ready(())) => (), Ok(Async::Ready(())) => (),
Err(e) => Err(e) => return Err(e)?,
return Err(e)?,
}; };
// Poll stream // Poll stream
@ -107,24 +112,23 @@ impl Stream for Component {
Ok(Async::NotReady) => { Ok(Async::NotReady) => {
self.state = ComponentState::Connected(stream); self.state = ComponentState::Connected(stream);
Ok(Async::NotReady) Ok(Async::NotReady)
}, }
Ok(Async::Ready(None)) => { Ok(Async::Ready(None)) => {
// EOF // EOF
self.state = ComponentState::Disconnected; self.state = ComponentState::Disconnected;
Ok(Async::Ready(Some(Event::Disconnected))) Ok(Async::Ready(Some(Event::Disconnected)))
}, }
Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => { Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
self.state = ComponentState::Connected(stream); self.state = ComponentState::Connected(stream);
Ok(Async::Ready(Some(Event::Stanza(stanza)))) Ok(Async::Ready(Some(Event::Stanza(stanza))))
}, }
Ok(Async::Ready(_)) => { Ok(Async::Ready(_)) => {
self.state = ComponentState::Connected(stream); self.state = ComponentState::Connected(stream);
Ok(Async::NotReady) Ok(Async::NotReady)
}, }
Err(e) => Err(e) => Err(e)?,
Err(e)?,
} }
}, }
} }
} }
} }
@ -135,30 +139,26 @@ impl Sink for Component {
fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> { fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
match self.state { match self.state {
ComponentState::Connected(ref mut stream) => ComponentState::Connected(ref mut stream) => match stream
match stream.start_send(Packet::Stanza(item)) { .start_send(Packet::Stanza(item))
Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) => {
Ok(AsyncSink::NotReady(stanza)), Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) => Ok(AsyncSink::NotReady(stanza)),
Ok(AsyncSink::NotReady(_)) => Ok(AsyncSink::NotReady(_)) => {
panic!("Component.start_send with stanza but got something else back"), panic!("Component.start_send with stanza but got something else back")
Ok(AsyncSink::Ready) => { }
Ok(AsyncSink::Ready) Ok(AsyncSink::Ready) => Ok(AsyncSink::Ready),
}, Err(e) => Err(e)?,
Err(e) => },
Err(e)?, _ => Ok(AsyncSink::NotReady(item)),
},
_ =>
Ok(AsyncSink::NotReady(item)),
} }
} }
fn poll_complete(&mut self) -> Poll<(), Self::SinkError> { fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
match &mut self.state { match &mut self.state {
&mut ComponentState::Connected(ref mut stream) => &mut ComponentState::Connected(ref mut stream) => {
stream.poll_complete() stream.poll_complete().map_err(|e| e.into())
.map_err(|e| e.into()), }
_ => _ => Ok(Async::Ready(())),
Ok(Async::Ready(())),
} }
} }
} }

12
src/error.rs
View file

@ -1,11 +1,11 @@
use std::io::Error as IoError;
use std::error::Error as StdError;
use std::str::Utf8Error;
use std::borrow::Cow;
use std::fmt;
use native_tls::Error as TlsError; use native_tls::Error as TlsError;
use trust_dns_resolver::error::ResolveError; use std::borrow::Cow;
use std::error::Error as StdError;
use std::fmt;
use std::io::Error as IoError;
use std::str::Utf8Error;
use trust_dns_proto::error::ProtoError; use trust_dns_proto::error::ProtoError;
use trust_dns_resolver::error::ResolveError;
use xmpp_parsers::error::Error as ParsersError; use xmpp_parsers::error::Error as ParsersError;
use xmpp_parsers::sasl::DefinedCondition as SaslDefinedCondition; use xmpp_parsers::sasl::DefinedCondition as SaslDefinedCondition;

106
src/happy_eyeballs.rs
View file

@ -1,18 +1,18 @@
use std::mem; use crate::{ConnecterError, Error};
use std::io::Error as IoError; use futures::{Async, Future, Poll};
use std::net::SocketAddr; use std::cell::RefCell;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::collections::VecDeque; use std::collections::VecDeque;
use std::cell::RefCell; use std::io::Error as IoError;
use futures::{Future, Poll, Async}; use std::mem;
use tokio::net::TcpStream; use std::net::SocketAddr;
use tokio::net::tcp::ConnectFuture; use tokio::net::tcp::ConnectFuture;
use trust_dns_resolver::{IntoName, Name, ResolverFuture, error::ResolveError}; use tokio::net::TcpStream;
use trust_dns_resolver::config::LookupIpStrategy;
use trust_dns_resolver::lookup::SrvLookupFuture; use trust_dns_resolver::lookup::SrvLookupFuture;
use trust_dns_resolver::lookup_ip::LookupIpFuture; use trust_dns_resolver::lookup_ip::LookupIpFuture;
use trust_dns_resolver::system_conf; use trust_dns_resolver::system_conf;
use trust_dns_resolver::config::LookupIpStrategy; use trust_dns_resolver::{error::ResolveError, IntoName, Name, ResolverFuture};
use {Error, ConnecterError};
enum State { enum State {
AwaitResolver(Box<Future<Item = ResolverFuture, Error = ResolveError> + Send>), AwaitResolver(Box<Future<Item = ResolverFuture, Error = ResolveError> + Send>),
@ -31,23 +31,26 @@ pub struct Connecter {
error: Option<Error>, error: Option<Error>,
} }
fn resolver_future() -> Result<Box<Future<Item = ResolverFuture, Error = ResolveError> + Send>, IoError> { fn resolver_future(
) -> Result<Box<Future<Item = ResolverFuture, Error = ResolveError> + Send>, IoError> {
let (conf, mut opts) = system_conf::read_system_conf()?; let (conf, mut opts) = system_conf::read_system_conf()?;
opts.ip_strategy = LookupIpStrategy::Ipv4AndIpv6; opts.ip_strategy = LookupIpStrategy::Ipv4AndIpv6;
Ok(ResolverFuture::new(conf, opts)) Ok(ResolverFuture::new(conf, opts))
} }
impl Connecter { impl Connecter {
pub fn from_lookup(domain: &str, srv: Option<&str>, fallback_port: u16) -> Result<Connecter, Error> { pub fn from_lookup(
domain: &str,
srv: Option<&str>,
fallback_port: u16,
) -> Result<Connecter, Error> {
if let Ok(ip) = domain.parse() { if let Ok(ip) = domain.parse() {
// use specified IP address, not domain name, skip the whole dns part // use specified IP address, not domain name, skip the whole dns part
let connect = let connect = RefCell::new(TcpStream::connect(&SocketAddr::new(ip, fallback_port)));
RefCell::new(TcpStream::connect(&SocketAddr::new(ip, fallback_port)));
return Ok(Connecter { return Ok(Connecter {
fallback_port, fallback_port,
srv_domain: None, srv_domain: None,
domain: "nohost".into_name() domain: "nohost".into_name().map_err(ConnecterError::Dns)?,
.map_err(ConnecterError::Dns)?,
state: State::Connecting(None, vec![connect]), state: State::Connecting(None, vec![connect]),
targets: VecDeque::new(), targets: VecDeque::new(),
error: None, error: None,
@ -56,20 +59,18 @@ impl Connecter {
let state = State::AwaitResolver(resolver_future()?); let state = State::AwaitResolver(resolver_future()?);
let srv_domain = match srv { let srv_domain = match srv {
Some(srv) => Some(srv) => Some(
Some(format!("{}.{}.", srv, domain) format!("{}.{}.", srv, domain)
.into_name() .into_name()
.map_err(ConnecterError::Dns)? .map_err(ConnecterError::Dns)?,
), ),
None => None => None,
None,
}; };
Ok(Connecter { Ok(Connecter {
fallback_port, fallback_port,
srv_domain, srv_domain,
domain: domain.into_name() domain: domain.into_name().map_err(ConnecterError::Dns)?,
.map_err(ConnecterError::Dns)?,
state, state,
targets: VecDeque::new(), targets: VecDeque::new(),
error: None, error: None,
@ -97,8 +98,8 @@ impl Future for Connecter {
self.state = State::ResolveSrv(resolver, srv_lookup); self.state = State::ResolveSrv(resolver, srv_lookup);
} }
None => { None => {
self.targets = self.targets = [(self.domain.clone(), self.fallback_port)]
[(self.domain.clone(), self.fallback_port)].into_iter() .into_iter()
.cloned() .cloned()
.collect(); .collect();
self.state = State::Connecting(Some(resolver), vec![]); self.state = State::Connecting(Some(resolver), vec![]);
@ -115,22 +116,19 @@ impl Future for Connecter {
Ok(Async::NotReady) Ok(Async::NotReady)
} }
Ok(Async::Ready(srv_result)) => { Ok(Async::Ready(srv_result)) => {
let mut srv_map: BTreeMap<_, _> = let srv_map: BTreeMap<_, _> = srv_result
srv_result.iter() .iter()
.map(|srv| (srv.priority(), (srv.target().clone(), srv.port()))) .map(|srv| (srv.priority(), (srv.target().clone(), srv.port())))
.collect(); .collect();
let targets = let targets = srv_map.into_iter().map(|(_, tp)| tp).collect();
srv_map.into_iter()
.map(|(_, tp)| tp)
.collect();
self.targets = targets; self.targets = targets;
self.state = State::Connecting(Some(resolver), vec![]); self.state = State::Connecting(Some(resolver), vec![]);
self.poll() self.poll()
} }
Err(_) => { Err(_) => {
// ignore, fallback // ignore, fallback
self.targets = self.targets = [(self.domain.clone(), self.fallback_port)]
[(self.domain.clone(), self.fallback_port)].into_iter() .into_iter()
.cloned() .cloned()
.collect(); .collect();
self.state = State::Connecting(Some(resolver), vec![]); self.state = State::Connecting(Some(resolver), vec![]);
@ -147,36 +145,31 @@ impl Future for Connecter {
self.poll() self.poll()
} else if connects.len() > 0 { } else if connects.len() > 0 {
let mut success = None; let mut success = None;
connects.retain(|connect| { connects.retain(|connect| match connect.borrow_mut().poll() {
match connect.borrow_mut().poll() { Ok(Async::NotReady) => true,
Ok(Async::NotReady) => true, Ok(Async::Ready(connection)) => {
Ok(Async::Ready(connection)) => { success = Some(connection);
success = Some(connection); false
false }
Err(e) => {
if self.error.is_none() {
self.error = Some(e.into());
} }
Err(e) => { false
if self.error.is_none() {
self.error = Some(e.into());
}
false
},
} }
}); });
match success { match success {
Some(connection) => Some(connection) => Ok(Async::Ready(connection)),
Ok(Async::Ready(connection)),
None => { None => {
self.state = State::Connecting(resolver, connects); self.state = State::Connecting(resolver, connects);
Ok(Async::NotReady) Ok(Async::NotReady)
}, }
} }
} else { } else {
// All targets tried // All targets tried
match self.error.take() { match self.error.take() {
None => None => Err(ConnecterError::AllFailed.into()),
Err(ConnecterError::AllFailed.into()), Some(e) => Err(e),
Some(e) =>
Err(e),
} }
} }
} }
@ -187,8 +180,8 @@ impl Future for Connecter {
Ok(Async::NotReady) Ok(Async::NotReady)
} }
Ok(Async::Ready(ip_result)) => { Ok(Async::Ready(ip_result)) => {
let connects = let connects = ip_result
ip_result.iter() .iter()
.map(|ip| RefCell::new(TcpStream::connect(&SocketAddr::new(ip, port)))) .map(|ip| RefCell::new(TcpStream::connect(&SocketAddr::new(ip, port))))
.collect(); .collect();
self.state = State::Connecting(Some(resolver), connects); self.state = State::Connecting(Some(resolver), connects);
@ -204,8 +197,7 @@ impl Future for Connecter {
} }
} }
} }
_ => panic!("") _ => panic!(""),
} }
} }
} }

33
src/lib.rs
View file

@ -2,37 +2,20 @@
//! XMPP implemeentation with asynchronous I/O using Tokio. //! XMPP implemeentation with asynchronous I/O using Tokio.
extern crate futures;
extern crate tokio;
extern crate tokio_io;
extern crate tokio_codec;
extern crate bytes;
extern crate xml5ever;
extern crate quick_xml;
extern crate minidom;
extern crate native_tls;
extern crate tokio_tls;
extern crate sasl;
extern crate jid;
extern crate trust_dns_resolver;
extern crate trust_dns_proto;
extern crate idna;
extern crate xmpp_parsers;
extern crate try_from;
#[macro_use] #[macro_use]
extern crate derive_error; extern crate derive_error;
mod starttls;
mod stream_start;
pub mod xmpp_codec; pub mod xmpp_codec;
pub mod xmpp_stream; pub mod xmpp_stream;
mod stream_start; pub use crate::starttls::StartTlsClient;
mod starttls;
pub use starttls::StartTlsClient;
mod happy_eyeballs;
mod event; mod event;
pub use event::Event; mod happy_eyeballs;
pub use crate::event::Event;
mod client; mod client;
pub use client::Client; pub use crate::client::Client;
mod component; mod component;
pub use component::Component; pub use crate::component::Component;
mod error; mod error;
pub use error::{Error, ProtocolError, AuthError, ConnecterError, ParseError, ParserError}; pub use crate::error::{AuthError, ConnecterError, Error, ParseError, ParserError, ProtocolError};

118
src/starttls.rs
View file

@ -1,21 +1,20 @@
use std::mem::replace;
use futures::{Future, Sink, Poll, Async};
use futures::stream::Stream;
use futures::sink; use futures::sink;
use tokio_io::{AsyncRead, AsyncWrite}; use futures::stream::Stream;
use tokio_tls::{TlsStream, TlsConnector, Connect}; use futures::{Async, Future, Poll, Sink};
use native_tls::TlsConnector as NativeTlsConnector;
use minidom::Element;
use jid::Jid; use jid::Jid;
use minidom::Element;
use native_tls::TlsConnector as NativeTlsConnector;
use std::mem::replace;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_tls::{Connect, TlsConnector, TlsStream};
use xmpp_codec::Packet; use crate::xmpp_codec::Packet;
use xmpp_stream::XMPPStream; use crate::xmpp_stream::XMPPStream;
use Error; use crate::Error;
/// XMPP TLS XML namespace /// XMPP TLS XML namespace
pub const NS_XMPP_TLS: &str = "urn:ietf:params:xml:ns:xmpp-tls"; pub const NS_XMPP_TLS: &str = "urn:ietf:params:xml:ns:xmpp-tls";
/// XMPP stream that switches to TLS if available in received features /// XMPP stream that switches to TLS if available in received features
pub struct StartTlsClient<S: AsyncRead + AsyncWrite> { pub struct StartTlsClient<S: AsyncRead + AsyncWrite> {
state: StartTlsClientState<S>, state: StartTlsClientState<S>,
@ -34,9 +33,7 @@ impl<S: AsyncRead + AsyncWrite> StartTlsClient<S> {
pub fn from_stream(xmpp_stream: XMPPStream<S>) -> Self { pub fn from_stream(xmpp_stream: XMPPStream<S>) -> Self {
let jid = xmpp_stream.jid.clone(); let jid = xmpp_stream.jid.clone();
let nonza = Element::builder("starttls") let nonza = Element::builder("starttls").ns(NS_XMPP_TLS).build();
.ns(NS_XMPP_TLS)
.build();
let packet = Packet::Stanza(nonza); let packet = Packet::Stanza(nonza);
let send = xmpp_stream.send(packet); let send = xmpp_stream.send(packet);
@ -56,51 +53,56 @@ impl<S: AsyncRead + AsyncWrite> Future for StartTlsClient<S> {
let mut retry = false; let mut retry = false;
let (new_state, result) = match old_state { let (new_state, result) = match old_state {
StartTlsClientState::SendStartTls(mut send) => StartTlsClientState::SendStartTls(mut send) => match send.poll() {
match send.poll() { Ok(Async::Ready(xmpp_stream)) => {
Ok(Async::Ready(xmpp_stream)) => { let new_state = StartTlsClientState::AwaitProceed(xmpp_stream);
let new_state = StartTlsClientState::AwaitProceed(xmpp_stream); retry = true;
retry = true; (new_state, Ok(Async::NotReady))
(new_state, Ok(Async::NotReady)) }
}, Ok(Async::NotReady) => {
Ok(Async::NotReady) => (StartTlsClientState::SendStartTls(send), Ok(Async::NotReady))
(StartTlsClientState::SendStartTls(send), Ok(Async::NotReady)), }
Err(e) => Err(e) => (StartTlsClientState::SendStartTls(send), Err(e.into())),
(StartTlsClientState::SendStartTls(send), Err(e.into())), },
}, StartTlsClientState::AwaitProceed(mut xmpp_stream) => match xmpp_stream.poll() {
StartTlsClientState::AwaitProceed(mut xmpp_stream) => Ok(Async::Ready(Some(Packet::Stanza(ref stanza))))
match xmpp_stream.poll() { if stanza.name() == "proceed" =>
Ok(Async::Ready(Some(Packet::Stanza(ref stanza)))) {
if stanza.name() == "proceed" => let stream = xmpp_stream.stream.into_inner();
{ let connect =
let stream = xmpp_stream.stream.into_inner(); TlsConnector::from(NativeTlsConnector::builder().build().unwrap())
let connect = TlsConnector::from(NativeTlsConnector::builder()
.build().unwrap())
.connect(&self.jid.domain, stream); .connect(&self.jid.domain, stream);
let new_state = StartTlsClientState::StartingTls(connect); let new_state = StartTlsClientState::StartingTls(connect);
retry = true; retry = true;
(new_state, Ok(Async::NotReady)) (new_state, Ok(Async::NotReady))
}, }
Ok(Async::Ready(value)) => { Ok(Async::Ready(value)) => {
println!("StartTlsClient ignore {:?}", value); println!("StartTlsClient ignore {:?}", value);
(StartTlsClientState::AwaitProceed(xmpp_stream), Ok(Async::NotReady)) (
}, StartTlsClientState::AwaitProceed(xmpp_stream),
Ok(_) => Ok(Async::NotReady),
(StartTlsClientState::AwaitProceed(xmpp_stream), Ok(Async::NotReady)), )
Err(e) => }
(StartTlsClientState::AwaitProceed(xmpp_stream), Err(Error::Protocol(e.into()))), Ok(_) => (
}, StartTlsClientState::AwaitProceed(xmpp_stream),
StartTlsClientState::StartingTls(mut connect) => Ok(Async::NotReady),
match connect.poll() { ),
Ok(Async::Ready(tls_stream)) => Err(e) => (
(StartTlsClientState::Invalid, Ok(Async::Ready(tls_stream))), StartTlsClientState::AwaitProceed(xmpp_stream),
Ok(Async::NotReady) => Err(Error::Protocol(e.into())),
(StartTlsClientState::StartingTls(connect), Ok(Async::NotReady)), ),
Err(e) => },
(StartTlsClientState::Invalid, Err(e.into())), StartTlsClientState::StartingTls(mut connect) => match connect.poll() {
}, Ok(Async::Ready(tls_stream)) => {
StartTlsClientState::Invalid => (StartTlsClientState::Invalid, Ok(Async::Ready(tls_stream)))
unreachable!(), }
Ok(Async::NotReady) => (
StartTlsClientState::StartingTls(connect),
Ok(Async::NotReady),
),
Err(e) => (StartTlsClientState::Invalid, Err(e.into())),
},
StartTlsClientState::Invalid => unreachable!(),
}; };
self.state = new_state; self.state = new_state;

137
src/stream_start.rs
View file

@ -1,13 +1,13 @@
use std::mem::replace; use futures::{sink, Async, Future, Poll, Sink, Stream};
use futures::{Future, Async, Poll, Stream, sink, Sink};
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_codec::Framed;
use jid::Jid; use jid::Jid;
use minidom::Element; use minidom::Element;
use std::mem::replace;
use tokio_codec::Framed;
use tokio_io::{AsyncRead, AsyncWrite};
use xmpp_codec::{XMPPCodec, Packet}; use crate::xmpp_codec::{Packet, XMPPCodec};
use xmpp_stream::XMPPStream; use crate::xmpp_stream::XMPPStream;
use {Error, ProtocolError}; use crate::{Error, ProtocolError};
const NS_XMPP_STREAM: &str = "http://etherx.jabber.org/streams"; const NS_XMPP_STREAM: &str = "http://etherx.jabber.org/streams";
@ -26,11 +26,15 @@ enum StreamStartState<S: AsyncWrite> {
impl<S: AsyncWrite> StreamStart<S> { impl<S: AsyncWrite> StreamStart<S> {
pub fn from_stream(stream: Framed<S, XMPPCodec>, jid: Jid, ns: String) -> Self { pub fn from_stream(stream: Framed<S, XMPPCodec>, jid: Jid, ns: String) -> Self {
let attrs = [("to".to_owned(), jid.domain.clone()), let attrs = [
("version".to_owned(), "1.0".to_owned()), ("to".to_owned(), jid.domain.clone()),
("xmlns".to_owned(), ns.clone()), ("version".to_owned(), "1.0".to_owned()),
("xmlns:stream".to_owned(), NS_XMPP_STREAM.to_owned()), ("xmlns".to_owned(), ns.clone()),
].iter().cloned().collect(); ("xmlns:stream".to_owned(), NS_XMPP_STREAM.to_owned()),
]
.iter()
.cloned()
.collect();
let send = stream.send(Packet::StreamStart(attrs)); let send = stream.send(Packet::StreamStart(attrs));
StreamStart { StreamStart {
@ -50,59 +54,66 @@ impl<S: AsyncRead + AsyncWrite> Future for StreamStart<S> {
let mut retry = false; let mut retry = false;
let (new_state, result) = match old_state { let (new_state, result) = match old_state {
StreamStartState::SendStart(mut send) => StreamStartState::SendStart(mut send) => match send.poll() {
match send.poll() { Ok(Async::Ready(stream)) => {
Ok(Async::Ready(stream)) => { retry = true;
(StreamStartState::RecvStart(stream), Ok(Async::NotReady))
}
Ok(Async::NotReady) => (StreamStartState::SendStart(send), Ok(Async::NotReady)),
Err(e) => (StreamStartState::Invalid, Err(e.into())),
},
StreamStartState::RecvStart(mut stream) => match stream.poll() {
Ok(Async::Ready(Some(Packet::StreamStart(stream_attrs)))) => {
let stream_ns = stream_attrs
.get("xmlns")
.ok_or(ProtocolError::NoStreamNamespace)?
.clone();
if self.ns == "jabber:client" {
retry = true; retry = true;
(StreamStartState::RecvStart(stream), Ok(Async::NotReady)) // TODO: skip RecvFeatures for version < 1.0
}, (
Ok(Async::NotReady) => StreamStartState::RecvFeatures(stream, stream_ns),
(StreamStartState::SendStart(send), Ok(Async::NotReady)), Ok(Async::NotReady),
Err(e) => )
(StreamStartState::Invalid, Err(e.into())), } else {
}, let id = stream_attrs
StreamStartState::RecvStart(mut stream) => .get("id")
match stream.poll() { .ok_or(ProtocolError::NoStreamId)?
Ok(Async::Ready(Some(Packet::StreamStart(stream_attrs)))) => {
let stream_ns = stream_attrs.get("xmlns")
.ok_or(ProtocolError::NoStreamNamespace)?
.clone(); .clone();
if self.ns == "jabber:client" { // FIXME: huge hack, shouldn’t be an element!
retry = true; let stream = XMPPStream::new(
// TODO: skip RecvFeatures for version < 1.0 self.jid.clone(),
(StreamStartState::RecvFeatures(stream, stream_ns), Ok(Async::NotReady)) stream,
} else { self.ns.clone(),
let id = stream_attrs.get("id") Element::builder(id).build(),
.ok_or(ProtocolError::NoStreamId)? );
.clone(); (StreamStartState::Invalid, Ok(Async::Ready(stream)))
// FIXME: huge hack, shouldn’t be an element! }
let stream = XMPPStream::new(self.jid.clone(), stream, self.ns.clone(), Element::builder(id).build()); }
(StreamStartState::Invalid, Ok(Async::Ready(stream))) Ok(Async::Ready(_)) => return Err(ProtocolError::InvalidToken.into()),
} Ok(Async::NotReady) => (StreamStartState::RecvStart(stream), Ok(Async::NotReady)),
}, Err(e) => return Err(ProtocolError::from(e).into()),
Ok(Async::Ready(_)) => },
return Err(ProtocolError::InvalidToken.into()), StreamStartState::RecvFeatures(mut stream, stream_ns) => match stream.poll() {
Ok(Async::NotReady) => Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
(StreamStartState::RecvStart(stream), Ok(Async::NotReady)), if stanza.is("features", NS_XMPP_STREAM) {
Err(e) => let stream =
return Err(ProtocolError::from(e).into()), XMPPStream::new(self.jid.clone(), stream, self.ns.clone(), stanza);
}, (StreamStartState::Invalid, Ok(Async::Ready(stream)))
StreamStartState::RecvFeatures(mut stream, stream_ns) => } else {
match stream.poll() { (
Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => StreamStartState::RecvFeatures(stream, stream_ns),
if stanza.is("features", NS_XMPP_STREAM) { Ok(Async::NotReady),
let stream = XMPPStream::new(self.jid.clone(), stream, self.ns.clone(), stanza); )
(StreamStartState::Invalid, Ok(Async::Ready(stream))) }
} else { }
(StreamStartState::RecvFeatures(stream, stream_ns), Ok(Async::NotReady)) Ok(Async::Ready(_)) | Ok(Async::NotReady) => (
}, StreamStartState::RecvFeatures(stream, stream_ns),
Ok(Async::Ready(_)) | Ok(Async::NotReady) => Ok(Async::NotReady),
(StreamStartState::RecvFeatures(stream, stream_ns), Ok(Async::NotReady)), ),
Err(e) => Err(e) => return Err(ProtocolError::from(e).into()),
return Err(ProtocolError::from(e).into()), },
}, StreamStartState::Invalid => unreachable!(),
StreamStartState::Invalid =>
unreachable!(),
}; };
self.state = new_state; self.state = new_state;

183
src/xmpp_codec.rs
View file

@ -1,22 +1,22 @@
//! XML stream parser for XMPP //! XML stream parser for XMPP
use std; use crate::{ParseError, ParserError};
use std::default::Default; use bytes::{BufMut, BytesMut};
use std::iter::FromIterator;
use std::cell::RefCell;
use std::rc::Rc;
use std::fmt::Write;
use std::str::from_utf8;
use std::io;
use std::collections::HashMap;
use std::collections::vec_deque::VecDeque;
use tokio_codec::{Encoder, Decoder};
use minidom::Element; use minidom::Element;
use xml5ever::tokenizer::{XmlTokenizer, TokenSink, Token, Tag, TagKind};
use xml5ever::interface::Attribute;
use bytes::{BytesMut, BufMut};
use quick_xml::Writer as EventWriter; use quick_xml::Writer as EventWriter;
use {ParserError, ParseError}; use std;
use std::cell::RefCell;
use std::collections::vec_deque::VecDeque;
use std::collections::HashMap;
use std::default::Default;
use std::fmt::Write;
use std::io;
use std::iter::FromIterator;
use std::rc::Rc;
use std::str::from_utf8;
use tokio_codec::{Decoder, Encoder};
use xml5ever::interface::Attribute;
use xml5ever::tokenizer::{Tag, TagKind, Token, TokenSink, XmlTokenizer};
/// Anything that can be sent or received on an XMPP/XML stream /// Anything that can be sent or received on an XMPP/XML stream
#[derive(Debug)] #[derive(Debug)]
@ -72,17 +72,21 @@ impl ParserSink {
fn handle_start_tag(&mut self, tag: Tag) { fn handle_start_tag(&mut self, tag: Tag) {
let mut nss = HashMap::new(); let mut nss = HashMap::new();
let is_prefix_xmlns = |attr: &Attribute| attr.name.prefix.as_ref() let is_prefix_xmlns = |attr: &Attribute| {
.map(|prefix| prefix.eq_str_ignore_ascii_case("xmlns")) attr.name
.unwrap_or(false); .prefix
.as_ref()
.map(|prefix| prefix.eq_str_ignore_ascii_case("xmlns"))
.unwrap_or(false)
};
for attr in &tag.attrs { for attr in &tag.attrs {
match attr.name.local.as_ref() { match attr.name.local.as_ref() {
"xmlns" => { "xmlns" => {
nss.insert(None, attr.value.as_ref().to_owned()); nss.insert(None, attr.value.as_ref().to_owned());
}, }
prefix if is_prefix_xmlns(attr) => { prefix if is_prefix_xmlns(attr) => {
nss.insert(Some(prefix.to_owned()), attr.value.as_ref().to_owned()); nss.insert(Some(prefix.to_owned()), attr.value.as_ref().to_owned());
}, }
_ => (), _ => (),
} }
} }
@ -90,10 +94,9 @@ impl ParserSink {
let el = { let el = {
let mut el_builder = Element::builder(tag.name.local.as_ref()); let mut el_builder = Element::builder(tag.name.local.as_ref());
if let Some(el_ns) = self.lookup_ns( if let Some(el_ns) =
&tag.name.prefix.map(|prefix| self.lookup_ns(&tag.name.prefix.map(|prefix| prefix.as_ref().to_owned()))
prefix.as_ref().to_owned()) {
) {
el_builder = el_builder.ns(el_ns); el_builder = el_builder.ns(el_ns);
} }
for attr in &tag.attrs { for attr in &tag.attrs {
@ -101,21 +104,20 @@ impl ParserSink {
"xmlns" => (), "xmlns" => (),
_ if is_prefix_xmlns(attr) => (), _ if is_prefix_xmlns(attr) => (),
_ => { _ => {
el_builder = el_builder.attr( el_builder = el_builder.attr(attr.name.local.as_ref(), attr.value.as_ref());
attr.name.local.as_ref(), }
attr.value.as_ref()
);
},
} }
} }
el_builder.build() el_builder.build()
}; };
if self.stack.is_empty() { if self.stack.is_empty() {
let attrs = HashMap::from_iter( let attrs = HashMap::from_iter(tag.attrs.iter().map(|attr| {
tag.attrs.iter() (
.map(|attr| (attr.name.local.as_ref().to_owned(), attr.value.as_ref().to_owned())) attr.name.local.as_ref().to_owned(),
); attr.value.as_ref().to_owned(),
)
}));
self.push_queue(Packet::StreamStart(attrs)); self.push_queue(Packet::StreamStart(attrs));
} }
@ -128,15 +130,13 @@ impl ParserSink {
match self.stack.len() { match self.stack.len() {
// </stream:stream> // </stream:stream>
0 => 0 => self.push_queue(Packet::StreamEnd),
self.push_queue(Packet::StreamEnd),
// </stanza> // </stanza>
1 => 1 => self.push_queue(Packet::Stanza(el)),
self.push_queue(Packet::Stanza(el)),
len => { len => {
let parent = &mut self.stack[len - 1]; let parent = &mut self.stack[len - 1];
parent.append_child(el); parent.append_child(el);
}, }
} }
} }
} }
@ -145,32 +145,26 @@ impl TokenSink for ParserSink {
fn process_token(&mut self, token: Token) { fn process_token(&mut self, token: Token) {
match token { match token {
Token::TagToken(tag) => match tag.kind { Token::TagToken(tag) => match tag.kind {
TagKind::StartTag => TagKind::StartTag => self.handle_start_tag(tag),
self.handle_start_tag(tag), TagKind::EndTag => self.handle_end_tag(),
TagKind::EndTag =>
self.handle_end_tag(),
TagKind::EmptyTag => { TagKind::EmptyTag => {
self.handle_start_tag(tag); self.handle_start_tag(tag);
self.handle_end_tag(); self.handle_end_tag();
}, }
TagKind::ShortTag => TagKind::ShortTag => self.push_queue_error(ParserError::ShortTag),
self.push_queue_error(ParserError::ShortTag),
}, },
Token::CharacterTokens(tendril) => Token::CharacterTokens(tendril) => match self.stack.len() {
match self.stack.len() { 0 | 1 => self.push_queue(Packet::Text(tendril.into())),
0 | 1 => len => {
self.push_queue(Packet::Text(tendril.into())), let el = &mut self.stack[len - 1];
len => { el.append_text_node(tendril);
let el = &mut self.stack[len - 1]; }
el.append_text_node(tendril); },
}, Token::EOFToken => self.push_queue(Packet::StreamEnd),
},
Token::EOFToken =>
self.push_queue(Packet::StreamEnd),
Token::ParseError(s) => { Token::ParseError(s) => {
// println!("ParseError: {:?}", s); // println!("ParseError: {:?}", s);
self.push_queue_error(ParserError::Parse(ParseError(s))); self.push_queue_error(ParserError::Parse(ParseError(s)));
}, }
_ => (), _ => (),
} }
} }
@ -219,23 +213,22 @@ impl Decoder for XMPPCodec {
type Error = ParserError; type Error = ParserError;
fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> { fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
let buf1: Box<AsRef<[u8]>> = let buf1: Box<AsRef<[u8]>> = if !self.buf.is_empty() && !buf.is_empty() {
if ! self.buf.is_empty() && ! buf.is_empty() { let mut prefix = std::mem::replace(&mut self.buf, vec![]);
let mut prefix = std::mem::replace(&mut self.buf, vec![]); prefix.extend_from_slice(buf.take().as_ref());
prefix.extend_from_slice(buf.take().as_ref()); Box::new(prefix)
Box::new(prefix) } else {
} else { Box::new(buf.take())
Box::new(buf.take()) };
};
let buf1 = buf1.as_ref().as_ref(); let buf1 = buf1.as_ref().as_ref();
match from_utf8(buf1) { match from_utf8(buf1) {
Ok(s) => { Ok(s) => {
if ! s.is_empty() { if !s.is_empty() {
// println!("<< {}", s); // println!("<< {}", s);
let tendril = FromIterator::from_iter(s.chars()); let tendril = FromIterator::from_iter(s.chars());
self.parser.feed(tendril); self.parser.feed(tendril);
} }
}, }
// Remedies for truncated utf8 // Remedies for truncated utf8
Err(e) if e.valid_up_to() >= buf1.len() - 3 => { Err(e) if e.valid_up_to() >= buf1.len() - 3 => {
// Prepare all the valid data // Prepare all the valid data
@ -249,17 +242,16 @@ impl Decoder for XMPPCodec {
self.buf.extend_from_slice(&buf1[e.valid_up_to()..]); self.buf.extend_from_slice(&buf1[e.valid_up_to()..]);
return result; return result;
}, }
Err(e) => { Err(e) => {
// println!("error {} at {}/{} in {:?}", e, e.valid_up_to(), buf1.len(), buf1); // println!("error {} at {}/{} in {:?}", e, e.valid_up_to(), buf1.len(), buf1);
return Err(ParserError::Utf8(e)); return Err(ParserError::Utf8(e));
}, }
} }
match self.queue.borrow_mut().pop_front() { match self.queue.borrow_mut().pop_front() {
None => Ok(None), None => Ok(None),
Some(result) => Some(result) => result.map(|pkt| Some(pkt)),
result.map(|pkt| Some(pkt)),
} }
} }
@ -284,8 +276,7 @@ impl Encoder for XMPPCodec {
let mut buf = String::new(); let mut buf = String::new();
write!(buf, "<stream:stream").unwrap(); write!(buf, "<stream:stream").unwrap();
for (name, value) in start_attrs { for (name, value) in start_attrs {
write!(buf, " {}=\"{}\"", escape(&name), escape(&value)) write!(buf, " {}=\"{}\"", escape(&name), escape(&value)).unwrap();
.unwrap();
if name == "xmlns" { if name == "xmlns" {
self.ns = Some(value); self.ns = Some(value);
} }
@ -293,17 +284,17 @@ impl Encoder for XMPPCodec {
write!(buf, ">\n").unwrap(); write!(buf, ">\n").unwrap();
print!(">> {}", buf); print!(">> {}", buf);
write!(dst, "{}", buf) write!(dst, "{}", buf).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))
.map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e)) }
},
Packet::Stanza(stanza) => { Packet::Stanza(stanza) => {
stanza.write_to_inner(&mut EventWriter::new(WriteBytes::new(dst))) stanza
.write_to_inner(&mut EventWriter::new(WriteBytes::new(dst)))
.and_then(|_| { .and_then(|_| {
// println!(">> {:?}", dst); // println!(">> {:?}", dst);
Ok(()) Ok(())
}) })
.map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("{}", e))) .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("{}", e)))
}, }
Packet::Text(text) => { Packet::Text(text) => {
write_text(&text, dst) write_text(&text, dst)
.and_then(|_| { .and_then(|_| {
@ -311,9 +302,9 @@ impl Encoder for XMPPCodec {
Ok(()) Ok(())
}) })
.map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("{}", e))) .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("{}", e)))
}, }
// TODO: Implement all // TODO: Implement all
_ => Ok(()) _ => Ok(()),
} }
} }
} }
@ -334,7 +325,7 @@ pub fn escape(input: &str) -> String {
'>' => result.push_str("&gt;"), '>' => result.push_str("&gt;"),
'\'' => result.push_str("&apos;"), '\'' => result.push_str("&apos;"),
'"' => result.push_str("&quot;"), '"' => result.push_str("&quot;"),
o => result.push(o) o => result.push(o),
} }
} }
result result
@ -364,7 +355,6 @@ impl<'a> std::io::Write for WriteBytes<'a> {
} }
} }
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
@ -405,10 +395,7 @@ mod tests {
b.put(r">"); b.put(r">");
let r = c.decode(&mut b); let r = c.decode(&mut b);
assert!(match r { assert!(match r {
Ok(Some(Packet::Stanza(ref el))) Ok(Some(Packet::Stanza(ref el))) if el.name() == "test" && el.text() == "ß" => true,
if el.name() == "test"
&& el.text() == "ß"
=> true,
_ => false, _ => false,
}); });
} }
@ -436,10 +423,7 @@ mod tests {
b.put(&b"\x9f</test>"[..]); b.put(&b"\x9f</test>"[..]);
let r = c.decode(&mut b); let r = c.decode(&mut b);
assert!(match r { assert!(match r {
Ok(Some(Packet::Stanza(ref el))) Ok(Some(Packet::Stanza(ref el))) if el.name() == "test" && el.text() == "ß" => true,
if el.name() == "test"
&& el.text() == "ß"
=> true,
_ => false, _ => false,
}); });
} }
@ -447,8 +431,8 @@ mod tests {
/// By default, encode() only get's a BytesMut that has 8kb space reserved. /// By default, encode() only get's a BytesMut that has 8kb space reserved.
#[test] #[test]
fn test_large_stanza() { fn test_large_stanza() {
use std::io::Cursor;
use futures::{Future, Sink}; use futures::{Future, Sink};
use std::io::Cursor;
use tokio_codec::FramedWrite; use tokio_codec::FramedWrite;
let framed = FramedWrite::new(Cursor::new(vec![]), XMPPCodec::new()); let framed = FramedWrite::new(Cursor::new(vec![]), XMPPCodec::new());
let mut text = "".to_owned(); let mut text = "".to_owned();
@ -456,15 +440,12 @@ mod tests {
text = text + "A"; text = text + "A";
} }
let stanza = Element::builder("message") let stanza = Element::builder("message")
.append( .append(Element::builder("body").append(&text).build())
Element::builder("body")
.append(&text)
.build()
)
.build(); .build();
let framed = framed.send(Packet::Stanza(stanza)) let framed = framed.send(Packet::Stanza(stanza)).wait().expect("send");
.wait() assert_eq!(
.expect("send"); framed.get_ref().get_ref(),
assert_eq!(framed.get_ref().get_ref(), &("<message><body>".to_owned() + &text + "</body></message>").as_bytes()); &("<message><body>".to_owned() + &text + "</body></message>").as_bytes()
);
} }
} }

29
src/xmpp_stream.rs
View file

@ -1,14 +1,14 @@
//! `XMPPStream` is the common container for all XMPP network connections //! `XMPPStream` is the common container for all XMPP network connections
use futures::{Poll, Stream, Sink, StartSend};
use futures::sink::Send; use futures::sink::Send;
use tokio_io::{AsyncRead, AsyncWrite}; use futures::{Poll, Sink, StartSend, Stream};
use tokio_codec::Framed;
use minidom::Element;
use jid::Jid; use jid::Jid;
use minidom::Element;
use tokio_codec::Framed;
use tokio_io::{AsyncRead, AsyncWrite};
use xmpp_codec::{XMPPCodec, Packet}; use crate::stream_start::StreamStart;
use stream_start::StreamStart; use crate::xmpp_codec::{Packet, XMPPCodec};
/// <stream:stream> namespace /// <stream:stream> namespace
pub const NS_XMPP_STREAM: &str = "http://etherx.jabber.org/streams"; pub const NS_XMPP_STREAM: &str = "http://etherx.jabber.org/streams";
@ -30,11 +30,18 @@ pub struct XMPPStream<S> {
impl<S: AsyncRead + AsyncWrite> XMPPStream<S> { impl<S: AsyncRead + AsyncWrite> XMPPStream<S> {
/// Constructor /// Constructor
pub fn new(jid: Jid, pub fn new(
stream: Framed<S, XMPPCodec>, jid: Jid,
ns: String, stream: Framed<S, XMPPCodec>,
stream_features: Element) -> Self { ns: String,
XMPPStream { jid, stream, stream_features, ns } stream_features: Element,
) -> Self {
XMPPStream {
jid,
stream,
stream_features,
ns,
}
} }
/// Send a `<stream:stream>` start tag /// Send a `<stream:stream>` start tag