Prepare for merge: Move all tokio-xmpp files into tokio-xmpp/

Signed-off-by: Maxime “pep” Buquet <pep@bouah.net>

tokio-xmpp/src/client/auth.rs

@@ -0,0 +1,116 @@
+use std::str::FromStr;
+use std::collections::HashSet;
+use std::convert::TryFrom;
+use futures::{Future, Poll, Stream, future::{ok, err, IntoFuture}};
+use sasl::client::mechanisms::{Anonymous, Plain, Scram};
+use sasl::client::Mechanism;
+use sasl::common::scram::{Sha1, Sha256};
+use sasl::common::Credentials;
+use tokio_io::{AsyncRead, AsyncWrite};
+use xmpp_parsers::sasl::{Auth, Challenge, Failure, Mechanism as XMPPMechanism, Response, Success};
+
+use crate::xmpp_codec::Packet;
+use crate::xmpp_stream::XMPPStream;
+use crate::{AuthError, Error, ProtocolError};
+
+const NS_XMPP_SASL: &str = "urn:ietf:params:xml:ns:xmpp-sasl";
+
+pub struct ClientAuth<S: AsyncRead + AsyncWrite> {
+    future: Box<dyn Future<Item = XMPPStream<S>, Error = Error>>,
+}
+
+impl<S: AsyncRead + AsyncWrite + 'static> ClientAuth<S> {
+    pub fn new(stream: XMPPStream<S>, creds: Credentials) -> Result<Self, Error> {
+        let local_mechs: Vec<Box<dyn Fn() -> Box<dyn Mechanism>>> = vec![
+            Box::new(|| Box::new(Scram::<Sha256>::from_credentials(creds.clone()).unwrap())),
+            Box::new(|| Box::new(Scram::<Sha1>::from_credentials(creds.clone()).unwrap())),
+            Box::new(|| Box::new(Plain::from_credentials(creds.clone()).unwrap())),
+            Box::new(|| Box::new(Anonymous::new())),
+        ];
+
+        let remote_mechs: HashSet<String> = stream
+            .stream_features
+            .get_child("mechanisms", NS_XMPP_SASL)
+            .ok_or(AuthError::NoMechanism)?
+            .children()
+            .filter(|child| child.is("mechanism", NS_XMPP_SASL))
+            .map(|mech_el| mech_el.text())
+            .collect();
+
+        for local_mech in local_mechs {
+            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 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,
+                });
+            }
+        }
+
+        Err(AuthError::NoMechanism)?
+    }
+
+    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 {
+                            // ignore and loop
+                            Self::handle_challenge(stream, mechanism)
+                        }
+                    }
+                    Some(_) => {
+                        // ignore and loop
+                        Self::handle_challenge(stream, mechanism)
+                    }
+                    None => Box::new(err(Error::Disconnected))
+                }
+            })
+        )
+    }
+}
+
+impl<S: AsyncRead + AsyncWrite> Future for ClientAuth<S> {
+    type Item = XMPPStream<S>;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        self.future.poll()
+    }
+}

tokio-xmpp/src/client/bind.rs

@@ -0,0 +1,102 @@
+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 crate::xmpp_codec::Packet;
+use crate::xmpp_stream::XMPPStream;
+use crate::{Error, ProtocolError};
+
+const NS_XMPP_BIND: &str = "urn:ietf:params:xml:ns:xmpp-bind";
+const BIND_REQ_ID: &str = "resource-bind";
+
+pub enum ClientBind<S: AsyncWrite> {
+    Unsupported(XMPPStream<S>),
+    WaitSend(sink::Send<XMPPStream<S>>),
+    WaitRecv(XMPPStream<S>),
+    Invalid,
+}
+
+impl<S: AsyncWrite> ClientBind<S> {
+    /// Consumes and returns the stream to express that you cannot use
+    /// the stream for anything else until the resource binding
+    /// req/resp are done.
+    pub fn new(stream: XMPPStream<S>) -> Self {
+        match stream.stream_features.get_child("bind", NS_XMPP_BIND) {
+            None =>
+            // No resource binding available,
+            // return the (probably // usable) stream immediately
+            {
+                ClientBind::Unsupported(stream)
+            }
+            Some(_) => {
+                let resource;
+                if let Jid::Full(jid) = stream.jid.clone() {
+                    resource = Some(jid.resource);
+                } else {
+                    resource = None;
+                }
+                let iq = Iq::from_set(BIND_REQ_ID, BindQuery::new(resource));
+                let send = stream.send_stanza(iq);
+                ClientBind::WaitSend(send)
+            }
+        }
+    }
+}
+
+impl<S: AsyncRead + AsyncWrite> Future for ClientBind<S> {
+    type Item = XMPPStream<S>;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        let state = replace(self, ClientBind::Invalid);
+
+        match state {
+            ClientBind::Unsupported(stream) => Ok(Async::Ready(stream)),
+            ClientBind::WaitSend(mut send) => match send.poll() {
+                Ok(Async::Ready(stream)) => {
+                    replace(self, ClientBind::WaitRecv(stream));
+                    self.poll()
+                }
+                Ok(Async::NotReady) => {
+                    replace(self, ClientBind::WaitSend(send));
+                    Ok(Async::NotReady)
+                }
+                Err(e) => Err(e)?,
+            },
+            ClientBind::WaitRecv(mut stream) => match stream.poll() {
+                Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => match Iq::try_from(stanza) {
+                    Ok(iq) => {
+                        if iq.id == BIND_REQ_ID {
+                            match iq.payload {
+                                IqType::Result(payload) => {
+                                    payload
+                                        .and_then(|payload| BindResponse::try_from(payload).ok())
+                                        .map(|bind| stream.jid = bind.into());
+                                    Ok(Async::Ready(stream))
+                                }
+                                _ => Err(ProtocolError::InvalidBindResponse)?,
+                            }
+                        } else {
+                            Ok(Async::NotReady)
+                        }
+                    }
+                    _ => Ok(Async::NotReady),
+                },
+                Ok(Async::Ready(_)) => {
+                    replace(self, ClientBind::WaitRecv(stream));
+                    self.poll()
+                }
+                Ok(Async::NotReady) => {
+                    replace(self, ClientBind::WaitRecv(stream));
+                    Ok(Async::NotReady)
+                }
+                Err(e) => Err(e)?,
+            },
+            ClientBind::Invalid => unreachable!(),
+        }
+    }
+}

tokio-xmpp/src/client/mod.rs

