Cargo.toml

@@ -19,7 +19,6 @@ tracing = "0.1.40"
 trust-dns-resolver = "0.22.0"
 try_map = "0.3.1"
 peanuts = { version = "0.1.0", path = "../peanuts" }
-futures = "0.3.31"
 
 [dev-dependencies]
 test-log = { version = "0.2", features = ["trace"] }

src/client.rs

@@ -1,266 +0,0 @@
-use std::{pin::pin, sync::Arc, task::Poll};
-
-use futures::{Sink, Stream, StreamExt};
-use rsasl::config::SASLConfig;
-
-use crate::{
-    connection::{Tls, Unencrypted},
-    jid::ParseError,
-    stanza::{
-        client::Stanza,
-        sasl::Mechanisms,
-        stream::{Feature, Features},
-    },
-    Connection, Error, JabberStream, Result, JID,
-};
-
-// feed it client stanzas, receive client stanzas
-pub struct JabberClient {
-    connection: ConnectionState,
-    jid: JID,
-    password: Arc<SASLConfig>,
-    server: String,
-}
-
-impl JabberClient {
-    pub fn new(
-        jid: impl TryInto<JID, Error = ParseError>,
-        password: impl ToString,
-    ) -> Result<JabberClient> {
-        let jid = jid.try_into()?;
-        let sasl_config = SASLConfig::with_credentials(
-            None,
-            jid.localpart.clone().ok_or(Error::NoLocalpart)?,
-            password.to_string(),
-        )?;
-        Ok(JabberClient {
-            connection: ConnectionState::Disconnected,
-            jid: jid.clone(),
-            password: sasl_config,
-            server: jid.domainpart,
-        })
-    }
-
-    pub async fn connect(&mut self) -> Result<()> {
-        match &self.connection {
-            ConnectionState::Disconnected => {
-                self.connection = ConnectionState::Disconnected
-                    .connect(&mut self.jid, self.password.clone(), &mut self.server)
-                    .await?;
-                Ok(())
-            }
-            ConnectionState::Connecting(_connecting) => Err(Error::AlreadyConnecting),
-            ConnectionState::Connected(_jabber_stream) => Ok(()),
-        }
-    }
-}
-
-impl Stream for JabberClient {
-    type Item = Result<Stanza>;
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Option<Self::Item>> {
-        let mut client = pin!(self);
-        match &mut client.connection {
-            ConnectionState::Disconnected => Poll::Pending,
-            ConnectionState::Connecting(_connecting) => Poll::Pending,
-            ConnectionState::Connected(jabber_stream) => jabber_stream.poll_next_unpin(cx),
-        }
-    }
-}
-
-pub enum ConnectionState {
-    Disconnected,
-    Connecting(Connecting),
-    Connected(JabberStream<Tls>),
-}
-
-impl ConnectionState {
-    pub async fn connect(
-        mut self,
-        jid: &mut JID,
-        auth: Arc<SASLConfig>,
-        server: &mut String,
-    ) -> Result<Self> {
-        loop {
-            match self {
-                ConnectionState::Disconnected => {
-                    self = ConnectionState::Connecting(Connecting::start(&server).await?);
-                }
-                ConnectionState::Connecting(connecting) => match connecting {
-                    Connecting::InsecureConnectionEstablised(tcp_stream) => {
-                        self = ConnectionState::Connecting(Connecting::InsecureStreamStarted(
-                            JabberStream::start_stream(tcp_stream, server).await?,
-                        ))
-                    }
-                    Connecting::InsecureStreamStarted(jabber_stream) => {
-                        self = ConnectionState::Connecting(Connecting::InsecureGotFeatures(
-                            jabber_stream.get_features().await?,
-                        ))
-                    }
-                    Connecting::InsecureGotFeatures((features, jabber_stream)) => {
-                        match features.negotiate()? {
-                            Feature::StartTls(_start_tls) => {
-                                self =
-                                    ConnectionState::Connecting(Connecting::StartTls(jabber_stream))
-                            }
-                            // TODO: better error
-                            _ => return Err(Error::TlsRequired),
-                        }
-                    }
-                    Connecting::StartTls(jabber_stream) => {
-                        self = ConnectionState::Connecting(Connecting::ConnectionEstablished(
-                            jabber_stream.starttls(&server).await?,
-                        ))
-                    }
-                    Connecting::ConnectionEstablished(tls_stream) => {
-                        self = ConnectionState::Connecting(Connecting::StreamStarted(
-                            JabberStream::start_stream(tls_stream, server).await?,
-                        ))
-                    }
-                    Connecting::StreamStarted(jabber_stream) => {
-                        self = ConnectionState::Connecting(Connecting::GotFeatures(
-                            jabber_stream.get_features().await?,
-                        ))
-                    }
-                    Connecting::GotFeatures((features, jabber_stream)) => {
-                        match features.negotiate()? {
-                            Feature::StartTls(_start_tls) => return Err(Error::AlreadyTls),
-                            Feature::Sasl(mechanisms) => {
-                                self = ConnectionState::Connecting(Connecting::Sasl(
-                                    mechanisms,
-                                    jabber_stream,
-                                ))
-                            }
-                            Feature::Bind => {
-                                self = ConnectionState::Connecting(Connecting::Bind(jabber_stream))
-                            }
-                            Feature::Unknown => return Err(Error::Unsupported),
-                        }
-                    }
-                    Connecting::Sasl(mechanisms, jabber_stream) => {
-                        self = ConnectionState::Connecting(Connecting::ConnectionEstablished(
-                            jabber_stream.sasl(mechanisms, auth.clone()).await?,
-                        ))
-                    }
-                    Connecting::Bind(jabber_stream) => {
-                        self = ConnectionState::Connected(jabber_stream.bind(jid).await?)
-                    }
-                },
-                connected => return Ok(connected),
-            }
-        }
-    }
-}
-
-pub enum Connecting {
-    InsecureConnectionEstablised(Unencrypted),
-    InsecureStreamStarted(JabberStream<Unencrypted>),
-    InsecureGotFeatures((Features, JabberStream<Unencrypted>)),
-    StartTls(JabberStream<Unencrypted>),
-    ConnectionEstablished(Tls),
-    StreamStarted(JabberStream<Tls>),
-    GotFeatures((Features, JabberStream<Tls>)),
-    Sasl(Mechanisms, JabberStream<Tls>),
-    Bind(JabberStream<Tls>),
-}
-
-impl Connecting {
-    pub async fn start(server: &str) -> Result<Self> {
-        match Connection::connect(server).await? {
-            Connection::Encrypted(tls_stream) => Ok(Connecting::ConnectionEstablished(tls_stream)),
-            Connection::Unencrypted(tcp_stream) => {
-                Ok(Connecting::InsecureConnectionEstablised(tcp_stream))
-            }
-        }
-    }
-}
-
-impl Features {
-    pub fn negotiate(self) -> Result<Feature> {
-        if let Some(Feature::StartTls(s)) = self
-            .features
-            .iter()
-            .find(|feature| matches!(feature, Feature::StartTls(_s)))
-        {
-            // TODO: avoid clone
-            return Ok(Feature::StartTls(s.clone()));
-        } else if let Some(Feature::Sasl(mechanisms)) = self
-            .features
-            .iter()
-            .find(|feature| matches!(feature, Feature::Sasl(_)))
-        {
-            // TODO: avoid clone
-            return Ok(Feature::Sasl(mechanisms.clone()));
-        } else if let Some(Feature::Bind) = self
-            .features
-            .into_iter()
-            .find(|feature| matches!(feature, Feature::Bind))
-        {
-            Ok(Feature::Bind)
-        } else {
-            // TODO: better error
-            return Err(Error::Negotiation);
-        }
-    }
-}
-
-pub enum InsecureConnecting {
-    Disconnected,
-    ConnectionEstablished(Connection),
-    PreStarttls(JabberStream<Unencrypted>),
-    PreAuthenticated(JabberStream<Tls>),
-    Authenticated(Tls),
-    PreBound(JabberStream<Tls>),
-    Bound(JabberStream<Tls>),
-}
-
-impl Sink<Stanza> for JabberClient {
-    type Error = Error;
-
-    fn poll_ready(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<std::result::Result<(), Self::Error>> {
-        todo!()
-    }
-
-    fn start_send(
-        self: std::pin::Pin<&mut Self>,
-        item: Stanza,
-    ) -> std::result::Result<(), Self::Error> {
-        todo!()
-    }
-
-    fn poll_flush(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<std::result::Result<(), Self::Error>> {
-        todo!()
-    }
-
-    fn poll_close(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<std::result::Result<(), Self::Error>> {
-        todo!()
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::time::Duration;
-
-    use super::JabberClient;
-    use test_log::test;
-    use tokio::time::sleep;
-
-    #[test(tokio::test)]
-    async fn login() {
-        let mut client = JabberClient::new("test@blos.sm", "slayed").unwrap();
-        client.connect().await.unwrap();
-        sleep(Duration::from_secs(5)).await
-    }
-}

src/connection.rs

@@ -10,6 +10,7 @@ use tokio_native_tls::native_tls::TlsConnector;
 use tokio_native_tls::TlsStream;
 use tracing::{debug, info, instrument, trace};
 
+use crate::Jabber;
 use crate::Result;
 use crate::{Error, JID};
 
@@ -18,51 +19,69 @@ pub type Unencrypted = TcpStream;
 
 #[derive(Debug)]
 pub enum Connection {
-    Encrypted(Tls),
-    Unencrypted(Unencrypted),
+    Encrypted(Jabber<Tls>),
+    Unencrypted(Jabber<Unencrypted>),
 }
 
 impl Connection {
-    // #[instrument]
+    #[instrument]
     /// stream not started
-    // pub async fn ensure_tls(self) -> Result<Jabber<Tls>> {
-    //     match self {
-    //         Connection::Encrypted(j) => Ok(j),
-    //         Connection::Unencrypted(mut j) => {
-    //             j.start_stream().await?;
-    //             info!("upgrading connection to tls");
-    //             j.get_features().await?;
-    //             let j = j.starttls().await?;
-    //             Ok(j)
-    //         }
-    //     }
-    // }
+    pub async fn ensure_tls(self) -> Result<Jabber<Tls>> {
+        match self {
+            Connection::Encrypted(j) => Ok(j),
+            Connection::Unencrypted(mut j) => {
+                j.start_stream().await?;
+                info!("upgrading connection to tls");
+                j.get_features().await?;
+                let j = j.starttls().await?;
+                Ok(j)
+            }
+        }
+    }
 
-    pub async fn connect_user(jid: impl AsRef<str>) -> Result<Self> {
+    pub async fn connect_user(jid: impl AsRef<str>, password: String) -> Result<Self> {
         let jid: JID = JID::from_str(jid.as_ref())?;
         let server = jid.domainpart.clone();
-        Self::connect(&server).await
+        let auth = SASLConfig::with_credentials(None, jid.localpart.clone().unwrap(), password)?;
+        println!("auth: {:?}", auth);
+        Self::connect(&server, Some(jid), Some(auth)).await
     }
 
     #[instrument]
-    pub async fn connect(server: impl AsRef<str> + std::fmt::Debug) -> Result<Self> {
-        info!("connecting to {}", server.as_ref());
-        let sockets = Self::get_sockets(server.as_ref()).await;
+    pub async fn connect(
+        server: &str,
+        jid: Option<JID>,
+        auth: Option<Arc<SASLConfig>>,
+    ) -> Result<Self> {
+        info!("connecting to {}", server);
+        let sockets = Self::get_sockets(&server).await;
         debug!("discovered sockets: {:?}", sockets);
         for (socket_addr, tls) in sockets {
             match tls {
                 true => {
-                    if let Ok(connection) = Self::connect_tls(socket_addr, server.as_ref()).await {
+                    if let Ok(connection) = Self::connect_tls(socket_addr, &server).await {
                         info!("connected via encrypted stream to {}", socket_addr);
-                        // let (readhalf, writehalf) = tokio::io::split(connection);
-                        return Ok(Self::Encrypted(connection));
+                        let (readhalf, writehalf) = tokio::io::split(connection);
+                        return Ok(Self::Encrypted(Jabber::new(
+                            readhalf,
+                            writehalf,
+                            jid,
+                            auth,
+                            server.to_owned(),
+                        )));
                     }
                 }
                 false => {
                     if let Ok(connection) = Self::connect_unencrypted(socket_addr).await {
                         info!("connected via unencrypted stream to {}", socket_addr);
-                        // let (readhalf, writehalf) = tokio::io::split(connection);
-                        return Ok(Self::Unencrypted(connection));
+                        let (readhalf, writehalf) = tokio::io::split(connection);
+                        return Ok(Self::Unencrypted(Jabber::new(
+                            readhalf,
+                            writehalf,
+                            jid,
+                            auth,
+                            server.to_owned(),
+                        )));
                     }
                 }
             }
@@ -169,16 +188,16 @@ mod tests {
 
     #[test(tokio::test)]
     async fn connect() {
-        Connection::connect("blos.sm").await.unwrap();
+        Connection::connect("blos.sm", None, None).await.unwrap();
     }
 
-    // #[test(tokio::test)]
-    // async fn test_tls() {
-    //     Connection::connect("blos.sm", None, None)
-    //         .await
-    //         .unwrap()
-    //         .ensure_tls()
-    //         .await
-    //         .unwrap();
-    // }
+    #[test(tokio::test)]
+    async fn test_tls() {
+        Connection::connect("blos.sm", None, None)
+            .await
+            .unwrap()
+            .ensure_tls()
+            .await
+            .unwrap();
+    }
 }

src/error.rs

@@ -8,16 +8,24 @@ use crate::{jid::ParseError, stanza::sasl::Failure};
 #[derive(Debug)]
 pub enum Error {
     Connection,
+    BadStream,
+    StartTlsUnavailable,
+    TlsNegotiation,
     Utf8Decode,
+    NoFeatures,
+    UnknownNamespace,
+    UnknownAttribute,
+    NoID,
+    NoType,
+    IDMismatch,
+    BindError,
+    ParseError,
     Negotiation,
     TlsRequired,
-    AlreadyTls,
-    Unsupported,
-    NoLocalpart,
-    AlreadyConnecting,
+    UnexpectedEnd,
     UnexpectedElement(peanuts::Element),
+    UnexpectedText,
     XML(peanuts::Error),
-    Deserialization(peanuts::DeserializeError),
     SASL(SASLError),
     JID(ParseError),
     Authentication(Failure),
@@ -29,6 +37,8 @@ pub enum Error {
 pub enum SASLError {
     SASL(rsasl::prelude::SASLError),
     MechanismName(MechanismNameError),
+    NoChallenge,
+    NoSuccess,
 }
 
 impl From<rsasl::prelude::SASLError> for Error {
@@ -37,12 +47,6 @@ impl From<rsasl::prelude::SASLError> for Error {
     }
 }
 
-impl From<peanuts::DeserializeError> for Error {
-    fn from(e: peanuts::DeserializeError) -> Self {
-        Error::Deserialization(e)
-    }
-}
-
 impl From<MechanismNameError> for Error {
     fn from(e: MechanismNameError) -> Self {
         Self::SASL(SASLError::MechanismName(e))

src/jabber.rs

@@ -1,10 +1,8 @@
-use std::pin::pin;
-use std::str::{self, FromStr};
+use std::str;
 use std::sync::Arc;
 
 use async_recursion::async_recursion;
-use futures::StreamExt;
-use peanuts::element::{FromContent, IntoElement};
+use peanuts::element::IntoElement;
 use peanuts::{Reader, Writer};
 use rsasl::prelude::{Mechname, SASLClient, SASLConfig};
 use tokio::io::{AsyncRead, AsyncWrite, ReadHalf, WriteHalf};
@@ -15,7 +13,6 @@ use crate::connection::{Tls, Unencrypted};
 use crate::error::Error;
 use crate::stanza::bind::{Bind, BindType, FullJidType, ResourceType};
 use crate::stanza::client::iq::{Iq, IqType, Query};
-use crate::stanza::client::Stanza;
 use crate::stanza::sasl::{Auth, Challenge, Mechanisms, Response, ServerResponse};
 use crate::stanza::starttls::{Proceed, StartTls};
 use crate::stanza::stream::{Feature, Features, Stream};
@@ -23,34 +20,47 @@ use crate::stanza::XML_VERSION;
 use crate::JID;
 use crate::{Connection, Result};
 
-// open stream (streams started)
-pub struct JabberStream<S> {
+pub struct Jabber<S> {
     reader: Reader<ReadHalf<S>>,
     writer: Writer<WriteHalf<S>>,
+    jid: Option<JID>,
+    auth: Option<Arc<SASLConfig>>,
+    server: String,
 }
 
-impl<S: AsyncRead> futures::Stream for JabberStream<S> {
-    type Item = Result<Stanza>;
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Option<Self::Item>> {
-        pin!(self).reader.poll_next_unpin(cx).map(|content| {
-            content.map(|content| -> Result<Stanza> {
-                let stanza = content.map(|content| Stanza::from_content(content))?;
-                Ok(stanza?)
-            })
-        })
-    }
-}
-
-impl<S> JabberStream<S>
+impl<S> Jabber<S>
 where
-    S: AsyncRead + AsyncWrite + Unpin + Send + std::fmt::Debug,
-    JabberStream<S>: std::fmt::Debug,
+    S: AsyncRead + AsyncWrite + Unpin,
 {
-    pub async fn sasl(mut self, mechanisms: Mechanisms, sasl_config: Arc<SASLConfig>) -> Result<S> {
+    pub fn new(
+        reader: ReadHalf<S>,
+        writer: WriteHalf<S>,
+        jid: Option<JID>,
+        auth: Option<Arc<SASLConfig>>,
+        server: String,
+    ) -> Self {
+        let reader = Reader::new(reader);
+        let writer = Writer::new(writer);
+        Self {
+            reader,
+            writer,
+            jid,
+            auth,
+            server,
+        }
+    }
+}
+
+impl<S> Jabber<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin + Send,
+    Jabber<S>: std::fmt::Debug,
+{
+    pub async fn sasl(
+        &mut self,
+        mechanisms: Mechanisms,
+        sasl_config: Arc<SASLConfig>,
+    ) -> Result<()> {
         let sasl = SASLClient::new(sasl_config);
         let mut offered_mechs: Vec<&Mechname> = Vec::new();
         for mechanism in &mechanisms.mechanisms {
@@ -133,15 +143,12 @@ where
                 }
             }
         }
-        let writer = self.writer.into_inner();
-        let reader = self.reader.into_inner();
-        let stream = reader.unsplit(writer);
-        Ok(stream)
+        Ok(())
     }
 
-    pub async fn bind(mut self, jid: &mut JID) -> Result<Self> {
+    pub async fn bind(&mut self) -> Result<()> {
         let iq_id = nanoid::nanoid!();
-        if let Some(resource) = &jid.resourcepart {
+        if let Some(resource) = self.jid.clone().unwrap().resourcepart {
             let iq = Iq {
                 from: None,
                 id: iq_id.clone(),
@@ -149,7 +156,7 @@ where
                 r#type: IqType::Set,
                 lang: None,
                 query: Some(Query::Bind(Bind {
-                    r#type: Some(BindType::Resource(ResourceType(resource.to_string()))),
+                    r#type: Some(BindType::Resource(ResourceType(resource))),
                 })),
                 errors: Vec::new(),
             };
@@ -164,12 +171,12 @@ where
                     lang: _,
                     query:
                         Some(Query::Bind(Bind {
-                            r#type: Some(BindType::Jid(FullJidType(new_jid))),
+                            r#type: Some(BindType::Jid(FullJidType(jid))),
                         })),
                     errors: _,
                 } if id == iq_id => {
-                    *jid = new_jid;
-                    return Ok(self);
+                    self.jid = Some(jid);
+                    return Ok(());
                 }
                 Iq {
                     from: _,
@@ -207,12 +214,12 @@ where
                     lang: _,
                     query:
                         Some(Query::Bind(Bind {
-                            r#type: Some(BindType::Jid(FullJidType(new_jid))),
+                            r#type: Some(BindType::Jid(FullJidType(jid))),
                         })),
                     errors: _,
                 } if id == iq_id => {
-                    *jid = new_jid;
-                    return Ok(self);
+                    self.jid = Some(jid);
+                    return Ok(());
                 }
                 Iq {
                     from: _,
@@ -233,44 +240,39 @@ where
     }
 
     #[instrument]
-    pub async fn start_stream(connection: S, server: &mut String) -> Result<Self> {
+    pub async fn start_stream(&mut self) -> Result<()> {
         // client to server
-        let (reader, writer) = tokio::io::split(connection);
-        let mut reader = Reader::new(reader);
-        let mut writer = Writer::new(writer);
 
         // declaration
-        writer.write_declaration(XML_VERSION).await?;
+        self.writer.write_declaration(XML_VERSION).await?;
 
         // opening stream element
-        let stream = Stream::new_client(
-            None,
-            JID::from_str(server.as_ref())?,
-            None,
-            "en".to_string(),
-        );
-        writer.write_start(&stream).await?;
+        let server = self.server.clone().try_into()?;
+        let stream = Stream::new_client(None, server, None, "en".to_string());
+        self.writer.write_start(&stream).await?;
 
         // server to client
 
         // may or may not send a declaration
-        let _decl = reader.read_prolog().await?;
+        let _decl = self.reader.read_prolog().await?;
 
         // receive stream element and validate
-        let stream: Stream = reader.read_start().await?;
+        let text = str::from_utf8(self.reader.buffer.data()).unwrap();
+        debug!("data: {}", text);
+        let stream: Stream = self.reader.read_start().await?;
         debug!("got stream: {:?}", stream);
         if let Some(from) = stream.from {
-            *server = from.to_string();
+            self.server = from.to_string()
         }
 
-        Ok(Self { reader, writer })
+        Ok(())
     }
 
-    pub async fn get_features(mut self) -> Result<(Features, Self)> {
+    pub async fn get_features(&mut self) -> Result<Features> {
         debug!("getting features");
         let features: Features = self.reader.read().await?;
         debug!("got features: {:?}", features);
-        Ok((features, self))
+        Ok(features)
     }
 
     pub fn into_inner(self) -> S {
@@ -278,89 +280,89 @@ where
     }
 }
 
-impl JabberStream<Unencrypted> {
-    // pub async fn negotiate<S: AsyncRead + AsyncWrite + Unpin>(
-    //     mut self,
-    //     features: Features,
-    // ) -> Result<Feature> {
-    //     // TODO: timeout
-    //     if let Some(Feature::StartTls(_)) = features
-    //         .features
-    //         .iter()
-    //         .find(|feature| matches!(feature, Feature::StartTls(_s)))
-    //     {
-    //         return Ok(self);
-    //     } else {
-    //         // TODO: better error
-    //         return Err(Error::TlsRequired);
-    //     }
-    // }
-
-    // #[async_recursion]
-    // pub async fn negotiate_tls_optional(mut self) -> Result<Connection> {
-    //     self.start_stream().await?;
-    //     // TODO: timeout
-    //     let features = self.get_features().await?.features;
-    //     if let Some(Feature::StartTls(_)) = features
-    //         .iter()
-    //         .find(|feature| matches!(feature, Feature::StartTls(_s)))
-    //     {
-    //         let jabber = self.starttls().await?;
-    //         let jabber = jabber.negotiate().await?;
-    //         return Ok(Connection::Encrypted(jabber));
-    //     } else if let (Some(sasl_config), Some(Feature::Sasl(mechanisms))) = (
-    //         self.auth.clone(),
-    //         features
-    //             .iter()
-    //             .find(|feature| matches!(feature, Feature::Sasl(_))),
-    //     ) {
-    //         self.sasl(mechanisms.clone(), sasl_config).await?;
-    //         let jabber = self.negotiate_tls_optional().await?;
-    //         Ok(jabber)
-    //     } else if let Some(Feature::Bind) = features
-    //         .iter()
-    //         .find(|feature| matches!(feature, Feature::Bind))
-    //     {
-    //         self.bind().await?;
-    //         Ok(Connection::Unencrypted(self))
-    //     } else {
-    //         // TODO: better error
-    //         return Err(Error::Negotiation);
-    //     }
-    // }
+impl Jabber<Unencrypted> {
+    pub async fn negotiate<S: AsyncRead + AsyncWrite + Unpin>(mut self) -> Result<Jabber<Tls>> {
+        self.start_stream().await?;
+        // TODO: timeout
+        let features = self.get_features().await?.features;
+        if let Some(Feature::StartTls(_)) = features
+            .iter()
+            .find(|feature| matches!(feature, Feature::StartTls(_s)))
+        {
+            let jabber = self.starttls().await?;
+            let jabber = jabber.negotiate().await?;
+            return Ok(jabber);
+        } else {
+            // TODO: better error
+            return Err(Error::TlsRequired);
+        }
     }
 
-impl JabberStream<Tls> {
-    // #[async_recursion]
-    // pub async fn negotiate(mut self) -> Result<JabberStream<Tls>> {
-    //     self.start_stream().await?;
-    //     let features = self.get_features().await?.features;
-
-    //     if let (Some(sasl_config), Some(Feature::Sasl(mechanisms))) = (
-    //         self.auth.clone(),
-    //         features
-    //             .iter()
-    //             .find(|feature| matches!(feature, Feature::Sasl(_))),
-    //     ) {
-    //         // TODO: avoid clone
-    //         self.sasl(mechanisms.clone(), sasl_config).await?;
-    //         let jabber = self.negotiate().await?;
-    //         Ok(jabber)
-    //     } else if let Some(Feature::Bind) = features
-    //         .iter()
-    //         .find(|feature| matches!(feature, Feature::Bind))
-    //     {
-    //         self.bind().await?;
-    //         Ok(self)
-    //     } else {
-    //         // TODO: better error
-    //         return Err(Error::Negotiation);
-    //     }
-    // }
+    #[async_recursion]
+    pub async fn negotiate_tls_optional(mut self) -> Result<Connection> {
+        self.start_stream().await?;
+        // TODO: timeout
+        let features = self.get_features().await?.features;
+        if let Some(Feature::StartTls(_)) = features
+            .iter()
+            .find(|feature| matches!(feature, Feature::StartTls(_s)))
+        {
+            let jabber = self.starttls().await?;
+            let jabber = jabber.negotiate().await?;
+            return Ok(Connection::Encrypted(jabber));
+        } else if let (Some(sasl_config), Some(Feature::Sasl(mechanisms))) = (
+            self.auth.clone(),
+            features
+                .iter()
+                .find(|feature| matches!(feature, Feature::Sasl(_))),
+        ) {
+            self.sasl(mechanisms.clone(), sasl_config).await?;
+            let jabber = self.negotiate_tls_optional().await?;
+            Ok(jabber)
+        } else if let Some(Feature::Bind) = features
+            .iter()
+            .find(|feature| matches!(feature, Feature::Bind))
+        {
+            self.bind().await?;
+            Ok(Connection::Unencrypted(self))
+        } else {
+            // TODO: better error
+            return Err(Error::Negotiation);
+        }
+    }
 }
 
-impl JabberStream<Unencrypted> {
-    pub async fn starttls(mut self, domain: impl AsRef<str>) -> Result<Tls> {
+impl Jabber<Tls> {
+    #[async_recursion]
+    pub async fn negotiate(mut self) -> Result<Jabber<Tls>> {
+        self.start_stream().await?;
+        let features = self.get_features().await?.features;
+
+        if let (Some(sasl_config), Some(Feature::Sasl(mechanisms))) = (
+            self.auth.clone(),
+            features
+                .iter()
+                .find(|feature| matches!(feature, Feature::Sasl(_))),
+        ) {
+            // TODO: avoid clone
+            self.sasl(mechanisms.clone(), sasl_config).await?;
+            let jabber = self.negotiate().await?;
+            Ok(jabber)
+        } else if let Some(Feature::Bind) = features
+            .iter()
+            .find(|feature| matches!(feature, Feature::Bind))
+        {
+            self.bind().await?;
+            Ok(self)
+        } else {
+            // TODO: better error
+            return Err(Error::Negotiation);
+        }
+    }
+}
+
+impl Jabber<Unencrypted> {
+    pub async fn starttls(mut self) -> Result<Jabber<Tls>> {
         self.writer
             .write_full(&StartTls { required: false })
             .await?;
@@ -368,31 +370,43 @@ impl JabberStream<Unencrypted> {
         debug!("got proceed: {:?}", proceed);
         let connector = TlsConnector::new().unwrap();
         let stream = self.reader.into_inner().unsplit(self.writer.into_inner());
-        if let Ok(tls_stream) = tokio_native_tls::TlsConnector::from(connector)
-            .connect(domain.as_ref(), stream)
+        if let Ok(tlsstream) = tokio_native_tls::TlsConnector::from(connector)
+            .connect(&self.server, stream)
             .await
         {
-            // let (read, write) = tokio::io::split(tlsstream);
-            // let client = JabberStream::new(read, write);
-            return Ok(tls_stream);
+            let (read, write) = tokio::io::split(tlsstream);
+            let client = Jabber::new(
+                read,
+                write,
+                self.jid.to_owned(),
+                self.auth.to_owned(),
+                self.server.to_owned(),
+            );
+            return Ok(client);
         } else {
             return Err(Error::Connection);
         }
     }
 }
 
-impl std::fmt::Debug for JabberStream<Tls> {
+impl std::fmt::Debug for Jabber<Tls> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Jabber")
             .field("connection", &"tls")
+            .field("jid", &self.jid)
+            .field("auth", &self.auth)
+            .field("server", &self.server)
             .finish()
     }
 }
 
-impl std::fmt::Debug for JabberStream<Unencrypted> {
+impl std::fmt::Debug for Jabber<Unencrypted> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Jabber")
             .field("connection", &"unencrypted")
+            .field("jid", &self.jid)
+            .field("auth", &self.auth)
+            .field("server", &self.server)
             .finish()
     }
 }
@@ -408,61 +422,61 @@ mod tests {
 
     #[test(tokio::test)]
     async fn start_stream() {
-        // let connection = Connection::connect("blos.sm", None, None).await.unwrap();
-        // match connection {
-        //     Connection::Encrypted(mut c) => c.start_stream().await.unwrap(),
-        //     Connection::Unencrypted(mut c) => c.start_stream().await.unwrap(),
-        // }
+        let connection = Connection::connect("blos.sm", None, None).await.unwrap();
+        match connection {
+            Connection::Encrypted(mut c) => c.start_stream().await.unwrap(),
+            Connection::Unencrypted(mut c) => c.start_stream().await.unwrap(),
+        }
     }
 
     #[test(tokio::test)]
     async fn sasl() {
-        // let mut jabber = Connection::connect_user("test@blos.sm", "slayed".to_string())
-        //     .await
-        //     .unwrap()
-        //     .ensure_tls()
-        //     .await
-        //     .unwrap();
-        // let text = str::from_utf8(jabber.reader.buffer.data()).unwrap();
-        // println!("data: {}", text);
-        // jabber.start_stream().await.unwrap();
+        let mut jabber = Connection::connect_user("test@blos.sm", "slayed".to_string())
+            .await
+            .unwrap()
+            .ensure_tls()
+            .await
+            .unwrap();
+        let text = str::from_utf8(jabber.reader.buffer.data()).unwrap();
+        println!("data: {}", text);
+        jabber.start_stream().await.unwrap();
 
-        // let text = str::from_utf8(jabber.reader.buffer.data()).unwrap();
-        // println!("data: {}", text);
-        // jabber.reader.read_buf().await.unwrap();
-        // let text = str::from_utf8(jabber.reader.buffer.data()).unwrap();
-        // println!("data: {}", text);
+        let text = str::from_utf8(jabber.reader.buffer.data()).unwrap();
+        println!("data: {}", text);
+        jabber.reader.read_buf().await.unwrap();
+        let text = str::from_utf8(jabber.reader.buffer.data()).unwrap();
+        println!("data: {}", text);
 
-        // let features = jabber.get_features().await.unwrap();
-        // let (sasl_config, feature) = (
-        //     jabber.auth.clone().unwrap(),
-        //     features
-        //         .features
-        //         .iter()
-        //         .find(|feature| matches!(feature, Feature::Sasl(_)))
-        //         .unwrap(),
-        // );
-        // match feature {
-        //     Feature::StartTls(_start_tls) => todo!(),
-        //     Feature::Sasl(mechanisms) => {
-        //         jabber.sasl(mechanisms.clone(), sasl_config).await.unwrap();
-        //     }
-        //     Feature::Bind => todo!(),
-        //     Feature::Unknown => todo!(),
-        // }
+        let features = jabber.get_features().await.unwrap();
+        let (sasl_config, feature) = (
+            jabber.auth.clone().unwrap(),
+            features
+                .features
+                .iter()
+                .find(|feature| matches!(feature, Feature::Sasl(_)))
+                .unwrap(),
+        );
+        match feature {
+            Feature::StartTls(_start_tls) => todo!(),
+            Feature::Sasl(mechanisms) => {
+                jabber.sasl(mechanisms.clone(), sasl_config).await.unwrap();
+            }
+            Feature::Bind => todo!(),
+            Feature::Unknown => todo!(),
+        }
     }
 
     #[tokio::test]
     async fn negotiate() {
-        // let _jabber = Connection::connect_user("test@blos.sm", "slayed".to_string())
-        //     .await
-        //     .unwrap()
-        //     .ensure_tls()
-        //     .await
-        //     .unwrap()
-        //     .negotiate()
-        //     .await
-        //     .unwrap();
-        // sleep(Duration::from_secs(5)).await
+        let _jabber = Connection::connect_user("test@blos.sm", "slayed".to_string())
+            .await
+            .unwrap()
+            .ensure_tls()
+            .await
+            .unwrap()
+            .negotiate()
+            .await
+            .unwrap();
+        sleep(Duration::from_secs(5)).await
     }
 }

src/lib.rs

@@ -2,7 +2,6 @@
 // #![feature(let_chains)]
 
 // TODO: logging (dropped errors)
-pub mod client;
 pub mod connection;
 pub mod error;
 pub mod jabber;
@@ -12,25 +11,24 @@ pub mod stanza;
 pub use connection::Connection;
 use connection::Tls;
 pub use error::Error;
-pub use jabber::JabberStream;
+pub use jabber::Jabber;
 pub use jid::JID;
 
 pub type Result<T> = std::result::Result<T, Error>;
 
-pub async fn login<J: AsRef<str>, P: AsRef<str>>(jid: J, password: P) -> Result<JabberStream<Tls>> {
-    todo!()
-    // Ok(Connection::connect_user(jid, password.as_ref().to_string())
-    //     .await?
-    //     .ensure_tls()
-    //     .await?
-    //     .negotiate()
-    //     .await?)
+pub async fn login<J: AsRef<str>, P: AsRef<str>>(jid: J, password: P) -> Result<Jabber<Tls>> {
+    Ok(Connection::connect_user(jid, password.as_ref().to_string())
+        .await?
+        .ensure_tls()
+        .await?
+        .negotiate()
+        .await?)
 }
 
 #[cfg(test)]
 mod tests {
-    // #[tokio::test]
-    // async fn test_login() {
-    //     crate::login("test@blos.sm/clown", "slayed").await.unwrap();
-    // }
+    #[tokio::test]
+    async fn test_login() {
+        crate::login("test@blos.sm/clown", "slayed").await.unwrap();
+    }
 }

src/stanza/client/mod.rs

@@ -1,7 +1,7 @@
 use iq::Iq;
 use message::Message;
 use peanuts::{
-    element::{Content, ContentBuilder, FromContent, FromElement, IntoContent, IntoElement},
+    element::{FromElement, IntoElement},
     DeserializeError,
 };
 use presence::Presence;
@@ -20,18 +20,6 @@ pub enum Stanza {
     Presence(Presence),
     Iq(Iq),
     Error(StreamError),
-    OtherContent(Content),
-}
-
-impl FromContent for Stanza {
-    fn from_content(content: Content) -> peanuts::element::DeserializeResult<Self> {
-        match content {
-            Content::Element(element) => Ok(Stanza::from_element(element)?),
-            Content::Text(_) => Ok(Stanza::OtherContent(content)),
-            Content::PI => Ok(Stanza::OtherContent(content)),
-            Content::Comment(_) => Ok(Stanza::OtherContent(content)),
-        }
-    }
 }
 
 impl FromElement for Stanza {
@@ -48,14 +36,13 @@ impl FromElement for Stanza {
     }
 }
 
-impl IntoContent for Stanza {
-    fn builder(&self) -> peanuts::element::ContentBuilder {
+impl IntoElement for Stanza {
+    fn builder(&self) -> peanuts::element::ElementBuilder {
         match self {
-            Stanza::Message(message) => <Message as IntoContent>::builder(message),
-            Stanza::Presence(presence) => <Presence as IntoContent>::builder(presence),
-            Stanza::Iq(iq) => <Iq as IntoContent>::builder(iq),
-            Stanza::Error(error) => <StreamError as IntoContent>::builder(error),
-            Stanza::OtherContent(_content) => ContentBuilder::Comment("other-content".to_string()),
+            Stanza::Message(message) => message.builder(),
+            Stanza::Presence(presence) => presence.builder(),
+            Stanza::Iq(iq) => iq.builder(),
+            Stanza::Error(error) => error.builder(),
         }
     }
 }