xmppftw/xmpp-rs
Watch
1
0
Fork
You've already forked xmpp-rs
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

client: use stanzastream!

Browse source
This commit is contained in:
Jonas Schäfer 2024-08-20 16:54:48 +02:00
commit 35ce86243f
29 changed files with 238 additions and 565 deletions
Download patch file Download diff file Expand all files Collapse all files

56
tokio-xmpp/src/client/bind.rs
View file

@ -1,56 +0,0 @@
use std::io;
use futures::{SinkExt, StreamExt};
use tokio::io::{AsyncBufRead, AsyncWrite};
use xmpp_parsers::bind::{BindQuery, BindResponse};
use xmpp_parsers::iq::{Iq, IqType};
use xmpp_parsers::stream_features::StreamFeatures;
use crate::error::{Error, ProtocolError};
use crate::event::Stanza;
use crate::jid::{FullJid, Jid};
use crate::xmlstream::{ReadError, XmppStream, XmppStreamElement};
const BIND_REQ_ID: &str = "resource-bind";
pub async fn bind<S: AsyncBufRead + AsyncWrite + Unpin>(
stream: &mut XmppStream<S>,
features: &StreamFeatures,
jid: &Jid,
) -> Result<Option<FullJid>, Error> {
if features.can_bind() {
let resource = jid
.resource()
.and_then(|resource| Some(resource.to_string()));
let iq = Iq::from_set(BIND_REQ_ID, BindQuery::new(resource));
stream.send(&iq).await?;
loop {
match stream.next().await {
Some(Ok(XmppStreamElement::Stanza(Stanza::Iq(iq)))) if iq.id == BIND_REQ_ID => {
match iq.payload {
IqType::Result(Some(payload)) => match BindResponse::try_from(payload) {
Ok(v) => {
return Ok(Some(v.into()));
}
Err(_) => return Err(ProtocolError::InvalidBindResponse.into()),
},
_ => return Err(ProtocolError::InvalidBindResponse.into()),
}
}
Some(Ok(_)) => {}
Some(Err(ReadError::SoftTimeout)) => {}
Some(Err(ReadError::HardError(e))) => return Err(e.into()),
Some(Err(ReadError::ParseError(e))) => {
return Err(io::Error::new(io::ErrorKind::InvalidData, e).into())
}
Some(Err(ReadError::StreamFooterReceived)) | None => {
return Err(Error::Disconnected)
}
}
}
} else {
// No resource binding available, do nothing.
return Ok(None);
}
}

18
tokio-xmpp/src/client/login.rs
View file