@@ -0,0 +1,236 @@
+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 super::event::Event;
+use super::happy_eyeballs::Connecter;
+use super::starttls::{StartTlsClient, NS_XMPP_TLS};
+use super::xmpp_codec::Packet;
+use super::xmpp_stream;
+use super::{Error, ProtocolError};
+
+mod auth;
+use self::auth::ClientAuth;
+mod bind;
+use self::bind::ClientBind;
+
+/// XMPP client connection and state
+pub struct Client {
+    state: ClientState,
+}
+
+type XMPPStream = xmpp_stream::XMPPStream<TlsStream<TcpStream>>;
+const NS_JABBER_CLIENT: &str = "jabber:client";
+
+enum ClientState {
+    Invalid,
+    Disconnected,
+    Connecting(Box<dyn Future<Item = XMPPStream, Error = Error>>),
+    Connected(XMPPStream),
+}
+
+impl Client {
+    /// Start a new XMPP client
+    ///
+    /// Start polling the returned instance so that it will connect
+    /// and yield events.
+    pub fn new(jid: &str, password: &str) -> Result<Self, JidParseError> {
+        let jid = Jid::from_str(jid)?;
+        let client = Self::new_with_jid(jid, password);
+        Ok(client)
+    }
+
+    /// Start a new client given that the JID is already parsed.
+    pub fn new_with_jid(jid: Jid, password: &str) -> Self {
+        let password = password.to_owned();
+        let connect = Self::make_connect(jid, password.clone());
+        let client = Client {
+            state: ClientState::Connecting(Box::new(connect)),
+        };
+        client
+    }
+
+    fn make_connect(jid: Jid, password: String) -> impl Future<Item = XMPPStream, Error = Error> {
+        let username = jid.clone().node().unwrap();
+        let jid1 = jid.clone();
+        let jid2 = jid.clone();
+        let password = password;
+        done(idna::domain_to_ascii(&jid.domain()))
+            .map_err(|_| Error::Idna)
+            .and_then(|domain| {
+                done(Connecter::from_lookup(
+                    &domain,
+                    Some("_xmpp-client._tcp"),
+                    5222,
+                ))
+            })
+            .flatten()
+            .and_then(move |tcp_stream| {
+                xmpp_stream::XMPPStream::start(tcp_stream, jid1, NS_JABBER_CLIENT.to_owned())
+            })
+            .and_then(|xmpp_stream| {
+                if Self::can_starttls(&xmpp_stream) {
+                    Ok(Self::starttls(xmpp_stream))
+                } else {
+                    Err(Error::Protocol(ProtocolError::NoTls))
+                }
+            })
+            .flatten()
+            .and_then(|tls_stream| XMPPStream::start(tls_stream, jid2, NS_JABBER_CLIENT.to_owned()))
+            .and_then(
+                move |xmpp_stream| done(Self::auth(xmpp_stream, username, password)), // TODO: flatten?
+            )
+            .and_then(|auth| auth)
+            .and_then(|xmpp_stream| Self::bind(xmpp_stream))
+            .and_then(|xmpp_stream| {
+                // println!("Bound to {}", xmpp_stream.jid);
+                Ok(xmpp_stream)
+            })
+    }
+
+    fn can_starttls<S>(stream: &xmpp_stream::XMPPStream<S>) -> bool {
+        stream
+            .stream_features
+            .get_child("starttls", NS_XMPP_TLS)
+            .is_some()
+    }
+
+    fn starttls<S: AsyncRead + AsyncWrite>(
+        stream: xmpp_stream::XMPPStream<S>,
+    ) -> StartTlsClient<S> {
+        StartTlsClient::from_stream(stream)
+    }
+
+    fn auth<S: AsyncRead + AsyncWrite + 'static>(
+        stream: xmpp_stream::XMPPStream<S>,
+        username: String,
+        password: String,
+    ) -> Result<ClientAuth<S>, Error> {
+        let creds = Credentials::default()
+            .with_username(username)
+            .with_password(password)
+            .with_channel_binding(ChannelBinding::None);
+        ClientAuth::new(stream, creds)
+    }
+
+    fn bind<S: AsyncWrite>(stream: xmpp_stream::XMPPStream<S>) -> ClientBind<S> {
+        ClientBind::new(stream)
+    }
+
+    /// Get the client's bound JID (the one reported by the XMPP
+    /// server).
+    pub fn bound_jid(&self) -> Option<&Jid> {
+        match self.state {
+            ClientState::Connected(ref stream) => Some(&stream.jid),
+            _ => None,
+        }
+    }
+}
+
+impl Stream for Client {
+    type Item = Event;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        let state = replace(&mut self.state, ClientState::Invalid);
+
+        match state {
+            ClientState::Invalid => Err(Error::InvalidState),
+            ClientState::Disconnected => Ok(Async::Ready(None)),
+            ClientState::Connecting(mut connect) => match connect.poll() {
+                Ok(Async::Ready(stream)) => {
+                    let jid = stream.jid.clone();
+                    self.state = ClientState::Connected(stream);
+                    Ok(Async::Ready(Some(Event::Online(jid))))
+                }
+                Ok(Async::NotReady) => {
+                    self.state = ClientState::Connecting(connect);
+                    Ok(Async::NotReady)
+                }
+                Err(e) => Err(e),
+            },
+            ClientState::Connected(mut stream) => {
+                // Poll sink
+                match stream.poll_complete() {
+                    Ok(Async::NotReady) => (),
+                    Ok(Async::Ready(())) => (),
+                    Err(e) => return Err(e)?,
+                };
+
+                // Poll stream
+                match stream.poll() {
+                    Ok(Async::Ready(None)) => {
+                        // EOF
+                        self.state = ClientState::Disconnected;
+                        Ok(Async::Ready(Some(Event::Disconnected)))
+                    }
+                    Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
+                        // Receive stanza
+                        self.state = ClientState::Connected(stream);
+                        Ok(Async::Ready(Some(Event::Stanza(stanza))))
+                    }
+                    Ok(Async::Ready(Some(Packet::Text(_)))) => {
+                        // Ignore text between stanzas
+                        Ok(Async::NotReady)
+                    }
+                    Ok(Async::Ready(Some(Packet::StreamStart(_)))) => {
+                        // <stream:stream>
+                        Err(ProtocolError::InvalidStreamStart.into())
+                    }
+                    Ok(Async::Ready(Some(Packet::StreamEnd))) => {
+                        // End of stream: </stream:stream>
+                        Ok(Async::Ready(None))
+                    }
+                    Ok(Async::NotReady) => {
+                        // Try again later
+                        self.state = ClientState::Connected(stream);
+                        Ok(Async::NotReady)
+                    }
+                    Err(e) => Err(e)?,
+                }
+            }
+        }
+    }
+}
+
+impl Sink for Client {
+    type SinkItem = Packet;
+    type SinkError = Error;
+
+    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)),
+        }
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        match self.state {
+            ClientState::Connected(ref mut stream) => stream.poll_complete().map_err(|e| e.into()),
+            _ => Ok(Async::Ready(())),
+        }
+    }
+
+    /// This closes the inner TCP stream.
+    ///
+    /// To synchronize your shutdown with the server side, you should
+    /// first send `Packet::StreamEnd` and wait for the end of the
+    /// 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(())),
+        }
+    }
+}

tokio-xmpp/src/component/auth.rs

@@ -0,0 +1,89 @@
+use futures::{sink, Async, Future, Poll, Stream};
+use std::mem::replace;
+use tokio_io::{AsyncRead, AsyncWrite};
+use xmpp_parsers::component::Handshake;
+
+use crate::xmpp_codec::Packet;
+use crate::xmpp_stream::XMPPStream;
+use crate::{AuthError, Error};
+
+const NS_JABBER_COMPONENT_ACCEPT: &str = "jabber:component:accept";
+
+pub struct ComponentAuth<S: AsyncWrite> {
+    state: ComponentAuthState<S>,
+}
+
+enum ComponentAuthState<S: AsyncWrite> {
+    WaitSend(sink::Send<XMPPStream<S>>),
+    WaitRecv(XMPPStream<S>),
+    Invalid,
+}
+
+impl<S: AsyncWrite> ComponentAuth<S> {
+    // TODO: doesn't have to be a Result<> actually
+    pub fn new(stream: XMPPStream<S>, password: String) -> Result<Self, Error> {
+        // FIXME: huge hack, shouldn’t be an element!
+        let sid = stream.stream_features.name().to_owned();
+        let mut this = ComponentAuth {
+            state: ComponentAuthState::Invalid,
+        };
+        this.send(
+            stream,
+            Handshake::from_password_and_stream_id(&password, &sid),
+        );
+        Ok(this)
+    }
+
+    fn send(&mut self, stream: XMPPStream<S>, handshake: Handshake) {
+        let nonza = handshake;
+        let send = stream.send_stanza(nonza);
+
+        self.state = ComponentAuthState::WaitSend(send);
+    }
+}
+
+impl<S: AsyncRead + AsyncWrite> Future for ComponentAuth<S> {
+    type Item = XMPPStream<S>;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        let state = replace(&mut self.state, ComponentAuthState::Invalid);
+
+        match state {
+            ComponentAuthState::WaitSend(mut send) => match send.poll() {
+                Ok(Async::Ready(stream)) => {
+                    self.state = ComponentAuthState::WaitRecv(stream);
+                    self.poll()
+                }
+                Ok(Async::NotReady) => {
+                    self.state = ComponentAuthState::WaitSend(send);
+                    Ok(Async::NotReady)
+                }
+                Err(e) => Err(e)?,
+            },
+            ComponentAuthState::WaitRecv(mut stream) => match stream.poll() {
+                Ok(Async::Ready(Some(Packet::Stanza(ref stanza))))
+                    if stanza.is("handshake", NS_JABBER_COMPONENT_ACCEPT) =>
+                {
+                    self.state = ComponentAuthState::Invalid;
+                    Ok(Async::Ready(stream))
+                }
+                Ok(Async::Ready(Some(Packet::Stanza(ref stanza))))
+                    if stanza.is("error", "http://etherx.jabber.org/streams") =>
+                {
+                    Err(AuthError::ComponentFail.into())
+                }
+                Ok(Async::Ready(_event)) => {
+                    // println!("ComponentAuth ignore {:?}", _event);
+                    Ok(Async::NotReady)
+                }
+                Ok(_) => {
+                    self.state = ComponentAuthState::WaitRecv(stream);
+                    Ok(Async::NotReady)
+                }
+                Err(e) => Err(e)?,
+            },
+            ComponentAuthState::Invalid => unreachable!(),
+        }
+    }
+}

