1use futures::{done, Async, AsyncSink, Future, Poll, Sink, StartSend, Stream};
2use idna;
3use jid::{Jid, JidParseError};
4use minidom::Element;
5use sasl::common::{ChannelBinding, Credentials};
6use std::mem::replace;
7use std::str::FromStr;
8use tokio::net::TcpStream;
9use tokio_io::{AsyncRead, AsyncWrite};
10use tokio_tls::TlsStream;
11
12use super::event::Event;
13use super::happy_eyeballs::Connecter;
14use super::starttls::{StartTlsClient, NS_XMPP_TLS};
15use super::xmpp_codec::Packet;
16use super::xmpp_stream;
17use super::{Error, ProtocolError};
18
19mod auth;
20use self::auth::ClientAuth;
21mod bind;
22use self::bind::ClientBind;
23
24/// XMPP client connection and state
25pub struct Client {
26 /// The client's current Jabber-Id
27 pub jid: Jid,
28 state: ClientState,
29}
30
31type XMPPStream = xmpp_stream::XMPPStream<TlsStream<TcpStream>>;
32const NS_JABBER_CLIENT: &str = "jabber:client";
33
34enum ClientState {
35 Invalid,
36 Disconnected,
37 Connecting(Box<Future<Item = XMPPStream, Error = Error>>),
38 Connected(XMPPStream),
39}
40
41impl Client {
42 /// Start a new XMPP client
43 ///
44 /// Start polling the returned instance so that it will connect
45 /// and yield events.
46 pub fn new(jid: &str, password: &str) -> Result<Self, JidParseError> {
47 let jid = Jid::from_str(jid)?;
48 let password = password.to_owned();
49 let connect = Self::make_connect(jid.clone(), password.clone());
50 Ok(Client {
51 jid,
52 state: ClientState::Connecting(Box::new(connect)),
53 })
54 }
55
56 fn make_connect(jid: Jid, password: String) -> impl Future<Item = XMPPStream, Error = Error> {
57 let username = jid.node.as_ref().unwrap().to_owned();
58 let jid1 = jid.clone();
59 let jid2 = jid.clone();
60 let password = password;
61 done(idna::domain_to_ascii(&jid.domain))
62 .map_err(|_| Error::Idna)
63 .and_then(|domain| {
64 done(Connecter::from_lookup(
65 &domain,
66 Some("_xmpp-client._tcp"),
67 5222,
68 ))
69 })
70 .flatten()
71 .and_then(move |tcp_stream| {
72 xmpp_stream::XMPPStream::start(tcp_stream, jid1, NS_JABBER_CLIENT.to_owned())
73 })
74 .and_then(|xmpp_stream| {
75 if Self::can_starttls(&xmpp_stream) {
76 Ok(Self::starttls(xmpp_stream))
77 } else {
78 Err(Error::Protocol(ProtocolError::NoTls))
79 }
80 })
81 .flatten()
82 .and_then(|tls_stream| XMPPStream::start(tls_stream, jid2, NS_JABBER_CLIENT.to_owned()))
83 .and_then(
84 move |xmpp_stream| done(Self::auth(xmpp_stream, username, password)), // TODO: flatten?
85 )
86 .and_then(|auth| auth)
87 .and_then(|xmpp_stream| Self::bind(xmpp_stream))
88 .and_then(|xmpp_stream| {
89 // println!("Bound to {}", xmpp_stream.jid);
90 Ok(xmpp_stream)
91 })
92 }
93
94 fn can_starttls<S>(stream: &xmpp_stream::XMPPStream<S>) -> bool {
95 stream
96 .stream_features
97 .get_child("starttls", NS_XMPP_TLS)
98 .is_some()
99 }
100
101 fn starttls<S: AsyncRead + AsyncWrite>(
102 stream: xmpp_stream::XMPPStream<S>,
103 ) -> StartTlsClient<S> {
104 StartTlsClient::from_stream(stream)
105 }
106
107 fn auth<S: AsyncRead + AsyncWrite + 'static>(
108 stream: xmpp_stream::XMPPStream<S>,
109 username: String,
110 password: String,
111 ) -> Result<ClientAuth<S>, Error> {
112 let creds = Credentials::default()
113 .with_username(username)
114 .with_password(password)
115 .with_channel_binding(ChannelBinding::None);
116 ClientAuth::new(stream, creds)
117 }
118
119 fn bind<S: AsyncWrite>(stream: xmpp_stream::XMPPStream<S>) -> ClientBind<S> {
120 ClientBind::new(stream)
121 }
122}
123
124impl Stream for Client {
125 type Item = Event;
126 type Error = Error;
127
128 fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
129 let state = replace(&mut self.state, ClientState::Invalid);
130
131 match state {
132 ClientState::Invalid => Err(Error::InvalidState),
133 ClientState::Disconnected => Ok(Async::Ready(None)),
134 ClientState::Connecting(mut connect) => match connect.poll() {
135 Ok(Async::Ready(stream)) => {
136 self.state = ClientState::Connected(stream);
137 Ok(Async::Ready(Some(Event::Online)))
138 }
139 Ok(Async::NotReady) => {
140 self.state = ClientState::Connecting(connect);
141 Ok(Async::NotReady)
142 }
143 Err(e) => Err(e),
144 },
145 ClientState::Connected(mut stream) => {
146 // Poll sink
147 match stream.poll_complete() {
148 Ok(Async::NotReady) => (),
149 Ok(Async::Ready(())) => (),
150 Err(e) => return Err(e)?,
151 };
152
153 // Poll stream
154 match stream.poll() {
155 Ok(Async::Ready(None)) => {
156 // EOF
157 self.state = ClientState::Disconnected;
158 Ok(Async::Ready(Some(Event::Disconnected)))
159 }
160 Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
161 self.state = ClientState::Connected(stream);
162 Ok(Async::Ready(Some(Event::Stanza(stanza))))
163 }
164 Ok(Async::NotReady) | Ok(Async::Ready(_)) => {
165 self.state = ClientState::Connected(stream);
166 Ok(Async::NotReady)
167 }
168 Err(e) => Err(e)?,
169 }
170 }
171 }
172 }
173}
174
175impl Sink for Client {
176 type SinkItem = Element;
177 type SinkError = Error;
178
179 fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
180 match self.state {
181 ClientState::Connected(ref mut stream) => match stream.start_send(Packet::Stanza(item))
182 {
183 Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) => Ok(AsyncSink::NotReady(stanza)),
184 Ok(AsyncSink::NotReady(_)) => {
185 panic!("Client.start_send with stanza but got something else back")
186 }
187 Ok(AsyncSink::Ready) => Ok(AsyncSink::Ready),
188 Err(e) => Err(e)?,
189 },
190 _ => Ok(AsyncSink::NotReady(item)),
191 }
192 }
193
194 fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
195 match self.state {
196 ClientState::Connected(ref mut stream) => stream.poll_complete().map_err(|e| e.into()),
197 _ => Ok(Async::Ready(())),
198 }
199 }
200}