Merge tokio-xmpp project

tokio-xmpp/.gitignore

@@ -0,0 +1 @@
+target

tokio-xmpp/.gitlab-ci.yml

@@ -0,0 +1,14 @@
+stages:
+  - build
+rust-latest:
+  stage: build
+  image: rust:latest
+  script:
+    - cargo build --verbose
+    - cargo test --verbose
+rust-nightly:
+  stage: build
+  image: rustlang/rust:nightly
+  script:
+    - cargo build --verbose
+    - cargo test --verbose

tokio-xmpp/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "tokio-xmpp"
+version = "1.0.1"
+authors = ["Astro <astro@spaceboyz.net>", "Emmanuel Gil Peyrot <linkmauve@linkmauve.fr>", "pep <pep+code@bouah.net>", "O01eg <o01eg@yandex.ru>"]
+description = "Asynchronous XMPP for Rust with tokio"
+license = "MPL-2.0"
+homepage = "https://gitlab.com/xmpp-rs/tokio-xmpp"
+repository = "https://gitlab.com/xmpp-rs/tokio-xmpp"
+documentation = "https://docs.rs/tokio-xmpp"
+categories = ["asynchronous", "network-programming"]
+keywords = ["xmpp", "tokio"]
+edition = "2018"
+
+[dependencies]
+bytes = "0.4"
+futures = "0.1"
+idna = "0.2"
+native-tls = "0.2"
+sasl = "0.4"
+tokio = "0.1"
+tokio-codec = "0.1"
+trust-dns-resolver = "0.12"
+trust-dns-proto = "0.8"
+tokio-io = "0.1"
+tokio-tls = "0.2"
+quick-xml = "0.17"
+xml5ever = "0.15"
+xmpp-parsers = "0.15"

tokio-xmpp/README.md

@@ -0,0 +1,6 @@
+# TODO
+
+- [ ] minidom ns
+- [ ] replace debug output with log crate
+- [ ] customize tls verify?
+- [ ] more tests

tokio-xmpp/examples/contact_addr.rs