tokio-xmpp/src/component/mod.rs

@@ -0,0 +1,163 @@
+//! Components in XMPP are services/gateways that are logged into an
+//! 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 super::event::Event;
+use super::happy_eyeballs::Connecter;
+use super::xmpp_codec::Packet;
+use super::xmpp_stream;
+use super::Error;
+
+mod auth;
+use self::auth::ComponentAuth;
+
+/// Component connection to an XMPP server
+pub struct Component {
+    /// The component's Jabber-Id
+    pub jid: Jid,
+    state: ComponentState,
+}
+
+type XMPPStream = xmpp_stream::XMPPStream<TcpStream>;
+const NS_JABBER_COMPONENT_ACCEPT: &str = "jabber:component:accept";
+
+enum ComponentState {
+    Invalid,
+    Disconnected,
+    Connecting(Box<dyn Future<Item = XMPPStream, Error = Error>>),
+    Connected(XMPPStream),
+}
+
+impl Component {
+    /// Start a new XMPP component
+    ///
+    /// Start polling the returned instance so that it will connect
+    /// and yield events.
+    pub fn new(jid: &str, password: &str, server: &str, port: u16) -> Result<Self, JidParseError> {
+        let jid = Jid::from_str(jid)?;
+        let password = password.to_owned();
+        let connect = Self::make_connect(jid.clone(), password, server, port);
+        Ok(Component {
+            jid,
+            state: ComponentState::Connecting(Box::new(connect)),
+        })
+    }
+
+    fn make_connect(
+        jid: Jid,
+        password: String,
+        server: &str,
+        port: u16,
+    ) -> impl Future<Item = XMPPStream, Error = Error> {
+        let jid1 = jid.clone();
+        let password = password;
+        done(Connecter::from_lookup(server, None, port))
+            .flatten()
+            .and_then(move |tcp_stream| {
+                xmpp_stream::XMPPStream::start(
+                    tcp_stream,
+                    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> {
+        ComponentAuth::new(stream, password)
+    }
+}
+
+impl Stream for Component {
+    type Item = Event;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        let state = replace(&mut self.state, ComponentState::Invalid);
+
+        match state {
+            ComponentState::Invalid => Err(Error::InvalidState),
+            ComponentState::Disconnected => Ok(Async::Ready(None)),
+            ComponentState::Connecting(mut connect) => match connect.poll() {
+                Ok(Async::Ready(stream)) => {
+                    self.state = ComponentState::Connected(stream);
+                    Ok(Async::Ready(Some(Event::Online(self.jid.clone()))))
+                }
+                Ok(Async::NotReady) => {
+                    self.state = ComponentState::Connecting(connect);
+                    Ok(Async::NotReady)
+                }
+                Err(e) => Err(e),
+            },
+            ComponentState::Connected(mut stream) => {
+                // Poll sink
+                match stream.poll_complete() {
+                    Ok(Async::NotReady) => (),
+                    Ok(Async::Ready(())) => (),
+                    Err(e) => return Err(e)?,
+                };
+
+                // Poll stream
+                match stream.poll() {
+                    Ok(Async::NotReady) => {
+                        self.state = ComponentState::Connected(stream);
+                        Ok(Async::NotReady)
+                    }
+                    Ok(Async::Ready(None)) => {
+                        // EOF
+                        self.state = ComponentState::Disconnected;
+                        Ok(Async::Ready(Some(Event::Disconnected)))
+                    }
+                    Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
+                        self.state = ComponentState::Connected(stream);
+                        Ok(Async::Ready(Some(Event::Stanza(stanza))))
+                    }
+                    Ok(Async::Ready(_)) => {
+                        self.state = ComponentState::Connected(stream);
+                        Ok(Async::NotReady)
+                    }
+                    Err(e) => Err(e)?,
+                }
+            }
+        }
+    }
+}
+
+impl Sink for Component {
+    type SinkItem = Element;
+    type SinkError = Error;
+
+    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
+        match self.state {
+            ComponentState::Connected(ref mut stream) => match stream
+                .start_send(Packet::Stanza(item))
+            {
+                Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) => Ok(AsyncSink::NotReady(stanza)),
+                Ok(AsyncSink::NotReady(_)) => {
+                    panic!("Component.start_send with stanza but got something else back")
+                }
+                Ok(AsyncSink::Ready) => Ok(AsyncSink::Ready),
+                Err(e) => Err(e)?,
+            },
+            _ => Ok(AsyncSink::NotReady(item)),
+        }
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        match &mut self.state {
+            &mut ComponentState::Connected(ref mut stream) => {
+                stream.poll_complete().map_err(|e| e.into())
+            }
+            _ => Ok(Async::Ready(())),
+        }
+    }
+}

tokio-xmpp/src/error.rs

