1 files changed, 20 insertions, 1 deletions
diff --git a/src/jabber.rs b/src/jabber.rs
index cf90f73..30dc15d 100644
--- a/src/jabber.rs
+++ b/src/jabber.rs
@@ -1,8 +1,10 @@
+use std::pin::pin;
 use std::str::{self, FromStr};
 use std::sync::Arc;
 
 use async_recursion::async_recursion;
-use peanuts::element::IntoElement;
+use futures::StreamExt;
+use peanuts::element::{FromContent, IntoElement};
 use peanuts::{Reader, Writer};
 use rsasl::prelude::{Mechname, SASLClient, SASLConfig};
 use tokio::io::{AsyncRead, AsyncWrite, ReadHalf, WriteHalf};
@@ -13,6 +15,7 @@ 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};
@@ -26,6 +29,22 @@ pub struct JabberStream<S> {
     writer: Writer<WriteHalf<S>>,
 }
 
+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>
 where
     S: AsyncRead + AsyncWrite + Unpin + Send + std::fmt::Debug,