1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184
use futures::{task::Poll, Future, Sink, Stream};
use std::mem::replace;
use std::pin::Pin;
use std::task::Context;
use tokio::task::JoinHandle;
use crate::{
client::login::client_login,
connect::{AsyncReadAndWrite, ServerConnector},
error::{Error, ProtocolError},
proto::{Packet, XmppStream},
Client, Event,
};
pub(crate) enum ClientState<S: AsyncReadAndWrite> {
Invalid,
Disconnected,
Connecting(JoinHandle<Result<XmppStream<S>, Error>>),
Connected(XmppStream<S>),
}
/// 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> {
type Item = Event;
/// Low-level read on the XMPP stream, allowing the underlying
/// machinery to:
///
/// * connect,
/// * starttls,
/// * authenticate,
/// * bind a session, and finally
/// * receive stanzas
///
/// ...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.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(stream))) => {
let bound_jid = stream.jid.clone();
self.state = ClientState::Connected(stream);
Poll::Ready(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) => {
// Poll sink
match Pin::new(&mut stream).poll_ready(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) => {
// EOF
self.state = ClientState::Disconnected;
return Poll::Ready(Some(Event::Disconnected(Error::Disconnected)));
}
Poll::Ready(Some(Ok(Packet::Stanza(stanza)))) => {
// Receive stanza
self.state = ClientState::Connected(stream);
return Poll::Ready(Some(Event::Stanza(stanza)));
}
Poll::Ready(Some(Ok(Packet::Text(_)))) => {
// Ignore text between stanzas
}
Poll::Ready(Some(Ok(Packet::StreamStart(_)))) => {
// <stream:stream>
self.state = ClientState::Disconnected;
return Poll::Ready(Some(Event::Disconnected(
ProtocolError::InvalidStreamStart.into(),
)));
}
Poll::Ready(Some(Ok(Packet::StreamEnd))) => {
// End of stream: </stream:stream>
self.state = ClientState::Disconnected;
return Poll::Ready(Some(Event::Disconnected(Error::Disconnected)));
}
Poll::Pending => {
// Try again later
self.state = ClientState::Connected(stream);
return Poll::Pending;
}
Poll::Ready(Some(Err(e))) => {
self.state = ClientState::Disconnected;
return Poll::Ready(Some(Event::Disconnected(e.into())));
}
}
}
}
}
}
}
/// Outgoing XMPP packets
///
/// See `send_stanza()` for an `async fn`
impl<C: ServerConnector> Sink<Packet> for Client<C> {
type Error = Error;
fn start_send(mut self: Pin<&mut Self>, item: Packet) -> 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) => {
Pin::new(stream).poll_ready(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) => {
Pin::new(stream).poll_flush(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) => {
Pin::new(stream).poll_close(cx).map_err(|e| e.into())
}
_ => Poll::Pending,
}
}
}