@@ -0,0 +1,224 @@
+use native_tls::Error as TlsError;
+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_resolver::error::ResolveError;
+
+use xmpp_parsers::Error as ParsersError;
+use xmpp_parsers::sasl::DefinedCondition as SaslDefinedCondition;
+
+/// Top-level error type
+#[derive(Debug)]
+pub enum Error {
+    /// I/O error
+    Io(IoError),
+    /// Error resolving DNS and establishing a connection
+    Connection(ConnecterError),
+    /// DNS label conversion error, no details available from module
+    /// `idna`
+    Idna,
+    /// Protocol-level error
+    Protocol(ProtocolError),
+    /// Authentication error
+    Auth(AuthError),
+    /// TLS error
+    Tls(TlsError),
+    /// Connection closed
+    Disconnected,
+    /// Shoud never happen
+    InvalidState,
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            Error::Io(e) => write!(fmt, "IO error: {}", e),
+            Error::Connection(e) => write!(fmt, "connection error: {}", e),
+            Error::Idna => write!(fmt, "IDNA error"),
+            Error::Protocol(e) => write!(fmt, "protocol error: {}", e),
+            Error::Auth(e) => write!(fmt, "authentication error: {}", e),
+            Error::Tls(e) => write!(fmt, "TLS error: {}", e),
+            Error::Disconnected => write!(fmt, "disconnected"),
+            Error::InvalidState => write!(fmt, "invalid state"),
+        }
+    }
+}
+
+impl From<IoError> for Error {
+    fn from(e: IoError) -> Self {
+        Error::Io(e)
+    }
+}
+
+impl From<ConnecterError> for Error {
+    fn from(e: ConnecterError) -> Self {
+        Error::Connection(e)
+    }
+}
+
+impl From<ProtocolError> for Error {
+    fn from(e: ProtocolError) -> Self {
+        Error::Protocol(e)
+    }
+}
+
+impl From<AuthError> for Error {
+    fn from(e: AuthError) -> Self {
+        Error::Auth(e)
+    }
+}
+
+impl From<TlsError> for Error {
+    fn from(e: TlsError) -> Self {
+        Error::Tls(e)
+    }
+}
+
+/// Causes for stream parsing errors
+#[derive(Debug)]
+pub enum ParserError {
+    /// Encoding error
+    Utf8(Utf8Error),
+    /// XML parse error
+    Parse(ParseError),
+    /// Illegal `</>`
+    ShortTag,
+    /// Required by `impl Decoder`
+    Io(IoError),
+}
+
+impl fmt::Display for ParserError {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            ParserError::Utf8(e) => write!(fmt, "UTF-8 error: {}", e),
+            ParserError::Parse(e) => write!(fmt, "parse error: {}", e),
+            ParserError::ShortTag => write!(fmt, "short tag"),
+            ParserError::Io(e) => write!(fmt, "IO error: {}", e),
+        }
+    }
+}
+
+impl From<IoError> for ParserError {
+    fn from(e: IoError) -> Self {
+        ParserError::Io(e)
+    }
+}
+
+impl From<ParserError> for Error {
+    fn from(e: ParserError) -> Self {
+        ProtocolError::Parser(e).into()
+    }
+}
+
+/// XML parse error wrapper type
+#[derive(Debug)]
+pub struct ParseError(pub Cow<'static, str>);
+
+impl StdError for ParseError {
+    fn description(&self) -> &str {
+        self.0.as_ref()
+    }
+    fn cause(&self) -> Option<&dyn StdError> {
+        None
+    }
+}
+
+impl fmt::Display for ParseError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+/// XMPP protocol-level error
+#[derive(Debug)]
+pub enum ProtocolError {
+    /// XML parser error
+    Parser(ParserError),
+    /// Error with expected stanza schema
+    Parsers(ParsersError),
+    /// No TLS available
+    NoTls,
+    /// Invalid response to resource binding
+    InvalidBindResponse,
+    /// No xmlns attribute in <stream:stream>
+    NoStreamNamespace,
+    /// No id attribute in <stream:stream>
+    NoStreamId,
+    /// Encountered an unexpected XML token
+    InvalidToken,
+    /// Unexpected <stream:stream> (shouldn't occur)
+    InvalidStreamStart,
+}
+
+impl fmt::Display for ProtocolError {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            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::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>"),
+        }
+    }
+}
+
+impl From<ParserError> for ProtocolError {
+    fn from(e: ParserError) -> Self {
+        ProtocolError::Parser(e)
+    }
+}
+
+impl From<ParsersError> for ProtocolError {
+    fn from(e: ParsersError) -> Self {
+        ProtocolError::Parsers(e)
+    }
+}
+
+/// Authentication error
+#[derive(Debug)]
+pub enum AuthError {
+    /// No matching SASL mechanism available
+    NoMechanism,
+    /// Local SASL implementation error
+    Sasl(String),
+    /// Failure from server
+    Fail(SaslDefinedCondition),
+    /// Component authentication failure
+    ComponentFail,
+}
+
+impl fmt::Display for AuthError {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            AuthError::NoMechanism => write!(fmt, "no matching SASL mechanism available"),
+            AuthError::Sasl(s) => write!(fmt, "local SASL implementation error: {}", s),
+            AuthError::Fail(c) => write!(fmt, "failure from the server: {:?}", c),
+            AuthError::ComponentFail => write!(fmt, "component authentication failure"),
+        }
+    }
+}
+
+/// Error establishing connection
+#[derive(Debug)]
+pub enum ConnecterError {
+    /// All attempts failed, no error available
+    AllFailed,
+    /// DNS protocol error
+    Dns(ProtoError),
+    /// DNS resolution error
+    Resolve(ResolveError),
+}
+
+impl std::error::Error for ConnecterError {}
+
+impl std::fmt::Display for ConnecterError {
+    fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> {
+        write!(fmt, "{:?}", self)
+    }
+}

tokio-xmpp/src/event.rs

@@ -0,0 +1,54 @@
+use xmpp_parsers::{Element, Jid};
+
+/// High-level event on the Stream implemented by Client and Component
+#[derive(Debug)]
+pub enum Event {
+    /// Stream is connected and initialized
+    Online(Jid),
+    /// Stream end
+    Disconnected,
+    /// Received stanza/nonza
+    Stanza(Element),
+}
+
+impl Event {
+    /// `Online` event?
+    pub fn is_online(&self) -> bool {
+        match *self {
+            Event::Online(_) => true,
+            _ => false,
+        }
+    }
+
+    /// Get the server-assigned JID for the `Online` event
+    pub fn get_jid(&self) -> Option<&Jid> {
+        match *self {
+            Event::Online(ref jid) => Some(jid),
+            _ => None,
+        }
+    }
+
+    /// `Stanza` event?
+    pub fn is_stanza(&self, name: &str) -> bool {
+        match *self {
+            Event::Stanza(ref stanza) => stanza.name() == name,
+            _ => false,
+        }
+    }
+
+    /// If this is a `Stanza` event, get its data
+    pub fn as_stanza(&self) -> Option<&Element> {
+        match *self {
+            Event::Stanza(ref stanza) => Some(stanza),
+            _ => None,
+        }
+    }
+
+    /// If this is a `Stanza` event, unwrap into its data
+    pub fn into_stanza(self) -> Option<Element> {
+        match self {
+            Event::Stanza(stanza) => Some(stanza),
+            _ => None,
+        }
+    }
+}

tokio-xmpp/src/happy_eyeballs.rs