@@ -0,0 +1,130 @@
+use futures::{future, Sink, Stream};
+use std::convert::TryFrom;
+use std::env::args;
+use std::process::exit;
+use tokio::runtime::current_thread::Runtime;
+use tokio_xmpp::{Client, xmpp_codec::Packet};
+use xmpp_parsers::{
+    Element,
+    Jid,
+    ns,
+    iq::{
+        Iq,
+        IqType,
+    },
+    disco::{
+        DiscoInfoResult,
+        DiscoInfoQuery,
+    },
+    server_info::ServerInfo,
+};
+
+fn main() {
+    let args: Vec<String> = args().collect();
+    if args.len() != 4 {
+        println!("Usage: {} <jid> <password> <target>", args[0]);
+        exit(1);
+    }
+    let jid = &args[1];
+    let password = &args[2];
+    let target = &args[3];
+
+    // tokio_core context
+    let mut rt = Runtime::new().unwrap();
+    // Client instance
+    let client = Client::new(jid, password).unwrap();
+
+    // Make the two interfaces for sending and receiving independent
+    // of each other so we can move one into a closure.
+    let (mut sink, stream) = client.split();
+    // Wrap sink in Option so that we can take() it for the send(self)
+    // to consume and return it back when ready.
+    let mut send = move |packet| {
+        sink.start_send(packet).expect("start_send");
+    };
+    // Main loop, processes events
+    let mut wait_for_stream_end = false;
+    let done = stream.for_each(|event| {
+        if wait_for_stream_end {
+            /* Do Nothing. */
+        } else if event.is_online() {
+            println!("Online!");
+
+            let target_jid: Jid = target.clone().parse().unwrap();
+            let iq = make_disco_iq(target_jid);
+            println!("Sending disco#info request to {}", target.clone());
+            println!(">> {}", String::from(&iq));
+            send(Packet::Stanza(iq));
+        } else if let Some(stanza) = event.into_stanza() {
+            if stanza.is("iq", "jabber:client") {
+                let iq = Iq::try_from(stanza).unwrap();
+                if let IqType::Result(Some(payload)) = iq.payload {
+                    if payload.is("query", ns::DISCO_INFO) {
+                        if let Ok(disco_info) = DiscoInfoResult::try_from(payload) {
+                            for ext in disco_info.extensions {
+                                if let Ok(server_info) = ServerInfo::try_from(ext) {
+                                    print_server_info(server_info);
+                                    wait_for_stream_end = true;
+                                    send(Packet::StreamEnd);
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        Box::new(future::ok(()))
+    });
+
+    // Start polling `done`
+    match rt.block_on(done) {
+        Ok(_) => (),
+        Err(e) => {
+            println!("Fatal: {}", e);
+            ()
+        }
+    }
+}
+
+fn make_disco_iq(target: Jid) -> Element {
+    Iq::from_get("disco", DiscoInfoQuery { node: None })
+        .with_id(String::from("contact"))
+        .with_to(target)
+        .into()
+}
+
+fn convert_field(field: Vec<String>) -> String {
+    field.iter()
+    .fold((field.len(), String::new()), |(l, mut acc), s| {
+        acc.push('<');
+        acc.push_str(&s);
+        acc.push('>');
+        if l > 1 {
+            acc.push(',');
+            acc.push(' ');
+        }
+        (0, acc)
+    }).1
+}
+
+fn print_server_info(server_info: ServerInfo) {
+    if server_info.abuse.len() != 0 {
+        println!("abuse: {}", convert_field(server_info.abuse));
+    }
+    if server_info.admin.len() != 0 {
+        println!("admin: {}", convert_field(server_info.admin));
+    }
+    if server_info.feedback.len() != 0 {
+        println!("feedback: {}", convert_field(server_info.feedback));
+    }
+    if server_info.sales.len() != 0 {
+        println!("sales: {}", convert_field(server_info.sales));
+    }
+    if server_info.security.len() != 0 {
+        println!("security: {}", convert_field(server_info.security));
+    }
+    if server_info.support.len() != 0 {
+        println!("support: {}", convert_field(server_info.support));
+    }
+}

tokio-xmpp/examples/download_avatars.rs

@@ -0,0 +1,232 @@
+use futures::{future, Future, Sink, Stream};
+use std::convert::TryFrom;
+use std::env::args;
+use std::fs::{create_dir_all, File};
+use std::io::{self, Write};
+use std::process::exit;
+use std::str::FromStr;
+use tokio::runtime::current_thread::Runtime;
+use tokio_xmpp::{Client, Packet};
+use xmpp_parsers::{
+    avatar::{Data as AvatarData, Metadata as AvatarMetadata},
+    caps::{compute_disco, hash_caps, Caps},
+    disco::{DiscoInfoQuery, DiscoInfoResult, Feature, Identity},
+    hashes::Algo,
+    iq::{Iq, IqType},
+    message::Message,
+    ns,
+    presence::{Presence, Type as PresenceType},
+    pubsub::{
+        event::PubSubEvent,
+        pubsub::{Items, PubSub},
+        NodeName,
+    },
+    stanza_error::{StanzaError, ErrorType, DefinedCondition},
+    Jid,
+};
+
+fn main() {
+    let args: Vec<String> = args().collect();
+    if args.len() != 3 {
+        println!("Usage: {} <jid> <password>", args[0]);
+        exit(1);
+    }
+    let jid = &args[1];
+    let password = &args[2];
+
+    // tokio_core context
+    let mut rt = Runtime::new().unwrap();
+    // Client instance
+    let client = Client::new(jid, password).unwrap();
+
+    // Make the two interfaces for sending and receiving independent
+    // of each other so we can move one into a closure.
+    let (sink, stream) = client.split();
+
+    // Create outgoing pipe
+    let (mut tx, rx) = futures::unsync::mpsc::unbounded();
+    rt.spawn(
+        rx.forward(
+            sink.sink_map_err(|_| panic!("Pipe"))
+        )
+            .map(|(rx, mut sink)| {
+                drop(rx);
+                let _ = sink.close();
+            })
+            .map_err(|e| {
+                panic!("Send error: {:?}", e);
+            })
+    );
+
+    let disco_info = make_disco();
+
+    // Main loop, processes events
+    let mut wait_for_stream_end = false;
+    let done = stream.for_each(move |event| {
+        // Helper function to send an iq error.
+        let mut send_error = |to, id, type_, condition, text: &str| {
+            let error = StanzaError::new(type_, condition, "en", text);
+            let iq = Iq::from_error(id, error)
+                .with_to(to);
+            tx.start_send(Packet::Stanza(iq.into())).unwrap();
+        };
+
+        if wait_for_stream_end {
+            /* Do nothing */
+        } else if event.is_online() {
+            println!("Online!");
+
+            let caps = get_disco_caps(&disco_info, "https://gitlab.com/xmpp-rs/tokio-xmpp");
+            let presence = make_presence(caps);
+            tx.start_send(Packet::Stanza(presence.into())).unwrap();
+        } else if let Some(stanza) = event.into_stanza() {
+            if stanza.is("iq", "jabber:client") {
+                let iq = Iq::try_from(stanza).unwrap();
+                if let IqType::Get(payload) = iq.payload {
+                    if payload.is("query", ns::DISCO_INFO) {
+                        let query = DiscoInfoQuery::try_from(payload);
+                        match query {
+                            Ok(query) => {
+                                let mut disco = disco_info.clone();
+                                disco.node = query.node;
+                                let iq = Iq::from_result(iq.id, Some(disco))
+                                    .with_to(iq.from.unwrap());
+                                tx.start_send(Packet::Stanza(iq.into())).unwrap();
+                            },
+                            Err(err) => {
+                                send_error(iq.from.unwrap(), iq.id, ErrorType::Modify, DefinedCondition::BadRequest, &format!("{}", err));
+                            },
+                        }
+                    } else {
+                        // We MUST answer unhandled get iqs with a service-unavailable error.
+                        send_error(iq.from.unwrap(), iq.id, ErrorType::Cancel, DefinedCondition::ServiceUnavailable, "No handler defined for this kind of iq.");
+                    }
+                } else if let IqType::Result(Some(payload)) = iq.payload {
+                    if payload.is("pubsub", ns::PUBSUB) {
+                        let pubsub = PubSub::try_from(payload).unwrap();
+                        let from =
+                            iq.from.clone().unwrap_or(Jid::from_str(jid).unwrap());
+                        handle_iq_result(pubsub, &from);
+                    }
+                } else if let IqType::Set(_) = iq.payload {
+                    // We MUST answer unhandled set iqs with a service-unavailable error.
+                    send_error(iq.from.unwrap(), iq.id, ErrorType::Cancel, DefinedCondition::ServiceUnavailable, "No handler defined for this kind of iq.");
+                }
+            } else if stanza.is("message", "jabber:client") {
+                let message = Message::try_from(stanza).unwrap();
+                let from = message.from.clone().unwrap();
+                if let Some(body) = message.get_best_body(vec!["en"]) {
+                    if body.1 .0 == "die" {
+                        println!("Secret die command triggered by {}", from);
+                        wait_for_stream_end = true;
+                        tx.start_send(Packet::StreamEnd).unwrap();
+                    }
+                }
+                for child in message.payloads {
+                    if child.is("event", ns::PUBSUB_EVENT) {
+                        let event = PubSubEvent::try_from(child).unwrap();
+                        if let PubSubEvent::PublishedItems { node, items } = event {
+                            if node.0 == ns::AVATAR_METADATA {
+                                for item in items.into_iter() {
+                                    let payload = item.payload.clone().unwrap();
+                                    if payload.is("metadata", ns::AVATAR_METADATA) {
+                                        // TODO: do something with these metadata.
+                                        let _metadata = AvatarMetadata::try_from(payload).unwrap();
+                                        println!(
+                                            "{} has published an avatar, downloading...",
+                                            from.clone()
+                                        );
+                                        let iq = download_avatar(from.clone());
+                                        tx.start_send(Packet::Stanza(iq.into())).unwrap();
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            } else if stanza.is("presence", "jabber:client") {
+                // Nothing to do here.
+            } else {
+                panic!("Unknown stanza: {}", String::from(&stanza));
+            }
+        }
+
+        future::ok(())
+    });
+
+    // Start polling `done`
+    match rt.block_on(done) {
+        Ok(_) => (),
+        Err(e) => {
+            println!("Fatal: {}", e);
+            ()
+        }
+    }
+}
+
+fn make_disco() -> DiscoInfoResult {
+    let identities = vec![Identity::new("client", "bot", "en", "tokio-xmpp")];
+    let features = vec![
+        Feature::new(ns::DISCO_INFO),
+        Feature::new(format!("{}+notify", ns::AVATAR_METADATA)),
+    ];
+    DiscoInfoResult {
+        node: None,
+        identities,
+        features,
+        extensions: vec![],
+    }
+}
+
+fn get_disco_caps(disco: &DiscoInfoResult, node: &str) -> Caps {
+    let caps_data = compute_disco(disco);
+    let hash = hash_caps(&caps_data, Algo::Sha_1).unwrap();
+    Caps::new(node, hash)
+}
+
+// Construct a <presence/>
+fn make_presence(caps: Caps) -> Presence {
+    let mut presence = Presence::new(PresenceType::None)
+        .with_priority(-1);
+    presence.set_status("en", "Downloading avatars.");
+    presence.add_payload(caps);
+    presence
+}
+
+fn download_avatar(from: Jid) -> Iq {
+    Iq::from_get("coucou", PubSub::Items(Items {
+        max_items: None,
+        node: NodeName(String::from(ns::AVATAR_DATA)),
+        subid: None,
+        items: Vec::new(),
+    }))
+    .with_to(from)
+}
+
+fn handle_iq_result(pubsub: PubSub, from: &Jid) {
+    if let PubSub::Items(items) = pubsub {
+        if items.node.0 == ns::AVATAR_DATA {
+            for item in items.items {
+                match (item.id.clone(), item.payload.clone()) {
+                    (Some(id), Some(payload)) => {
+                        let data = AvatarData::try_from(payload).unwrap();
+                        save_avatar(from, id.0, &data.data).unwrap();
+                    }
+                    _ => {}
+                }
+            }
+        }
+    }
+}
+
+fn save_avatar(from: &Jid, id: String, data: &[u8]) -> io::Result<()> {
+    let directory = format!("data/{}", from);
+    let filename = format!("data/{}/{}", from, id);
+    println!(
+        "Saving avatar from {} to {}.",
+        from, filename
+    );
+    create_dir_all(directory)?;
+    let mut file = File::create(filename)?;
+    file.write_all(data)
+}

tokio-xmpp/examples/echo_bot.rs

@@ -0,0 +1,106 @@
+use futures::{future, Future, Sink, Stream};
+use std::convert::TryFrom;
+use std::env::args;
+use std::process::exit;
+use tokio::runtime::current_thread::Runtime;
+use tokio_xmpp::{Client, Packet};
+use xmpp_parsers::{Jid, Element};
+use xmpp_parsers::message::{Body, Message, MessageType};
+use xmpp_parsers::presence::{Presence, Show as PresenceShow, Type as PresenceType};
+
+fn main() {
+    let args: Vec<String> = args().collect();
+    if args.len() != 3 {
+        println!("Usage: {} <jid> <password>", args[0]);
+        exit(1);
+    }
+    let jid = &args[1];
+    let password = &args[2];
+
+    // tokio_core context
+    let mut rt = Runtime::new().unwrap();
+    // Client instance
+    let client = Client::new(jid, password).unwrap();
+
+    // Make the two interfaces for sending and receiving independent
+    // of each other so we can move one into a closure.
+    let (sink, stream) = client.split();
+
+    // Create outgoing pipe
+    let (mut tx, rx) = futures::unsync::mpsc::unbounded();
+    rt.spawn(
+        rx.forward(
+            sink.sink_map_err(|_| panic!("Pipe"))
+        )
+            .map(|(rx, mut sink)| {
+                drop(rx);
+                let _ = sink.close();
+            })
+            .map_err(|e| {
+                panic!("Send error: {:?}", e);
+            })
+    );
+
+    // Main loop, processes events
+    let mut wait_for_stream_end = false;
+    let done = stream.for_each(move |event| {
+        if wait_for_stream_end {
+            /* Do nothing */
+        } else if event.is_online() {
+            let jid = event.get_jid()
+                .map(|jid| format!("{}", jid))
+                .unwrap_or("unknown".to_owned());
+            println!("Online at {}", jid);
+
+            let presence = make_presence();
+            tx.start_send(Packet::Stanza(presence)).unwrap();
+        } else if let Some(message) = event
+            .into_stanza()
+            .and_then(|stanza| Message::try_from(stanza).ok())
+        {
+            match (message.from, message.bodies.get("")) {
+                (Some(ref from), Some(ref body)) if body.0 == "die" => {
+                    println!("Secret die command triggered by {}", from);
+                    wait_for_stream_end = true;
+                    tx.start_send(Packet::StreamEnd).unwrap();
+                }
+                (Some(ref from), Some(ref body)) => {
+                    if message.type_ != MessageType::Error {
+                        // This is a message we'll echo
+                        let reply = make_reply(from.clone(), &body.0);
+                        tx.start_send(Packet::Stanza(reply)).unwrap();
+                    }
+                }
+                _ => {}
+            }
+        }
+
+        future::ok(())
+    });
+
+    // Start polling `done`
+    match rt.block_on(done) {
+        Ok(_) => (),
+        Err(e) => {
+            println!("Fatal: {}", e);
+            ()
+        }
+    }
+}
+
+// Construct a <presence/>
+fn make_presence() -> Element {
+    let mut presence = Presence::new(PresenceType::None);
+    presence.show = Some(PresenceShow::Chat);
+    presence
+        .statuses
+        .insert(String::from("en"), String::from("Echoing messages."));
+    presence.into()
+}
+
+// Construct a chat <message/>
+fn make_reply(to: Jid, body: &str) -> Element {
+    let mut message = Message::new(Some(to));
+    message.bodies.insert(String::new(), Body(body.to_owned()));
+    message.into()
+}

tokio-xmpp/examples/echo_component.rs

@@ -0,0 +1,99 @@
+use futures::{future, Sink, Stream};
+use std::convert::TryFrom;
+use std::env::args;
+use std::process::exit;
+use std::str::FromStr;
+use tokio::runtime::current_thread::Runtime;
+use tokio_xmpp::Component;
+use xmpp_parsers::{Jid, Element};
+use xmpp_parsers::message::{Body, Message, MessageType};
+use xmpp_parsers::presence::{Presence, Show as PresenceShow, Type as PresenceType};
+
+fn main() {
+    let args: Vec<String> = args().collect();
+    if args.len() < 3 || args.len() > 5 {
+        println!("Usage: {} <jid> <password> [server] [port]", args[0]);
+        exit(1);
+    }
+    let jid = &args[1];
+    let password = &args[2];
+    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);
+
+    // tokio_core context
+    let mut rt = Runtime::new().unwrap();
+    // Component instance
+    println!("{} {} {} {}", jid, password, server, port);
+    let component = Component::new(jid, password, server, port).unwrap();
+
+    // Make the two interfaces for sending and receiving independent
+    // of each other so we can move one into a closure.
+    println!("Got it: {}", component.jid.clone());
+    let (mut sink, stream) = component.split();
+    // Wrap sink in Option so that we can take() it for the send(self)
+    // to consume and return it back when ready.
+    let mut send = move |stanza| {
+        sink.start_send(stanza).expect("start_send");
+    };
+    // Main loop, processes events
+    let done = stream.for_each(|event| {
+        if event.is_online() {
+            println!("Online!");
+
+            // 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(),
+            );
+            send(presence);
+        } else if let Some(message) = event
+            .into_stanza()
+            .and_then(|stanza| Message::try_from(stanza).ok())
+        {
+            // This is a message we'll echo
+            match (message.from, message.bodies.get("")) {
+                (Some(from), Some(body)) => {
+                    if message.type_ != MessageType::Error {
+                        let reply = make_reply(from, &body.0);
+                        send(reply);
+                    }
+                }
+                _ => (),
+            }
+        }
+
+        Box::new(future::ok(()))
+    });
+
+    // Start polling `done`
+    match rt.block_on(done) {
+        Ok(_) => (),
+        Err(e) => {
+            println!("Fatal: {}", e);
+            ()
+        }
+    }
+}
+
+// Construct a <presence/>
+fn make_presence(from: Jid, to: Jid) -> Element {
+    let mut presence = Presence::new(PresenceType::None);
+    presence.from = Some(from);
+    presence.to = Some(to);
+    presence.show = Some(PresenceShow::Chat);
+    presence
+        .statuses
+        .insert(String::from("en"), String::from("Echoing messages."));
+    presence.into()
+}
+
+// Construct a chat <message/>
+fn make_reply(to: Jid, body: &str) -> Element {
+    let mut message = Message::new(Some(to));
+    message.bodies.insert(String::new(), Body(body.to_owned()));
+    message.into()
+}

tokio-xmpp/logo.svg

@@ -0,0 +1,172 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<!-- Generator: Adobe Illustrator 13.0.2, SVG Export Plug-In . SVG Version: 6.00 Build 14948)  -->
+
+<svg
+   xmlns:dc="http://purl.org/dc/elements/1.1/"
+   xmlns:cc="http://creativecommons.org/ns#"
+   xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+   xmlns:svg="http://www.w3.org/2000/svg"
+   xmlns="http://www.w3.org/2000/svg"
+   xmlns:xlink="http://www.w3.org/1999/xlink"
+   xmlns:sodipodi="http://sodipodi.sourceforge.net/DTD/sodipodi-0.dtd"
+   xmlns:inkscape="http://www.inkscape.org/namespaces/inkscape"
+   id="Layer_1"
+   xml:space="preserve"
+   height="1070.54"
+   viewBox="0 0 1251.4018 1070.5223"
+   width="1251.4302"
+   version="1.1"
+   y="0px"
+   x="0px"
+   enable-background="new 0 0 176.486 181.437"
+   sodipodi:docname="logo.svg"
+   inkscape:version="0.92.3 (2405546, 2018-03-11)"
+   inkscape:export-filename="/tmp/tokio-xmpp.png"
+   inkscape:export-xdpi="63.689999"
+   inkscape:export-ydpi="63.689999"><metadata
+   id="metadata41"><rdf:RDF><cc:Work
+       rdf:about=""><dc:format>image/svg+xml</dc:format><dc:type
+         rdf:resource="http://purl.org/dc/dcmitype/StillImage" /><dc:title></dc:title></cc:Work></rdf:RDF></metadata><defs
+   id="defs39"><radialGradient
+     inkscape:collect="always"
+     xlink:href="#SVGID_1_"
+     id="radialGradient4654"
+     cx="166.88985"
+     cy="40.555923"
+     fx="166.88985"
+     fy="40.555923"
+     r="498.49179"
+     gradientTransform="matrix(1.2093833,0,0,1.0737613,423.8704,491.7138)"
+     gradientUnits="userSpaceOnUse" /><linearGradient
+     inkscape:collect="always"
+     xlink:href="#SVGID_1_"
+     id="linearGradient4656"
+     gradientUnits="userSpaceOnUse"
+     gradientTransform="translate(1885.557,-704.53708)"
+     x1="-1807.2"
+     y1="125.86"
+     x2="-1807.2"
+     y2="0.00048828" /></defs><sodipodi:namedview
+   pagecolor="#ffffff"
+   bordercolor="#666666"
+   borderopacity="1"
+   objecttolerance="10"
+   gridtolerance="10"
+   guidetolerance="10"
+   inkscape:pageopacity="0"
+   inkscape:pageshadow="2"
+   inkscape:window-width="1446"
+   inkscape:window-height="1056"
+   id="namedview37"
+   showgrid="false"
+   fit-margin-top="0"
+   fit-margin-left="0"
+   fit-margin-right="0"
+   fit-margin-bottom="0"
+   inkscape:zoom="0.32517636"
+   inkscape:cx="488.51154"
+   inkscape:cy="460.25958"
+   inkscape:window-x="0"
+   inkscape:window-y="20"
+   inkscape:window-maximized="1"
+   inkscape:current-layer="Layer_1" />
+<linearGradient
+   id="SVGID_1_"
+   y2="0.00048828"
+   gradientUnits="userSpaceOnUse"
+   x2="-1807.2"
+   gradientTransform="translate(1885.557,-704.53708)"
+   y1="125.86"
+   x1="-1807.2">
+	<stop
+   stop-color="#1b3967"
+   offset=".011"
+   id="stop2" />
+	<stop
+   stop-color="#13b5ea"
+   offset=".467"
+   id="stop4" />
+	<stop
+   stop-color="#002b5c"
+   offset=".9945"
+   id="stop6" />
+</linearGradient>
+
+
+<linearGradient
+   id="SVGID_2_"
+   y2="1.279e-13"
+   gradientUnits="userSpaceOnUse"
+   x2="-1073.2"
+   gradientTransform="matrix(-1,0,0,1,-1038.643,-704.53708)"
+   y1="126.85"
+   x1="-1073.2">
+	<stop
+   stop-color="#1b3967"
+   offset=".011"
+   id="stop13" />
+	<stop
+   stop-color="#13b5ea"
+   offset=".467"
+   id="stop15" />
+	<stop
+   stop-color="#002b5c"
+   offset=".9945"
+   id="stop17" />
+</linearGradient>
+
+
+
+
+	
+	
+	
+	
+<path
+   style="fill:url(#radialGradient4654);fill-opacity:1;stroke-width:1.20936334"
+   d="m 314.30911,1061.7683 c -8.97073,-5.2395 -16.48089,-10.0377 -16.68926,-10.6628 -0.26982,-0.8095 166.28356,-288.11102 202.75565,-349.74973 l 2.54893,-4.30777 -9.36559,-8.39055 c -11.144,-9.98383 -22.35733,-21.92978 -28.0257,-29.85676 -2.25924,-3.15946 -4.2669,-5.74448 -4.46148,-5.74448 -0.19457,0 -30.6124,17.4714 -67.59521,38.82531 -36.98279,21.35393 -67.38191,38.63271 -67.55361,38.39729 -3.29619,-4.51935 -18.81062,-33.32415 -18.43899,-34.23468 0.28394,-0.69568 26.01362,-15.92783 57.17709,-33.84921 31.16345,-17.92139 61.09657,-35.1893 66.51803,-38.37317 l 9.85719,-5.78884 -4.65622,-13.21887 c -4.07486,-11.56836 -8.04326,-28.12263 -10.25106,-42.76242 l -0.68393,-4.53511 H 224.14076 22.836563 v -19.9545 -19.9545 H 224.14076 425.44495 l 0.68393,-4.5351 c 2.20688,-14.63373 6.17521,-31.19125 10.24412,-42.74274 l 4.64929,-13.19918 -9.85025,-5.77551 c -5.41765,-3.17654 -35.35061,-20.45332 -66.51769,-38.39286 -31.16709,-17.93955 -56.89677,-33.18655 -57.17709,-33.88223 -0.36614,-0.90866 15.08023,-29.58386 18.43579,-34.22491 0.16633,-0.23003 30.60498,17.07925 67.64147,38.4651 37.03649,21.38584 67.48667,38.71678 67.66708,38.51319 0.18041,-0.20358 3.5815,-4.45175 7.55796,-9.44037 8.55901,-10.73754 23.43746,-25.50319 33.56957,-33.31503 3.99089,-3.07699 7.42447,-6.0522 7.63019,-6.6116 0.20571,-0.55942 -45.66778,-80.05268 -101.94109,-176.65173 C 349.23634,94.87494 306.17509,19.72792 306.78588,19.11713 308.19161,17.71141 338.56097,0 339.56561,0 c 0.4291,0 47.02676,79.44412 103.55039,176.54249 78.60729,135.03455 103.19671,176.3654 104.58425,175.78931 14.01846,-5.82028 30.16056,-10.04645 55.32837,-14.48557 l 2.72106,-0.47993 v -75.56217 -75.56217 h 19.9545 19.9545 v 75.56217 75.56217 l 2.72106,0.54898 c 1.49659,0.30194 7.8911,1.43066 14.21003,2.50826 6.31892,1.07761 18.17561,4.17686 26.3482,6.88724 8.1726,2.71038 15.0185,4.72127 15.21314,4.46861 0.19462,-0.25264 46.29667,-79.48313 102.44897,-176.06777 C 862.75239,79.12698 909.11172,0.07996 909.62079,0.05159 c 0.50906,-0.0284 8.26445,4.16607 17.23414,9.32099 11.34048,6.51737 16.17929,9.93831 15.88418,11.22974 -0.23343,1.02145 -46.19384,80.42165 -102.13425,176.4449 l -101.70985,174.58771 8.10553,6.13315 c 11.47436,8.68226 26.67576,23.56058 35.4381,34.68493 6.76288,8.58593 7.73722,9.38055 10.03334,8.18278 1.39869,-0.72964 30.57007,-17.49239 64.82528,-37.25058 34.25522,-19.75818 63.46697,-36.5502 64.91498,-37.3156 2.49056,-1.31647 3.16664,-0.47069 12.51549,15.65711 5.43546,9.37681 9.42076,17.4631 8.85617,17.96951 -0.5646,0.5064 -30.14195,17.6627 -65.72746,38.12508 -35.58552,20.46239 -65.35304,37.81597 -66.15006,38.56352 -1.11691,1.04762 -0.36064,4.54738 3.29904,15.26687 4.30684,12.61506 9.00358,32.4176 9.89584,41.72304 l 0.34788,3.62809 201.66136,0.30589 201.6614,0.30588 v 19.94741 19.94741 l -201.6614,0.30589 -201.66136,0.30588 -0.34788,3.62809 c -0.89828,9.3683 -5.60703,29.15966 -10.02924,42.15394 -2.68469,7.88877 -4.71343,14.45948 -4.50826,14.60158 0.20516,0.14208 30.03265,17.30357 66.28331,38.13661 36.25066,20.83304 66.37223,38.28975 66.93683,38.7927 0.56459,0.50293 -3.42071,8.58638 -8.85617,17.96319 -9.34354,16.11862 -10.02635,16.97321 -12.51549,15.66392 -1.44801,-0.76167 -30.65976,-17.55164 -64.91498,-37.31105 -34.25521,-19.75941 -63.42659,-36.52524 -64.82528,-37.25737 -2.2992,-1.20352 -3.27874,-0.39721 -10.21442,8.40809 -4.21924,5.3566 -13.12221,14.88134 -19.78439,21.16609 -10.1228,9.54933 -11.90518,11.76555 -10.84787,13.48828 1.04335,1.69992 192.71096,330.83154 202.00872,346.88934 l 2.89322,4.9967 -17.11682,9.9052 c -9.41425,5.4479 -17.52207,9.4578 -18.01738,8.9109 -0.49531,-0.5468 -46.48965,-79.36099 -102.20961,-175.14256 -55.71998,-95.78159 -101.69517,-174.5851 -102.16708,-175.11893 -0.47193,-0.53385 -4.0675,0.47777 -7.99015,2.24805 -16.83432,7.59718 -33.19012,12.19105 -60.64646,17.03382 l -2.72106,0.47994 v 75.56218 75.56216 h -19.9545 -19.9545 V 813.3199 737.75772 l -2.72106,-0.56168 c -1.49659,-0.30894 -5.44214,-0.91495 -8.76789,-1.34671 -10.96151,-1.42303 -32.49085,-7.42089 -44.73737,-12.4634 -6.62528,-2.72794 -12.43791,-4.56618 -12.91696,-4.08498 -0.47905,0.4812 -46.66546,79.51377 -102.63647,175.62793 -55.971,96.11415 -102.12208,175.11582 -102.55794,175.55932 -0.43585,0.4435 -8.13215,-3.4805 -17.10288,-8.7199 z"
+   id="path4674" /><path
+   style="fill:url(#radialGradient4654);fill-opacity:1;stroke-width:1.20936334"
+   d="m 615.24584,987.5771 c -16.03866,-5.72964 -23.70609,-24.21313 -16.75096,-40.38075 3.16819,-7.36465 7.25094,-11.42494 14.88068,-14.79874 11.24336,-4.97174 24.69485,-2.54323 33.23901,6.00096 5.52442,5.5244 7.6728,10.22156 8.41406,18.39619 1.94732,21.47547 -19.46247,38.04153 -39.78279,30.78234 z"
+   id="path4668" /><path
+   style="fill:url(#radialGradient4654);fill-opacity:1;stroke-width:1.20936334"
+   d="m 254.0458,776.59921 c -20.23221,-7.20535 -26.62109,-33.06165 -11.95085,-48.3661 18.83855,-19.65296 51.66062,-6.70059 51.60062,20.36281 -0.0456,20.59195 -20.2976,34.89522 -39.64977,28.00329 z"
+   id="path4666" /><path
+   style="fill:url(#radialGradient4654);fill-opacity:1;stroke-width:1.20936334"
+   d="m 977.24507,776.53001 c -12.06564,-4.16718 -19.66881,-14.82142 -19.73015,-27.64767 -0.0405,-8.47044 2.45519,-15.09499 7.77931,-20.64927 14.76548,-15.4038 39.87537,-11.50094 48.75977,7.5788 7.7005,16.53727 -0.9274,36.08511 -18.2018,41.23878 -7.65886,2.28498 -10.73282,2.19896 -18.60713,-0.52064 z"
+   id="path4664" /><path
+   style="fill:url(#radialGradient4654);fill-opacity:1;stroke-width:1.20936334"
+   d="m 259.07812,360.23187 c -2.76453,-0.48737 -7.11823,-1.95868 -9.6749,-3.26957 -16.90031,-8.66541 -20.31807,-33.3237 -6.52993,-47.11183 5.52441,-5.5244 10.22156,-7.67278 18.39619,-8.41404 18.41005,-1.66937 33.63973,13.36629 32.16375,31.75398 -0.67162,8.36694 -2.80945,13.19288 -8.25786,18.6413 -6.98389,6.98389 -16.59066,10.07611 -26.09725,8.40016 z"
+   id="path4662" /><path
+   style="fill:url(#radialGradient4654);fill-opacity:1;stroke-width:1.20936334"
+   d="m 982.2774,360.23187 c -2.76454,-0.48737 -7.11825,-1.95868 -9.67491,-3.26957 -16.90031,-8.66541 -20.31806,-33.3237 -6.52993,-47.11183 5.5244,-5.5244 10.22156,-7.67278 18.3962,-8.41404 18.41004,-1.66937 33.63974,13.36629 32.16374,31.75398 -0.6716,8.36694 -2.8094,13.19288 -8.2578,18.6413 -6.9839,6.98389 -16.59071,10.07611 -26.0973,8.40016 z"
+   id="path4660" /><path
+   style="fill:url(#radialGradient4654);fill-opacity:1;stroke-width:1.20936334"
+   d="m 620.67777,144.9652 c -2.76453,-0.48737 -7.11824,-1.95868 -9.67491,-3.26957 -16.90031,-8.66541 -20.31807,-33.3237 -6.52994,-47.11183 5.52441,-5.52441 10.22157,-7.6728 18.3962,-8.41404 18.41005,-1.66937 33.63973,13.36628 32.16375,31.75398 -0.67161,8.36694 -2.80945,13.19288 -8.25786,18.6413 -6.98388,6.98389 -16.59066,10.07611 -26.09724,8.40016 z"
+   id="path4616" /><g
+   id="g4646"
+   transform="matrix(2.5779309,0,0,2.5779309,481.32066,2207.8469)"><path
+     id="path9"
+     d="m 105.84704,-690.34808 c 0.077,1.313 -1.786,0.968 -1.786,2.293 0,38.551 -44.72,96.831 -89.847,108.19 v 1.182 c 59.957,-5.51 126.73,-66.8 128.24,-125.85 l -36.6,14.189 z"
+     style="fill:url(#linearGradient4656)"
+     inkscape:connector-curvature="0" /><path
+     id="path11"
+     d="m 89.78704,-686.57708 c 0.077,1.313 0.121,2.633 0.121,3.958 0,38.551 -30.7,90.497 -75.827,101.86 v 1.637 c 59.065,-3.823 105.81,-63.023 105.81,-109.2 0,-2.375 -0.125,-4.729 -0.371,-7.056 l -29.73,8.796 z"
+     inkscape:connector-curvature="0"
+     style="fill:#e96d1f" /><path
+     id="path20"
+     d="m 119.89704,-695.77708 -7.833,2.625 c 0.041,0.963 0.191,2.203 0.191,3.173 0,41.219 -37.272,98.205 -87.274,107.12 -3.243,1.089 -7.538,2.077 -10.93,2.932 v 1.639 c 68.344,-8.66 111.18,-71.719 105.84,-117.49 z"
+     inkscape:connector-curvature="0"
+     style="fill:#d9541e" /><path
+     id="path22"
+     d="m 6.15604,-690.34808 c -0.077,1.313 1.787,0.968 1.787,2.293 0,38.551 46.558,97.366 91.688,108.73 v 1.639 c -59.953,-5.52 -128.56,-67.8 -130.07,-126.85 l 36.599,14.189 z"
+     style="fill:url(#SVGID_2_)"
+     inkscape:connector-curvature="0" /><path
+     id="path24"
+     d="m 24.28804,-685.60508 c -0.076,1.313 -0.12,2.63 -0.12,3.957 0,38.551 30.699,90.497 75.827,101.86 v 1.639 c -59.044,-2.79 -105.81,-63.024 -105.81,-109.2 0,-2.375 0.128,-4.729 0.371,-7.056 l 29.73,8.798 z"
+     inkscape:connector-curvature="0"
+     style="fill:#a0ce67" /><path
+     id="path26"
+     d="m -5.72996,-694.95408 7.617,2.722 c -0.041,0.962 -0.066,2.254 -0.066,3.225 0,41.219 37.271,98.204 87.272,107.12 3.245,1.088 7.538,2.077 10.932,2.931 v 1.638 c -65.254,-5.56 -111.1,-71.866 -105.76,-117.64 z"
+     inkscape:connector-curvature="0"
+     style="fill:#439639" /></g></svg>

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()
+    }
+}