@ -9,14 +9,13 @@ use std::io;
use std::str::FromStr;
use tokio::io::{AsyncBufRead, AsyncWrite};
use xmpp_parsers::{
jid::{FullJid, Jid},
jid::Jid,
ns,
sasl::{Auth, Mechanism as XMPPMechanism, Nonza, Response},
stream_features::{SaslMechanisms, StreamFeatures},
};
use crate::{
client::bind::bind,
connect::ServerConnector,
error::{AuthError, Error, ProtocolError},
xmlstream::{
@ -133,18 +132,3 @@ pub async fn client_auth<C: ServerConnector>(
.await?;
Ok(stream.recv_features().await?)
}
/// Log into an XMPP server as a client with a jid+pass
/// does channel binding if supported
pub async fn client_login<C: ServerConnector>(
server: C,
jid: Jid,
password: String,
timeouts: Timeouts,
) -> Result<(Option<FullJid>, StreamFeatures, XmppStream<C::Stream>), Error> {
let (features, mut stream) = client_auth(server, jid.clone(), password, timeouts).await?;
// XmppStream bound to user session
let full_jid = bind(&mut stream, &features, &jid).await?;
Ok((full_jid, features, stream))
}

119
tokio-xmpp/src/client/mod.rs
View file

@ -1,11 +1,18 @@
use futures::sink::SinkExt;
// Copyright (c) 2019 Emmanuel Gil Peyrot <linkmauve@linkmauve.fr>
//
// 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 std::io;
use xmpp_parsers::{jid::Jid, stream_features::StreamFeatures};
use crate::{
client::{login::client_login, stream::ClientState},
connect::ServerConnector,
error::Error,
xmlstream::{Timeouts, XmppStream, XmppStreamElement},
stanzastream::{StanzaStage, StanzaState, StanzaStream, StanzaToken},
xmlstream::Timeouts,
Stanza,
};
@ -16,93 +23,68 @@ use crate::connect::StartTlsServerConnector;
#[cfg(feature = "insecure-tcp")]
use crate::connect::TcpServerConnector;
mod bind;
pub(crate) mod login;
mod stream;
/// XMPP client connection and state
///
/// It is able to reconnect. TODO: implement session management.
///
/// This implements the `futures` crate's [`Stream`](#impl-Stream) and
/// [`Sink`](#impl-Sink<Packet>) traits.
pub struct Client<C: ServerConnector> {
jid: Jid,
password: String,
connector: C,
state: ClientState<C::Stream>,
timeouts: Timeouts,
reconnect: bool,
// TODO: tls_required=true
pub struct Client {
stream: StanzaStream,
bound_jid: Option<Jid>,
features: Option<StreamFeatures>,
}
impl<C: ServerConnector> Client<C> {
/// Set whether to reconnect (`true`) or let the stream end
/// (`false`) when a connection to the server has ended.
pub fn set_reconnect(&mut self, reconnect: bool) -> &mut Self {
self.reconnect = reconnect;
self
}
impl Client {
/// 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 bound_jid, .. } => Some(bound_jid),
_ => None,
}
self.bound_jid.as_ref()
}
/// Send stanza
pub async fn send_stanza(&mut self, mut stanza: Stanza) -> Result<(), Error> {
pub async fn send_stanza(&mut self, mut stanza: Stanza) -> Result<StanzaToken, io::Error> {
stanza.ensure_id();
self.send(stanza).await
let mut token = self.stream.send(Box::new(stanza)).await;
match token.wait_for(StanzaStage::Sent).await {
// Queued < Sent, so it cannot be reached.
Some(StanzaState::Queued) => unreachable!(),
None | Some(StanzaState::Dropped) => Err(io::Error::new(
io::ErrorKind::NotConnected,
"stream disconnected fatally before stanza could be sent",
)),
Some(StanzaState::Failed { error }) => Err(error.into_io_error()),
Some(StanzaState::Sent { .. }) | Some(StanzaState::Acked { .. }) => Ok(token),
}
}
/// Get the stream features (`<stream:features/>`) of the underlying stream
pub fn get_stream_features(&self) -> Option<&StreamFeatures> {
match self.state {
ClientState::Connected { ref features, .. } => Some(features),
_ => None,
}
self.features.as_ref()
}
/// End connection by sending `</stream:stream>`
///
/// You may expect the server to respond with the same. This
/// client will then drop its connection.
///
/// Make sure to disable reconnect.
pub async fn send_end(&mut self) -> Result<(), Error> {
match self.state {
ClientState::Connected { ref mut stream, .. } => {
Ok(<XmppStream<C::Stream> as SinkExt<&XmppStreamElement>>::close(stream).await?)
}
ClientState::Connecting { .. } => {
self.state = ClientState::Disconnected;
Ok(())
}
_ => Ok(()),
}
pub async fn send_end(self) -> Result<(), Error> {
self.stream.close().await;
Ok(())
}
}
#[cfg(feature = "starttls")]
impl Client<StartTlsServerConnector> {
impl Client {
/// Start a new XMPP client using StartTLS transport and autoreconnect
///
/// Start polling the returned instance so that it will connect
/// and yield events.
pub fn new<J: Into<Jid>, P: Into<String>>(jid: J, password: P) -> Self {
let jid = jid.into();
let mut client = Self::new_starttls(
jid.clone(),
password,
DnsConfig::srv(&jid.domain().to_string(), "_xmpp-client._tcp", 5222),
Timeouts::default(),
);
client.set_reconnect(true);
client
let dns_config = DnsConfig::srv(&jid.domain().to_string(), "_xmpp-client._tcp", 5222);
Self::new_starttls(jid, password, dns_config, Timeouts::default())
}
/// Start a new XMPP client with StartTLS transport and specific DNS config
@ -122,7 +104,7 @@ impl Client<StartTlsServerConnector> {
}
#[cfg(feature = "insecure-tcp")]
impl Client<TcpServerConnector> {
impl Client {
/// Start a new XMPP client with plaintext insecure connection and specific DNS config
pub fn new_plaintext<J: Into<Jid>, P: Into<String>>(
jid: J,
@ -139,31 +121,18 @@ impl Client<TcpServerConnector> {
}
}
impl<C: ServerConnector> Client<C> {
impl Client {
/// Start a new client given that the JID is already parsed.
pub fn new_with_connector<J: Into<Jid>, P: Into<String>>(
pub fn new_with_connector<J: Into<Jid>, P: Into<String>, C: ServerConnector>(
jid: J,
password: P,
connector: C,
timeouts: Timeouts,
) -> Self {
let jid = jid.into();
let password = password.into();
let connect = tokio::spawn(client_login(
connector.clone(),
jid.clone(),
password.clone(),
timeouts,
));
let client = Client {
jid,
password,
connector,
state: ClientState::Connecting(connect),
reconnect: false,
timeouts,
};
client
Self {
stream: StanzaStream::new_c2s(connector, jid.into(), password.into(), timeouts, 16),
bound_jid: None,
features: None,
}
}
}

222
tokio-xmpp/src/client/stream.rs
View file

@ -1,38 +1,24 @@
use futures::{task::Poll, Future, Sink, Stream};
use std::io;
use std::mem::replace;
// Copyright (c) 2019 Emmanuel Gil Peyrot <linkmauve@linkmauve.fr>
//
// 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 futures::{ready, task::Poll, Stream};
use std::pin::Pin;
use std::task::Context;
use tokio::task::JoinHandle;
use xmpp_parsers::{
jid::{FullJid, Jid},
stream_features::StreamFeatures,
};
use crate::{
client::{login::client_login, Client},
connect::{AsyncReadAndWrite, ServerConnector},
error::Error,
xmlstream::{xmpp::XmppStreamElement, ReadError, XmppStream},
Event, Stanza,
client::Client,
stanzastream::{Event as StanzaStreamEvent, StreamEvent},
Event,
};
pub(crate) enum ClientState<S: AsyncReadAndWrite> {
Invalid,
Disconnected,
Connecting(JoinHandle<Result<(Option<FullJid>, StreamFeatures, XmppStream<S>), Error>>),
Connected {
stream: XmppStream<S>,
features: StreamFeatures,
bound_jid: Jid,
},
}
/// Incoming XMPP events
///
/// In an `async fn` you may want to use this with `use
/// futures::stream::StreamExt;`
impl<C: ServerConnector> Stream for Client<C> {
impl Stream for Client {
type Item = Event;
/// Low-level read on the XMPP stream, allowing the underlying
@ -46,177 +32,27 @@ impl<C: ServerConnector> Stream for Client<C> {
///
/// ...for your client
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
let state = replace(&mut self.state, ClientState::Invalid);
match state {
ClientState::Invalid => panic!("Invalid client state"),
ClientState::Disconnected if self.reconnect => {
// TODO: add timeout
let connect = tokio::spawn(client_login(
self.connector.clone(),
self.jid.clone(),
self.password.clone(),
self.timeouts,
));
self.state = ClientState::Connecting(connect);
self.poll_next(cx)
}
ClientState::Disconnected => {
self.state = ClientState::Disconnected;
Poll::Ready(None)
}
ClientState::Connecting(mut connect) => match Pin::new(&mut connect).poll(cx) {
Poll::Ready(Ok(Ok((bound_jid, features, stream)))) => {
let bound_jid = bound_jid.map(Jid::from).unwrap_or_else(|| self.jid.clone());
self.state = ClientState::Connected {
stream,
bound_jid: bound_jid.clone(),
features,
};
Poll::Ready(Some(Event::Online {
loop {
return Poll::Ready(match ready!(Pin::new(&mut self.stream).poll_next(cx)) {
None => None,
Some(StanzaStreamEvent::Stanza(st)) => Some(Event::Stanza(st)),
Some(StanzaStreamEvent::Stream(StreamEvent::Reset {
bound_jid,
features,
})) => {
self.features = Some(features);
self.bound_jid = Some(bound_jid.clone());
Some(Event::Online {
bound_jid,
resumed: false,
}))
})
}
Poll::Ready(Ok(Err(e))) => {
self.state = ClientState::Disconnected;
return Poll::Ready(Some(Event::Disconnected(e.into())));
}
Poll::Ready(Err(e)) => {
self.state = ClientState::Disconnected;
panic!("connect task: {}", e);
}
Poll::Pending => {
self.state = ClientState::Connecting(connect);
Poll::Pending
}
},
ClientState::Connected {
mut stream,
features,
bound_jid,
} => {
// Poll sink
match <XmppStream<C::Stream> as Sink<&XmppStreamElement>>::poll_ready(
Pin::new(&mut stream),
cx,
) {
Poll::Pending => (),
Poll::Ready(Ok(())) => (),
Poll::Ready(Err(e)) => {
self.state = ClientState::Disconnected;
return Poll::Ready(Some(Event::Disconnected(e.into())));
}
};
// Poll stream
//
// This needs to be a loop in order to ignore packets we don’t care about, or those
// we want to handle elsewhere. Returning something isn’t correct in those two
// cases because it would signal to tokio that the XmppStream is also done, while
// there could be additional packets waiting for us.
//
// The proper solution is thus a loop which we exit once we have something to
// return.
loop {
match Pin::new(&mut stream).poll_next(cx) {
Poll::Ready(None)
| Poll::Ready(Some(Err(ReadError::StreamFooterReceived))) => {
// EOF
self.state = ClientState::Disconnected;
return Poll::Ready(Some(Event::Disconnected(Error::Disconnected)));
}
Poll::Ready(Some(Err(ReadError::HardError(e)))) => {
// Treat stream as dead on I/O errors
self.state = ClientState::Disconnected;
return Poll::Ready(Some(Event::Disconnected(e.into())));
}
Poll::Ready(Some(Err(ReadError::ParseError(e)))) => {
// Treat stream as dead on parse errors, too (for now...)
self.state = ClientState::Disconnected;
return Poll::Ready(Some(Event::Disconnected(
io::Error::new(io::ErrorKind::InvalidData, e).into(),
)));
}
Poll::Ready(Some(Err(ReadError::SoftTimeout))) => {
// TODO: do something smart about this.
}
Poll::Ready(Some(Ok(XmppStreamElement::Stanza(stanza)))) => {
// Receive stanza
self.state = ClientState::Connected {
stream,
features,
bound_jid,
};
return Poll::Ready(Some(Event::Stanza(stanza)));
}
Poll::Ready(Some(Ok(_))) => {
// We ignore these for now.
}
Poll::Pending => {
// Try again later
self.state = ClientState::Connected {
stream,
features,
bound_jid,
};
return Poll::Pending;
}
}
}
}
}
}
}
/// Outgoing XMPP packets
///
/// See `send_stanza()` for an `async fn`
impl<C: ServerConnector> Sink<Stanza> for Client<C> {
type Error = Error;
fn start_send(mut self: Pin<&mut Self>, item: Stanza) -> Result<(), Self::Error> {
match self.state {
ClientState::Connected { ref mut stream, .. } => {
Pin::new(stream).start_send(&item).map_err(|e| e.into())
}
_ => Err(Error::InvalidState),
}
}
fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
match self.state {
ClientState::Connected { ref mut stream, .. } => <XmppStream<C::Stream> as Sink<
&XmppStreamElement,
>>::poll_ready(
Pin::new(stream), cx
)
.map_err(|e| e.into()),
_ => Poll::Pending,
}
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
match self.state {
ClientState::Connected { ref mut stream, .. } => <XmppStream<C::Stream> as Sink<
&XmppStreamElement,
>>::poll_flush(
Pin::new(stream), cx
)
.map_err(|e| e.into()),
_ => Poll::Pending,
}
}
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
match self.state {
ClientState::Connected { ref mut stream, .. } => <XmppStream<C::Stream> as Sink<
&XmppStreamElement,
>>::poll_close(
Pin::new(stream), cx
)
.map_err(|e| e.into()),
_ => Poll::Pending,
Some(StanzaStreamEvent::Stream(StreamEvent::Resumed)) => Some(Event::Online {
bound_jid: self.bound_jid.as_ref().unwrap().clone(),
resumed: true,
}),
Some(StanzaStreamEvent::Stream(StreamEvent::Suspended)) => continue,
});
}
}
}