@@ -0,0 +1,196 @@
+use crate::{ConnecterError, Error};
+use futures::{Async, Future, Poll};
+use std::cell::RefCell;
+use std::collections::BTreeMap;
+use std::collections::VecDeque;
+use std::io::Error as IoError;
+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;
+
+
+enum State {
+    ResolveSrv(AsyncResolver, BackgroundLookup<SrvLookupFuture>),
+    ResolveTarget(AsyncResolver, Background<LookupIpFuture>, u16),
+    Connecting(Option<AsyncResolver>, Vec<RefCell<ConnectFuture>>),
+    Invalid,
+}
+
+pub struct Connecter {
+    fallback_port: u16,
+    srv_domain: Option<Name>,
+    domain: Name,
+    state: State,
+    targets: VecDeque<(Name, u16)>,
+    error: Option<Error>,
+}
+
+fn resolver() -> Result<AsyncResolver, IoError> {
+    let (config, mut opts) = trust_dns_resolver::system_conf::read_system_conf()?;
+    opts.ip_strategy = LookupIpStrategy::Ipv4AndIpv6;
+    let (resolver, resolver_background) = AsyncResolver::new(config, opts);
+    tokio::runtime::current_thread::spawn(resolver_background);
+    Ok(resolver)
+}
+
+impl Connecter {
+    pub fn from_lookup(
+        domain: &str,
+        srv: Option<&str>,
+        fallback_port: u16,
+    ) -> Result<Connecter, Error> {
+        if let Ok(ip) = domain.parse() {
+            // use specified IP address, not domain name, skip the whole dns part
+            let connect = RefCell::new(TcpStream::connect(&SocketAddr::new(ip, fallback_port)));
+            return Ok(Connecter {
+                fallback_port,
+                srv_domain: None,
+                domain: "nohost".into_name().map_err(ConnecterError::Dns)?,
+                state: State::Connecting(None, vec![connect]),
+                targets: VecDeque::new(),
+                error: None,
+            });
+        }
+
+        let srv_domain = match srv {
+            Some(srv) => Some(
+                format!("{}.{}.", srv, domain)
+                    .into_name()
+                    .map_err(ConnecterError::Dns)?,
+            ),
+            None => None,
+        };
+
+        let mut self_ = Connecter {
+            fallback_port,
+            srv_domain,
+            domain: domain.into_name().map_err(ConnecterError::Dns)?,
+            state: State::Invalid,
+            targets: VecDeque::new(),
+            error: None,
+        };
+
+        let resolver = resolver()?;
+        // Initialize state
+        match &self_.srv_domain {
+            &Some(ref srv_domain) => {
+                let srv_lookup = resolver.lookup_srv(srv_domain.clone());
+                self_.state = State::ResolveSrv(resolver, srv_lookup);
+            }
+            None => {
+                self_.targets = [(self_.domain.clone(), self_.fallback_port)]
+                    .into_iter()
+                    .cloned()
+                    .collect();
+                self_.state = State::Connecting(Some(resolver), vec![]);
+            }
+        }
+
+        Ok(self_)
+    }
+}
+
+impl Future for Connecter {
+    type Item = TcpStream;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        let state = mem::replace(&mut self.state, State::Invalid);
+        match state {
+            State::ResolveSrv(resolver, mut srv_lookup) => {
+                match srv_lookup.poll() {
+                    Ok(Async::NotReady) => {
+                        self.state = State::ResolveSrv(resolver, srv_lookup);
+                        Ok(Async::NotReady)
+                    }
+                    Ok(Async::Ready(srv_result)) => {
+                        let srv_map: BTreeMap<_, _> = srv_result
+                            .iter()
+                            .map(|srv| (srv.priority(), (srv.target().clone(), srv.port())))
+                            .collect();
+                        let targets = srv_map.into_iter().map(|(_, tp)| tp).collect();
+                        self.targets = targets;
+                        self.state = State::Connecting(Some(resolver), vec![]);
+                        self.poll()
+                    }
+                    Err(_) => {
+                        // ignore, fallback
+                        self.targets = [(self.domain.clone(), self.fallback_port)]
+                            .into_iter()
+                            .cloned()
+                            .collect();
+                        self.state = State::Connecting(Some(resolver), vec![]);
+                        self.poll()
+                    }
+                }
+            }
+            State::Connecting(resolver, mut connects) => {
+                if resolver.is_some() && connects.len() == 0 && self.targets.len() > 0 {
+                    let resolver = resolver.unwrap();
+                    let (host, port) = self.targets.pop_front().unwrap();
+                    let ip_lookup = resolver.lookup_ip(host);
+                    self.state = State::ResolveTarget(resolver, ip_lookup, port);
+                    self.poll()
+                } else if connects.len() > 0 {
+                    let mut success = None;
+                    connects.retain(|connect| match connect.borrow_mut().poll() {
+                        Ok(Async::NotReady) => true,
+                        Ok(Async::Ready(connection)) => {
+                            success = Some(connection);
+                            false
+                        }
+                        Err(e) => {
+                            if self.error.is_none() {
+                                self.error = Some(e.into());
+                            }
+                            false
+                        }
+                    });
+                    match success {
+                        Some(connection) => Ok(Async::Ready(connection)),
+                        None => {
+                            self.state = State::Connecting(resolver, connects);
+                            Ok(Async::NotReady)
+                        }
+                    }
+                } else {
+                    // All targets tried
+                    match self.error.take() {
+                        None => Err(ConnecterError::AllFailed.into()),
+                        Some(e) => Err(e),
+                    }
+                }
+            }
+            State::ResolveTarget(resolver, mut ip_lookup, port) => {
+                match ip_lookup.poll() {
+                    Ok(Async::NotReady) => {
+                        self.state = State::ResolveTarget(resolver, ip_lookup, port);
+                        Ok(Async::NotReady)
+                    }
+                    Ok(Async::Ready(ip_result)) => {
+                        let connects = ip_result
+                            .iter()
+                            .map(|ip| RefCell::new(TcpStream::connect(&SocketAddr::new(ip, port))))
+                            .collect();
+                        self.state = State::Connecting(Some(resolver), connects);
+                        self.poll()
+                    }
+                    Err(e) => {
+                        if self.error.is_none() {
+                            self.error = Some(ConnecterError::Resolve(e).into());
+                        }
+                        // ignore, next…
+                        self.state = State::Connecting(Some(resolver), vec![]);
+                        self.poll()
+                    }
+                }
+            }
+            _ => panic!(""),
+        }
+    }
+}

tokio-xmpp/src/lib.rs

@@ -0,0 +1,19 @@
+#![deny(unsafe_code, unused, missing_docs, bare_trait_objects)]
+
+//! XMPP implementation with asynchronous I/O using Tokio.
+
+mod starttls;
+mod stream_start;
+pub mod xmpp_codec;
+pub use crate::xmpp_codec::Packet;
+pub mod xmpp_stream;
+pub use crate::starttls::StartTlsClient;
+mod event;
+mod happy_eyeballs;
+pub use crate::event::Event;
+mod client;
+pub use crate::client::Client;
+mod component;
+pub use crate::component::Component;
+mod error;
+pub use crate::error::{AuthError, ConnecterError, Error, ParseError, ParserError, ProtocolError};

tokio-xmpp/src/starttls.rs

@@ -0,0 +1,114 @@
+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 crate::xmpp_codec::Packet;
+use crate::xmpp_stream::XMPPStream;
+use crate::Error;
+
+/// XMPP TLS XML namespace
+pub const NS_XMPP_TLS: &str = "urn:ietf:params:xml:ns:xmpp-tls";
+
+/// XMPP stream that switches to TLS if available in received features
+pub struct StartTlsClient<S: AsyncRead + AsyncWrite> {
+    state: StartTlsClientState<S>,
+    jid: Jid,
+}
+
+enum StartTlsClientState<S: AsyncRead + AsyncWrite> {
+    Invalid,
+    SendStartTls(sink::Send<XMPPStream<S>>),
+    AwaitProceed(XMPPStream<S>),
+    StartingTls(Connect<S>),
+}
+
+impl<S: AsyncRead + AsyncWrite> StartTlsClient<S> {
+    /// Waits for <stream:features>
+    pub fn from_stream(xmpp_stream: XMPPStream<S>) -> Self {
+        let jid = xmpp_stream.jid.clone();
+
+        let nonza = Element::builder("starttls").ns(NS_XMPP_TLS).build();
+        let packet = Packet::Stanza(nonza);
+        let send = xmpp_stream.send(packet);
+
+        StartTlsClient {
+            state: StartTlsClientState::SendStartTls(send),
+            jid,
+        }
+    }
+}
+
+impl<S: AsyncRead + AsyncWrite> Future for StartTlsClient<S> {
+    type Item = TlsStream<S>;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        let old_state = replace(&mut self.state, StartTlsClientState::Invalid);
+        let mut retry = false;
+
+        let (new_state, result) = match old_state {
+            StartTlsClientState::SendStartTls(mut send) => match send.poll() {
+                Ok(Async::Ready(xmpp_stream)) => {
+                    let new_state = StartTlsClientState::AwaitProceed(xmpp_stream);
+                    retry = true;
+                    (new_state, Ok(Async::NotReady))
+                }
+                Ok(Async::NotReady) => {
+                    (StartTlsClientState::SendStartTls(send), Ok(Async::NotReady))
+                }
+                Err(e) => (StartTlsClientState::SendStartTls(send), Err(e.into())),
+            },
+            StartTlsClientState::AwaitProceed(mut xmpp_stream) => match xmpp_stream.poll() {
+                Ok(Async::Ready(Some(Packet::Stanza(ref stanza))))
+                    if stanza.name() == "proceed" =>
+                {
+                    let stream = xmpp_stream.stream.into_inner();
+                    let connect =
+                        TlsConnector::from(NativeTlsConnector::builder().build().unwrap())
+                            .connect(&self.jid.clone().domain(), stream);
+                    let new_state = StartTlsClientState::StartingTls(connect);
+                    retry = true;
+                    (new_state, Ok(Async::NotReady))
+                }
+                Ok(Async::Ready(_value)) => {
+                    // println!("StartTlsClient ignore {:?}", _value);
+                    (
+                        StartTlsClientState::AwaitProceed(xmpp_stream),
+                        Ok(Async::NotReady),
+                    )
+                }
+                Ok(_) => (
+                    StartTlsClientState::AwaitProceed(xmpp_stream),
+                    Ok(Async::NotReady),
+                ),
+                Err(e) => (
+                    StartTlsClientState::AwaitProceed(xmpp_stream),
+                    Err(Error::Protocol(e.into())),
+                ),
+            },
+            StartTlsClientState::StartingTls(mut connect) => match connect.poll() {
+                Ok(Async::Ready(tls_stream)) => {
+                    (StartTlsClientState::Invalid, Ok(Async::Ready(tls_stream)))
+                }
+                Ok(Async::NotReady) => (
+                    StartTlsClientState::StartingTls(connect),
+                    Ok(Async::NotReady),
+                ),
+                Err(e) => (StartTlsClientState::Invalid, Err(e.into())),
+            },
+            StartTlsClientState::Invalid => unreachable!(),
+        };
+
+        self.state = new_state;
+        if retry {
+            self.poll()
+        } else {
+            result
+        }
+    }
+}

