diff --git a/tokio-xmpp/ChangeLog b/tokio-xmpp/ChangeLog index c2eb377f..d9316581 100644 --- a/tokio-xmpp/ChangeLog +++ b/tokio-xmpp/ChangeLog @@ -22,6 +22,7 @@ Version NEXT: * Changed: - Update hickory-dns to 0.26 (!671) - Drive the XMPP client stream in the background (!631) + - Add option to split XMPP client into read and write half (!631) Version 5.0.0: 2025-10-28 pep diff --git a/tokio-xmpp/src/client/iq.rs b/tokio-xmpp/src/client/iq.rs index 453c481e..9de562e9 100644 --- a/tokio-xmpp/src/client/iq.rs +++ b/tokio-xmpp/src/client/iq.rs @@ -252,7 +252,7 @@ impl IqResponseSink { } /// Utility struct to track IQ responses. -#[derive(Debug)] +#[derive(Clone, Debug)] pub struct IqResponseTracker { map: Arc>, account_jid: Arc>>, diff --git a/tokio-xmpp/src/client/mod.rs b/tokio-xmpp/src/client/mod.rs index dd687e77..4974e216 100644 --- a/tokio-xmpp/src/client/mod.rs +++ b/tokio-xmpp/src/client/mod.rs @@ -4,18 +4,19 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at http://mozilla.org/MPL/2.0/. +use crate::client::{receiver::ClientReceiver, sender::ClientSender}; +use crate::connect::ServerConnector; +use crate::error::Error; +use crate::event::Event; +use crate::stanzastream::{self, StanzaStage, StanzaState, StanzaStream, StanzaToken}; +use crate::xmlstream::Timeouts; +use crate::Stanza; use std::io; - +use std::sync::Arc; +use tokio::sync::{mpsc, oneshot, Mutex}; +use tokio::task::JoinHandle; use xmpp_parsers::{jid::Jid, stream_features::StreamFeatures}; -use crate::{ - connect::ServerConnector, - error::Error, - stanzastream::{StanzaStage, StanzaState, StanzaStream, StanzaToken}, - xmlstream::Timeouts, - Stanza, -}; - #[cfg(feature = "direct-tls")] use crate::connect::DirectTlsServerConnector; #[cfg(any(feature = "direct-tls", feature = "starttls", feature = "insecure-tcp"))] @@ -27,10 +28,13 @@ use crate::connect::TcpServerConnector; mod iq; pub(crate) mod login; +pub(crate) mod receiver; +pub(crate) mod sender; mod stream; -pub use login::auth; +mod worker; pub use iq::{IqFailure, IqRequest, IqResponse, IqResponseToken}; +pub use login::auth; /// XMPP client connection and state /// @@ -41,9 +45,19 @@ pub use iq::{IqFailure, IqRequest, IqResponse, IqResponseToken}; /// used. #[derive(Debug)] pub struct Client { - stream: StanzaStream, + // Stanza receiver from the client worker + stanza_rx: mpsc::Receiver, + // Stanza sender to the StanzaStream + stream_tx: stanzastream::StanzaSender, + // Shutdown handle for the client worker + shutdown_tx: oneshot::Sender<()>, + // Client worker task + worker: JoinHandle, + // JID of the logged-in client bound_jid: Option, + // Stream features of the currently connected stream features: Option, + // Response tracker for IQs iq_response_tracker: iq::IqResponseTracker, } @@ -72,7 +86,8 @@ impl Client { /// [`send_iq`][`Self::send_iq`], which allows awaiting the response. pub async fn send_stanza(&mut self, mut stanza: Stanza) -> Result { stanza.ensure_id(); - let mut token = self.stream.send(Box::new(stanza)).await; + let mut token = self.stream_tx.send(Box::new(stanza)).await; + match token.wait_for(StanzaStage::Sent).await { // Queued < Sent, so it cannot be reached. Some(StanzaState::Queued) => unreachable!(), @@ -93,13 +108,6 @@ impl Client { /// response. See also the documentation of [`IqResponseToken`] for more /// information on the behaviour of these tokens. /// - /// **Important**: Even though IQ responses are delivered through the - /// returned token (and never through the `Stream`), the - /// [`Stream`][`futures::Stream`] - /// implementation of the [`Client`] **must be polled** to make progress - /// on the stream and to process incoming stanzas and thus to deliver them - /// to the returned token. - /// /// **Note**: If an IQ response arrives after the `token` has been /// dropped (e.g. due to a timeout), it will be delivered through the /// `Stream` like any other stanza. @@ -108,7 +116,8 @@ impl Client { // from is always None for a client None, to, req, ); - let stanza_token = self.stream.send(Box::new(iq.into())).await; + let stanza_token = self.stream_tx.send(Box::new(iq.into())).await; + token.set_stanza_token(stanza_token); token } @@ -128,9 +137,42 @@ impl Client { /// This performs an orderly stream shutdown, ensuring that all resources /// are correctly cleaned up. pub async fn send_end(self) -> Result<(), Error> { - self.stream.close().await; + self.shutdown_tx.send(()).expect("ClientWorker crashed."); + + let stream_rx = self.worker.await.unwrap(); + let stream = StanzaStream::reunite(self.stream_tx, stream_rx); + stream.close().await; + Ok(()) } + + /// Split the client into [`ClientSender`] and [`ClientReceiver`]. + pub fn split(self) -> (ClientSender, ClientReceiver) { + let client = Arc::new(Mutex::new(self)); + + let sender = ClientSender(client.clone()); + let receiver = ClientReceiver(client); + + (sender, receiver) + } + + /// Reunite a [`ClientSender`] and [`ClientReceiver`]. + /// + /// # Panics + /// + /// This functions returns an error if the [`ClientSender`] and + /// [`ClientReceiver`] don't come from the same [`Client`]. + pub fn reunite(sender: ClientSender, receiver: ClientReceiver) -> Self { + assert!( + Arc::ptr_eq(&sender.0, &receiver.0), + "Unrelated ClientSender and ClientReceiver passed to reunite." + ); + + drop(sender); + + let inner = Arc::try_unwrap(receiver.0).expect("Failed to unwrap ClientReceiver Arc"); + inner.into_inner() + } } #[cfg(feature = "direct-tls")] @@ -220,11 +262,23 @@ impl Client { connector: C, timeouts: Timeouts, ) -> Self { + let stream = StanzaStream::new_c2s(connector, jid.into(), password.into(), timeouts, 16); + let (stream_tx, stream_rx) = stream.split(); + + let iq_response_tracker = iq::IqResponseTracker::new(); + let (worker, shutdown_tx, stanza_rx) = + worker::ClientWorker::new(stream_rx, iq_response_tracker.clone(), 16); + + let worker = tokio::task::spawn(async move { worker.run().await }); + Self { - stream: StanzaStream::new_c2s(connector, jid.into(), password.into(), timeouts, 16), + stream_tx, + stanza_rx, + worker, + shutdown_tx, + iq_response_tracker, bound_jid: None, features: None, - iq_response_tracker: iq::IqResponseTracker::new(), } } } diff --git a/tokio-xmpp/src/client/receiver.rs b/tokio-xmpp/src/client/receiver.rs new file mode 100644 index 00000000..ec78adb0 --- /dev/null +++ b/tokio-xmpp/src/client/receiver.rs @@ -0,0 +1,48 @@ +// Copyright (c) 2025 xmpp-rs contributors. +// +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +use crate::Client; +use crate::Event; +use core::{pin::Pin, task::Context}; +use futures::StreamExt; +use futures::{task::Poll, Stream}; +use std::sync::Arc; +use tokio::sync::Mutex; +use xmpp_parsers::{jid::Jid, stream_features::StreamFeatures}; + +/// Read half of a [`Client`](crate::Client). +#[derive(Debug)] +pub struct ClientReceiver(pub(super) Arc>); + +impl ClientReceiver { + /// Return the bound JID. + /// + /// See the documentation of [`Client::bound_jid`](crate::Client::bound_jid) for more + /// information. + pub async fn bound_jid(&self) -> Option { + self.0.lock().await.bound_jid.clone() + } + + /// Return the received stream features. + /// + /// See the documentation of [`Client::get_stream_features`](crate::Client::get_stream_features) + /// for more information. + pub async fn get_stream_features(&self) -> Option { + self.0.lock().await.features.clone() + } +} + +impl Stream for ClientReceiver { + type Item = Event; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + let Ok(mut client) = self.0.try_lock() else { + return Poll::Pending; + }; + + client.poll_next_unpin(cx) + } +} diff --git a/tokio-xmpp/src/client/sender.rs b/tokio-xmpp/src/client/sender.rs new file mode 100644 index 00000000..97be5a89 --- /dev/null +++ b/tokio-xmpp/src/client/sender.rs @@ -0,0 +1,35 @@ +// Copyright (c) 2025 xmpp-rs contributors. +// +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +use crate::stanzastream::StanzaToken; +use crate::IqRequest; +use crate::IqResponseToken; +use crate::Stanza; +use std::io; +use std::sync::Arc; +use tokio::sync::Mutex; +use xmpp_parsers::jid::Jid; + +/// Write half of a [`Client`](crate::Client). +#[derive(Debug)] +pub struct ClientSender(pub(super) Arc>); + +impl ClientSender { + /// Send a stanza. + /// + /// See the documentation of [`Client::send_stanza`](crate::Client::send_stanza) for more + /// information. + pub async fn send_stanza(&self, stanza: Stanza) -> Result { + self.0.lock().await.send_stanza(stanza).await + } + + /// Send in iq. + /// + /// See the documentation of [`Client::send_iq`](crate::Client::send_iq) for more information. + pub async fn send_iq(&self, to: Option, req: IqRequest) -> IqResponseToken { + self.0.lock().await.send_iq(to, req).await + } +} diff --git a/tokio-xmpp/src/client/stream.rs b/tokio-xmpp/src/client/stream.rs index cd324509..4e1c4e32 100644 --- a/tokio-xmpp/src/client/stream.rs +++ b/tokio-xmpp/src/client/stream.rs @@ -4,15 +4,10 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at http://mozilla.org/MPL/2.0/. -use core::ops::ControlFlow; use core::{pin::Pin, task::Context}; use futures::{ready, task::Poll, Stream}; -use crate::{ - client::Client, - stanzastream::{Event as StanzaStreamEvent, StreamEvent}, - Event, Stanza, -}; +use crate::{client::Client, Event}; /// Incoming XMPP events /// @@ -32,39 +27,21 @@ impl Stream for Client { /// /// ...for your client fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll> { - loop { - return Poll::Ready(match ready!(Pin::new(&mut self.stream).poll_next(cx)) { - None => None, - Some(StanzaStreamEvent::Stanza(st)) => match st { - Stanza::Iq(iq) => match self.iq_response_tracker.handle_iq(iq) { - ControlFlow::Break(()) => continue, - ControlFlow::Continue(iq) => Some(Event::Stanza(Stanza::Iq(iq))), - }, - other => Some(Event::Stanza(other)), - }, - Some(StanzaStreamEvent::Stream(StreamEvent::Reset { - bound_jid, - features, - })) => { - self.features = Some(features.clone()); + Poll::Ready(match ready!(self.stanza_rx.poll_recv(cx)) { + None => None, + Some(event) => { + if let Event::Online { + ref bound_jid, + ref features, + .. + } = event + { self.bound_jid = Some(bound_jid.clone()); - - self.iq_response_tracker - .set_account_jid(bound_jid.to_bare()); - - Some(Event::Online { - bound_jid, - features, - resumed: false, - }) + self.features = Some(features.clone()); } - Some(StanzaStreamEvent::Stream(StreamEvent::Resumed)) => Some(Event::Online { - bound_jid: self.bound_jid.as_ref().unwrap().clone(), - features: self.features.as_ref().unwrap().clone(), - resumed: true, - }), - Some(StanzaStreamEvent::Stream(StreamEvent::Suspended)) => continue, - }); - } + + Some(event) + } + }) } } diff --git a/tokio-xmpp/src/client/worker.rs b/tokio-xmpp/src/client/worker.rs new file mode 100644 index 00000000..9f26ad53 --- /dev/null +++ b/tokio-xmpp/src/client/worker.rs @@ -0,0 +1,105 @@ +// Copyright (c) 2025 xmpp-rs contributors. +// +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +use crate::client::iq; +use crate::stanzastream::StanzaReceiver; +use crate::stanzastream::{Event as StanzaStreamEvent, StreamEvent}; +use crate::{Event, Stanza}; +use core::ops::ControlFlow; +use futures::StreamExt; +use tokio::sync::mpsc; +use tokio::sync::oneshot; +use xmpp_parsers::jid::Jid; +use xmpp_parsers::stream_features::StreamFeatures; + +/// Worker to drive the [`crate::stanzastream`] of a client in the background and continue to +/// acknowledge IQs, even when the client is not polled. +pub struct ClientWorker { + // Receiver from the StanzaStream + stream_rx: StanzaReceiver, + // Sender to the client (worker-to-frontend) + stanza_w2f_tx: mpsc::Sender, + // Shutdown signal receiver from frontend + shutdown_rx: oneshot::Receiver<()>, + // JID of the logged-in client + bound_jid: Option, + // Stream features of the currently connected stream + features: Option, + // Response tracker for IQs + iq_response_tracker: iq::IqResponseTracker, +} + +impl ClientWorker { + pub fn new( + stream_rx: StanzaReceiver, + iq_response_tracker: iq::IqResponseTracker, + depth: usize, + ) -> (Self, oneshot::Sender<()>, mpsc::Receiver) { + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + + // worker-to-frontend connection + let (stanza_w2f_tx, stanza_w2f_rx) = mpsc::channel(depth); + + let worker = Self { + stream_rx, + stanza_w2f_tx, + iq_response_tracker, + shutdown_rx, + bound_jid: None, + features: None, + }; + + (worker, shutdown_tx, stanza_w2f_rx) + } + + pub async fn run(mut self) -> StanzaReceiver { + loop { + tokio::select! { + _ = &mut self.shutdown_rx => { + return self.stream_rx; + } + Some(event) = self.stream_rx.next() => { + self.handle_event(event).await; + } + } + } + } + + async fn handle_event(&mut self, event: StanzaStreamEvent) { + let send_event = match event { + StanzaStreamEvent::Stanza(st) => match st { + Stanza::Iq(iq) => match self.iq_response_tracker.handle_iq(iq) { + ControlFlow::Break(()) => return, + ControlFlow::Continue(iq) => Event::Stanza(Stanza::Iq(iq)), + }, + other => Event::Stanza(other), + }, + StanzaStreamEvent::Stream(StreamEvent::Reset { + bound_jid, + features, + }) => { + self.iq_response_tracker + .set_account_jid(bound_jid.to_bare()); + + Event::Online { + bound_jid, + features, + resumed: false, + } + } + StanzaStreamEvent::Stream(StreamEvent::Resumed) => Event::Online { + bound_jid: self.bound_jid.as_ref().unwrap().clone(), + features: self.features.as_ref().unwrap().clone(), + resumed: true, + }, + StanzaStreamEvent::Stream(StreamEvent::Suspended) => return, + }; + + let Ok(()) = self.stanza_w2f_tx.send(send_event).await else { + panic!("All clients have been dropped."); + }; + } +} diff --git a/tokio-xmpp/src/lib.rs b/tokio-xmpp/src/lib.rs index 2fa684b4..b7599339 100644 --- a/tokio-xmpp/src/lib.rs +++ b/tokio-xmpp/src/lib.rs @@ -128,7 +128,11 @@ pub use xso::{asxml::PrintRawXml, error::FromElementError}; #[doc(inline)] /// Generic tokio_xmpp Error pub use crate::error::Error; -pub use client::{auth as client_login, Client, IqFailure, IqRequest, IqResponse, IqResponseToken}; +pub use client::{ + auth as client_login, receiver::ClientReceiver, sender::ClientSender, Client, IqFailure, + IqRequest, IqResponse, IqResponseToken, +}; + #[cfg(feature = "insecure-tcp")] pub use component::Component; pub use event::{Event, Stanza};