diff --git a/tokio-xmpp/ChangeLog b/tokio-xmpp/ChangeLog index 57bceae6..ff95d083 100644 --- a/tokio-xmpp/ChangeLog +++ b/tokio-xmpp/ChangeLog @@ -18,6 +18,7 @@ Version NEXT: always sending h='0' in our ``. * Changed: - Update hickory-dns to 0.26 (!671) + - Drive the XMPP client stream in the background (!631) Version 5.0.0: 2025-10-28 pep diff --git a/tokio-xmpp/src/stanzastream/mod.rs b/tokio-xmpp/src/stanzastream/mod.rs index 8eab86c5..fa8ccc21 100644 --- a/tokio-xmpp/src/stanzastream/mod.rs +++ b/tokio-xmpp/src/stanzastream/mod.rs @@ -30,9 +30,9 @@ use core::time::Duration; // meant for that stream, replacing it is racy. use futures::{SinkExt, Stream}; - +use std::sync::Arc; +use tokio::sync::Mutex; use tokio::sync::{mpsc, oneshot}; - use xmpp_parsers::{jid::Jid, stream_features::StreamFeatures}; use crate::connect::ServerConnector; @@ -258,6 +258,34 @@ impl StanzaStream { self.assert_send(queue_entry).await; token } + + /// Split the stream into the [`StanzaSender`] and [`StanzaReceiver`]. + pub fn split(self) -> (StanzaSender, StanzaReceiver) { + let stream = Arc::new(Mutex::new(self)); + + let tx = StanzaSender(stream.clone()); + let rx = StanzaReceiver(stream); + + (tx, rx) + } + + /// Reunite the [`StanzaSender`] and [`StanzaReceiver`] back into a single stream. + /// + /// # Panics + /// + /// This function panics if the `Sender` and `Receiver` don't come from + /// the same [`Stream`]. + pub fn reunite(tx: StanzaSender, rx: StanzaReceiver) -> Self { + assert!( + Arc::ptr_eq(&tx.0, &rx.0), + "Unrelated Sender and Receiver passed to reunite." + ); + + drop(tx); + + let inner = Arc::try_unwrap(rx.0).expect("Failed to unwrap Receiver Arc"); + inner.into_inner() + } } impl Stream for StanzaStream { @@ -267,3 +295,32 @@ impl Stream for StanzaStream { self.rx.poll_recv(cx) } } + +/// Send half of the [`StanzaStream`] +#[derive(Debug)] +pub struct StanzaSender(pub(super) Arc>); + +impl StanzaSender { + /// Send a stanza via the stream. + /// + /// See the documentation of [`StanzaStream::send()`]. + pub async fn send(&self, stanza: Box) -> StanzaToken { + self.0.lock().await.send(stanza).await + } +} + +/// Receive half of the [`StanzaStream`] +#[derive(Debug)] +pub struct StanzaReceiver(pub(super) Arc>); + +impl Stream for StanzaReceiver { + type Item = Event; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + let Ok(mut stream) = self.0.try_lock() else { + return Poll::Pending; + }; + + stream.rx.poll_recv(cx) + } +}