tokio-xmpp/src/stream_start.rs

@@ -0,0 +1,125 @@
+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 crate::xmpp_codec::{Packet, XMPPCodec};
+use crate::xmpp_stream::XMPPStream;
+use crate::{Error, ProtocolError};
+
+const NS_XMPP_STREAM: &str = "http://etherx.jabber.org/streams";
+
+pub struct StreamStart<S: AsyncWrite> {
+    state: StreamStartState<S>,
+    jid: Jid,
+    ns: String,
+}
+
+enum StreamStartState<S: AsyncWrite> {
+    SendStart(sink::Send<Framed<S, XMPPCodec>>),
+    RecvStart(Framed<S, XMPPCodec>),
+    RecvFeatures(Framed<S, XMPPCodec>, String),
+    Invalid,
+}
+
+impl<S: AsyncWrite> StreamStart<S> {
+    pub fn from_stream(stream: Framed<S, XMPPCodec>, jid: Jid, ns: String) -> Self {
+        let attrs = [
+            ("to".to_owned(), jid.clone().domain()),
+            ("version".to_owned(), "1.0".to_owned()),
+            ("xmlns".to_owned(), ns.clone()),
+            ("xmlns:stream".to_owned(), NS_XMPP_STREAM.to_owned()),
+        ]
+        .iter()
+        .cloned()
+        .collect();
+        let send = stream.send(Packet::StreamStart(attrs));
+
+        StreamStart {
+            state: StreamStartState::SendStart(send),
+            jid,
+            ns,
+        }
+    }
+}
+
+impl<S: AsyncRead + AsyncWrite> Future for StreamStart<S> {
+    type Item = XMPPStream<S>;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        let old_state = replace(&mut self.state, StreamStartState::Invalid);
+        let mut retry = false;
+
+        let (new_state, result) = match old_state {
+            StreamStartState::SendStart(mut send) => match send.poll() {
+                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;
+                        // TODO: skip RecvFeatures for version < 1.0
+                        (
+                            StreamStartState::RecvFeatures(stream, stream_ns),
+                            Ok(Async::NotReady),
+                        )
+                    } else {
+                        let id = stream_attrs
+                            .get("id")
+                            .ok_or(ProtocolError::NoStreamId)?
+                            .clone();
+                        // 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()),
+            },
+            StreamStartState::RecvFeatures(mut stream, stream_ns) => match stream.poll() {
+                Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
+                    if stanza.is("features", NS_XMPP_STREAM) {
+                        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::NotReady),
+                ),
+                Err(e) => return Err(ProtocolError::from(e).into()),
+            },
+            StreamStartState::Invalid => unreachable!(),
+        };
+
+        self.state = new_state;
+        if retry {
+            self.poll()
+        } else {
+            result
+        }
+    }
+}

tokio-xmpp/src/xmpp_codec.rs

@@ -0,0 +1,532 @@
+//! XML stream parser for XMPP
+
+use crate::{ParseError, ParserError};
+use bytes::{BufMut, BytesMut};
+use xmpp_parsers::Element;
+use quick_xml::Writer as EventWriter;
+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 std::borrow::Cow;
+use tokio_codec::{Decoder, Encoder};
+use xml5ever::interface::Attribute;
+use xml5ever::tokenizer::{Tag, TagKind, Token, TokenSink, XmlTokenizer};
+use xml5ever::buffer_queue::BufferQueue;
+
+/// Anything that can be sent or received on an XMPP/XML stream
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Packet {
+    /// `<stream:stream>` start tag
+    StreamStart(HashMap<String, String>),
+    /// A complete stanza or nonza
+    Stanza(Element),
+    /// Plain text (think whitespace keep-alive)
+    Text(String),
+    /// `</stream:stream>` closing tag
+    StreamEnd,
+}
+
+type QueueItem = Result<Packet, ParserError>;
+
+/// Parser state
+struct ParserSink {
+    // Ready stanzas, shared with XMPPCodec
+    queue: Rc<RefCell<VecDeque<QueueItem>>>,
+    // Parsing stack
+    stack: Vec<Element>,
+    ns_stack: Vec<HashMap<Option<String>, String>>,
+}
+
+impl ParserSink {
+    pub fn new(queue: Rc<RefCell<VecDeque<QueueItem>>>) -> Self {
+        ParserSink {
+            queue,
+            stack: vec![],
+            ns_stack: vec![],
+        }
+    }
+
+    fn push_queue(&self, pkt: Packet) {
+        self.queue.borrow_mut().push_back(Ok(pkt));
+    }
+
+    fn push_queue_error(&self, e: ParserError) {
+        self.queue.borrow_mut().push_back(Err(e));
+    }
+
+    /// Lookup XML namespace declaration for given prefix (or no prefix)
+    fn lookup_ns(&self, prefix: &Option<String>) -> Option<&str> {
+        for nss in self.ns_stack.iter().rev() {
+            if let Some(ns) = nss.get(prefix) {
+                return Some(ns);
+            }
+        }
+
+        None
+    }
+
+    fn handle_start_tag(&mut self, tag: Tag) {
+        let mut nss = HashMap::new();
+        let is_prefix_xmlns = |attr: &Attribute| {
+            attr.name
+                .prefix
+                .as_ref()
+                .map(|prefix| prefix.eq_str_ignore_ascii_case("xmlns"))
+                .unwrap_or(false)
+        };
+        for attr in &tag.attrs {
+            match attr.name.local.as_ref() {
+                "xmlns" => {
+                    nss.insert(None, attr.value.as_ref().to_owned());
+                }
+                prefix if is_prefix_xmlns(attr) => {
+                    nss.insert(Some(prefix.to_owned()), attr.value.as_ref().to_owned());
+                }
+                _ => (),
+            }
+        }
+        self.ns_stack.push(nss);
+
+        let el = {
+            let mut el_builder = Element::builder(tag.name.local.as_ref());
+            if let Some(el_ns) =
+                self.lookup_ns(&tag.name.prefix.map(|prefix| prefix.as_ref().to_owned()))
+            {
+                el_builder = el_builder.ns(el_ns);
+            }
+            for attr in &tag.attrs {
+                match attr.name.local.as_ref() {
+                    "xmlns" => (),
+                    _ if is_prefix_xmlns(attr) => (),
+                    _ => {
+                        let attr_name = if let Some(ref prefix) = attr.name.prefix {
+                            Cow::Owned(format!("{}:{}", prefix, attr.name.local))
+                        } else {
+                            Cow::Borrowed(attr.name.local.as_ref())
+                        };
+                        el_builder = el_builder.attr(attr_name, attr.value.as_ref());
+                    }
+                }
+            }
+            el_builder.build()
+        };
+
+        if self.stack.is_empty() {
+            let attrs = HashMap::from_iter(tag.attrs.iter().map(|attr| {
+                (
+                    attr.name.local.as_ref().to_owned(),
+                    attr.value.as_ref().to_owned(),
+                )
+            }));
+            self.push_queue(Packet::StreamStart(attrs));
+        }
+
+        self.stack.push(el);
+    }
+
+    fn handle_end_tag(&mut self) {
+        let el = self.stack.pop().unwrap();
+        self.ns_stack.pop();
+
+        match self.stack.len() {
+            // </stream:stream>
+            0 => self.push_queue(Packet::StreamEnd),
+            // </stanza>
+            1 => self.push_queue(Packet::Stanza(el)),
+            len => {
+                let parent = &mut self.stack[len - 1];
+                parent.append_child(el);
+            }
+        }
+    }
+}
+
+impl TokenSink for ParserSink {
+    fn process_token(&mut self, token: Token) {
+        match token {
+            Token::TagToken(tag) => match tag.kind {
+                TagKind::StartTag => self.handle_start_tag(tag),
+                TagKind::EndTag => self.handle_end_tag(),
+                TagKind::EmptyTag => {
+                    self.handle_start_tag(tag);
+                    self.handle_end_tag();
+                }
+                TagKind::ShortTag => self.push_queue_error(ParserError::ShortTag),
+            },
+            Token::CharacterTokens(tendril) => match self.stack.len() {
+                0 | 1 => self.push_queue(Packet::Text(tendril.into())),
+                len => {
+                    let el = &mut self.stack[len - 1];
+                    el.append_text_node(tendril);
+                }
+            },
+            Token::EOFToken => self.push_queue(Packet::StreamEnd),
+            Token::ParseError(s) => {
+                // println!("ParseError: {:?}", s);
+                self.push_queue_error(ParserError::Parse(ParseError(s)));
+            }
+            _ => (),
+        }
+    }
+
+    // fn end(&mut self) {
+    // }
+}
+
+/// Stateful encoder/decoder for a bytestream from/to XMPP `Packet`
+pub struct XMPPCodec {
+    /// Outgoing
+    ns: Option<String>,
+    /// Incoming
+    parser: XmlTokenizer<ParserSink>,
+    /// For handling incoming truncated utf8
+    // TODO: optimize using  tendrils?
+    buf: Vec<u8>,
+    /// Shared with ParserSink
+    queue: Rc<RefCell<VecDeque<QueueItem>>>,
+}
+
+impl XMPPCodec {
+    /// Constructor
+    pub fn new() -> Self {
+        let queue = Rc::new(RefCell::new(VecDeque::new()));
+        let sink = ParserSink::new(queue.clone());
+        // TODO: configure parser?
+        let parser = XmlTokenizer::new(sink, Default::default());
+        XMPPCodec {
+            ns: None,
+            parser,
+            queue,
+            buf: vec![],
+        }
+    }
+}
+
+impl Default for XMPPCodec {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Decoder for XMPPCodec {
+    type Item = Packet;
+    type Error = ParserError;
+
+    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        let buf1: Box<dyn AsRef<[u8]>> = if !self.buf.is_empty() && !buf.is_empty() {
+            let mut prefix = std::mem::replace(&mut self.buf, vec![]);
+            prefix.extend_from_slice(buf.take().as_ref());
+            Box::new(prefix)
+        } else {
+            Box::new(buf.take())
+        };
+        let buf1 = buf1.as_ref().as_ref();
+        match from_utf8(buf1) {
+            Ok(mut s) => {
+                s = s.trim();
+                if !s.is_empty() {
+                    // println!("<< {}", s);
+                    let mut buffer_queue = BufferQueue::new();
+                    let tendril = FromIterator::from_iter(s.chars());
+                    buffer_queue.push_back(tendril);
+                    self.parser.feed(&mut buffer_queue);
+                }
+            }
+            // Remedies for truncated utf8
+            Err(e) if e.valid_up_to() >= buf1.len() - 3 => {
+                // Prepare all the valid data
+                let mut b = BytesMut::with_capacity(e.valid_up_to());
+                b.put(&buf1[0..e.valid_up_to()]);
+
+                // Retry
+                let result = self.decode(&mut b);
+
+                // Keep the tail back in
+                self.buf.extend_from_slice(&buf1[e.valid_up_to()..]);
+
+                return result;
+            }
+            Err(e) => {
+                // println!("error {} at {}/{} in {:?}", e, e.valid_up_to(), buf1.len(), buf1);
+                return Err(ParserError::Utf8(e));
+            }
+        }
+
+        match self.queue.borrow_mut().pop_front() {
+            None => Ok(None),
+            Some(result) => result.map(|pkt| Some(pkt)),
+        }
+    }
+
+    fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        self.decode(buf)
+    }
+}
+
+impl Encoder for XMPPCodec {
+    type Item = Packet;
+    type Error = io::Error;
+
+    fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        let remaining = dst.capacity() - dst.len();
+        let max_stanza_size: usize = 2usize.pow(16);
+        if remaining < max_stanza_size {
+            dst.reserve(max_stanza_size - remaining);
+        }
+
+        fn to_io_err<E: Into<Box<dyn std::error::Error + Send + Sync>>>(e: E) -> io::Error {
+            io::Error::new(io::ErrorKind::InvalidInput, e)
+        }
+
+        match item {
+            Packet::StreamStart(start_attrs) => {
+                let mut buf = String::new();
+                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)?;
+                    if name == "xmlns" {
+                        self.ns = Some(value);
+                    }
+                }
+                write!(buf, ">\n")
+                    .map_err(to_io_err)?;
+
+                // print!(">> {}", buf);
+                write!(dst, "{}", buf)
+                    .map_err(to_io_err)
+            }
+            Packet::Stanza(stanza) => {
+                stanza
+                    .write_to_inner(&mut EventWriter::new(WriteBytes::new(dst)))
+                    .and_then(|_| {
+                        // println!(">> {:?}", dst);
+                        Ok(())
+                    })
+                    .map_err(|e| to_io_err(format!("{}", e)))
+            }
+            Packet::Text(text) => {
+                write_text(&text, dst)
+                    .and_then(|_| {
+                        // println!(">> {:?}", dst);
+                        Ok(())
+                    })
+                    .map_err(to_io_err)
+            }
+            Packet::StreamEnd => {
+                write!(dst, "</stream:stream>\n")
+                    .map_err(to_io_err)
+            }
+        }
+    }
+}
+
+/// Write XML-escaped text string
+pub fn write_text<W: Write>(text: &str, writer: &mut W) -> Result<(), std::fmt::Error> {
+    write!(writer, "{}", escape(text))
+}
+
+/// Copied from `RustyXML` for now
+pub fn escape(input: &str) -> String {
+    let mut result = String::with_capacity(input.len());
+
+    for c in input.chars() {
+        match c {
+            '&' => result.push_str("&amp;"),
+            '<' => result.push_str("&lt;"),
+            '>' => result.push_str("&gt;"),
+            '\'' => result.push_str("&apos;"),
+            '"' => result.push_str("&quot;"),
+            o => result.push(o),
+        }
+    }
+    result
+}
+
+/// BytesMut impl only std::fmt::Write but not std::io::Write. The
+/// latter trait is required for minidom's
+/// `Element::write_to_inner()`.
+struct WriteBytes<'a> {
+    dst: &'a mut BytesMut,
+}
+
+impl<'a> WriteBytes<'a> {
+    fn new(dst: &'a mut BytesMut) -> Self {
+        WriteBytes { dst }
+    }
+}
+
+impl<'a> std::io::Write for WriteBytes<'a> {
+    fn write(&mut self, buf: &[u8]) -> std::result::Result<usize, std::io::Error> {
+        self.dst.put_slice(buf);
+        Ok(buf.len())
+    }
+
+    fn flush(&mut self) -> std::result::Result<(), std::io::Error> {
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use bytes::BytesMut;
+
+    #[test]
+    fn test_stream_start() {
+        let mut c = XMPPCodec::new();
+        let mut b = BytesMut::with_capacity(1024);
+        b.put(r"<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' version='1.0' xmlns='jabber:client'>");
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(Some(Packet::StreamStart(_))) => true,
+            _ => false,
+        });
+    }
+
+    #[test]
+    fn test_stream_end() {
+        let mut c = XMPPCodec::new();
+        let mut b = BytesMut::with_capacity(1024);
+        b.put(r"<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' version='1.0' xmlns='jabber:client'>");
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(Some(Packet::StreamStart(_))) => true,
+            _ => false,
+        });
+        b.clear();
+        b.put(r"</stream:stream>");
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(Some(Packet::StreamEnd)) => true,
+            _ => false,
+        });
+    }
+
+    #[test]
+    fn test_truncated_stanza() {
+        let mut c = XMPPCodec::new();
+        let mut b = BytesMut::with_capacity(1024);
+        b.put(r"<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' version='1.0' xmlns='jabber:client'>");
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(Some(Packet::StreamStart(_))) => true,
+            _ => false,
+        });
+
+        b.clear();
+        b.put(r"<test>ß</test");
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(None) => true,
+            _ => false,
+        });
+
+        b.clear();
+        b.put(r">");
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(Some(Packet::Stanza(ref el))) if el.name() == "test" && el.text() == "ß" => true,
+            _ => false,
+        });
+    }
+
+    #[test]
+    fn test_truncated_utf8() {
+        let mut c = XMPPCodec::new();
+        let mut b = BytesMut::with_capacity(1024);
+        b.put(r"<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' version='1.0' xmlns='jabber:client'>");
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(Some(Packet::StreamStart(_))) => true,
+            _ => false,
+        });
+
+        b.clear();
+        b.put(&b"<test>\xc3"[..]);
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(None) => true,
+            _ => false,
+        });
+
+        b.clear();
+        b.put(&b"\x9f</test>"[..]);
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(Some(Packet::Stanza(ref el))) if el.name() == "test" && el.text() == "ß" => true,
+            _ => false,
+        });
+    }
+
+    /// test case for https://gitlab.com/xmpp-rs/tokio-xmpp/issues/3
+    #[test]
+    fn test_atrribute_prefix() {
+        let mut c = XMPPCodec::new();
+        let mut b = BytesMut::with_capacity(1024);
+        b.put(r"<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' version='1.0' xmlns='jabber:client'>");
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(Some(Packet::StreamStart(_))) => true,
+            _ => false,
+        });
+
+        b.clear();
+        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,
+            _ => false,
+        });
+
+    }
+
+    /// By default, encode() only get's a BytesMut that has 8kb space reserved.
+    #[test]
+    fn test_large_stanza() {
+        use futures::{Future, Sink};
+        use std::io::Cursor;
+        use tokio_codec::FramedWrite;
+        let framed = FramedWrite::new(Cursor::new(vec![]), XMPPCodec::new());
+        let mut text = "".to_owned();
+        for _ in 0..2usize.pow(15) {
+            text = text + "A";
+        }
+        let stanza = Element::builder("message")
+            .append(Element::builder("body").append(text.as_ref()).build())
+            .build();
+        let framed = framed.send(Packet::Stanza(stanza)).wait().expect("send");
+        assert_eq!(
+            framed.get_ref().get_ref(),
+            &("<message><body>".to_owned() + &text + "</body></message>").as_bytes()
+        );
+    }
+
+    #[test]
+    fn test_lone_whitespace() {
+        let mut c = XMPPCodec::new();
+        let mut b = BytesMut::with_capacity(1024);
+        b.put(r"<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' version='1.0' xmlns='jabber:client'>");
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(Some(Packet::StreamStart(_))) => true,
+            _ => false,
+        });
+
+        b.clear();
+        b.put(r" ");
+        let r = c.decode(&mut b);
+        assert!(match r {
+            Ok(None) => true,
+            _ => false,
+        });
+    }
+}

tokio-xmpp/src/xmpp_stream.rs

@@ -0,0 +1,92 @@
+//! `XMPPStream` is the common container for all XMPP network connections
+
+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 crate::stream_start::StreamStart;
+use crate::xmpp_codec::{Packet, XMPPCodec};
+
+/// <stream:stream> namespace
+pub const NS_XMPP_STREAM: &str = "http://etherx.jabber.org/streams";
+
+/// Wraps a `stream`
+pub struct XMPPStream<S> {
+    /// The local Jabber-Id
+    pub jid: Jid,
+    /// Codec instance
+    pub stream: Framed<S, XMPPCodec>,
+    /// `<stream:features/>` for XMPP version 1.0
+    pub stream_features: Element,
+    /// Root namespace
+    ///
+    /// This is different for either c2s, s2s, or component
+    /// connections.
+    pub ns: String,
+}
+
+impl<S: AsyncRead + AsyncWrite> XMPPStream<S> {
+    /// Constructor
+    pub fn new(
+        jid: Jid,
+        stream: Framed<S, XMPPCodec>,
+        ns: String,
+        stream_features: Element,
+    ) -> Self {
+        XMPPStream {
+            jid,
+            stream,
+            stream_features,
+            ns,
+        }
+    }
+
+    /// Send a `<stream:stream>` start tag
+    pub fn start(stream: S, jid: Jid, ns: String) -> StreamStart<S> {
+        let xmpp_stream = Framed::new(stream, XMPPCodec::new());
+        StreamStart::from_stream(xmpp_stream, jid, ns)
+    }
+
+    /// Unwraps the inner stream
+    pub fn into_inner(self) -> S {
+        self.stream.into_inner()
+    }
+
+    /// Re-run `start()`
+    pub fn restart(self) -> StreamStart<S> {
+        Self::start(self.stream.into_inner(), self.jid, self.ns)
+    }
+}
+
+impl<S: AsyncWrite> XMPPStream<S> {
+    /// Convenience method
+    pub fn send_stanza<E: Into<Element>>(self, e: E) -> Send<Self> {
+        self.send(Packet::Stanza(e.into()))
+    }
+}
+
+/// Proxy to self.stream
+impl<S: AsyncWrite> Sink for XMPPStream<S> {
+    type SinkItem = <Framed<S, XMPPCodec> as Sink>::SinkItem;
+    type SinkError = <Framed<S, XMPPCodec> as Sink>::SinkError;
+
+    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
+        self.stream.start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        self.stream.poll_complete()
+    }
+}
+
+/// Proxy to self.stream
+impl<S: AsyncRead> Stream for XMPPStream<S> {
+    type Item = <Framed<S, XMPPCodec> as Stream>::Item;
+    type Error = <Framed<S, XMPPCodec> as Stream>::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.stream.poll()
+    }
+}