1use std::mem::replace;
2use std::str::FromStr;
3use tokio::net::TcpStream;
4use tokio_io::{AsyncRead, AsyncWrite};
5use tokio_tls::TlsStream;
6use futures::{Future, Stream, Poll, Async, Sink, StartSend, AsyncSink, done};
7use minidom::Element;
8use jid::{Jid, JidParseError};
9use sasl::common::{Credentials, ChannelBinding};
10use idna;
11
12use super::xmpp_codec::Packet;
13use super::xmpp_stream;
14use super::starttls::{NS_XMPP_TLS, StartTlsClient};
15use super::happy_eyeballs::Connecter;
16use super::event::Event;
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(&domain, Some("_xmpp-client._tcp"), 5222))
65 .map_err(Error::Connection)
66 )
67 .flatten()
68 .and_then(move |tcp_stream|
69 xmpp_stream::XMPPStream::start(tcp_stream, jid1, NS_JABBER_CLIENT.to_owned())
70 ).and_then(|xmpp_stream| {
71 if Self::can_starttls(&xmpp_stream) {
72 Ok(Self::starttls(xmpp_stream))
73 } else {
74 Err(Error::Protocol(ProtocolError::NoTls))
75 }
76 }).flatten()
77 .and_then(|tls_stream|
78 XMPPStream::start(tls_stream, jid2, NS_JABBER_CLIENT.to_owned())
79 ).and_then(move |xmpp_stream|
80 done(Self::auth(xmpp_stream, username, password))
81 // TODO: flatten?
82 ).and_then(|auth| auth)
83 .and_then(|xmpp_stream| {
84 Self::bind(xmpp_stream)
85 }).and_then(|xmpp_stream| {
86 // println!("Bound to {}", xmpp_stream.jid);
87 Ok(xmpp_stream)
88 })
89 }
90
91 fn can_starttls<S>(stream: &xmpp_stream::XMPPStream<S>) -> bool {
92 stream.stream_features
93 .get_child("starttls", NS_XMPP_TLS)
94 .is_some()
95 }
96
97 fn starttls<S: AsyncRead + AsyncWrite>(stream: xmpp_stream::XMPPStream<S>) -> StartTlsClient<S> {
98 StartTlsClient::from_stream(stream)
99 }
100
101 fn auth<S: AsyncRead + AsyncWrite>(stream: xmpp_stream::XMPPStream<S>, username: String, password: String) -> Result<ClientAuth<S>, Error> {
102 let creds = Credentials::default()
103 .with_username(username)
104 .with_password(password)
105 .with_channel_binding(ChannelBinding::None);
106 ClientAuth::new(stream, creds)
107 }
108
109 fn bind<S: AsyncWrite>(stream: xmpp_stream::XMPPStream<S>) -> ClientBind<S> {
110 ClientBind::new(stream)
111 }
112}
113
114impl Stream for Client {
115 type Item = Event;
116 type Error = Error;
117
118 fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
119 let state = replace(&mut self.state, ClientState::Invalid);
120
121 match state {
122 ClientState::Invalid =>
123 Err(Error::InvalidState),
124 ClientState::Disconnected =>
125 Ok(Async::Ready(None)),
126 ClientState::Connecting(mut connect) => {
127 match connect.poll() {
128 Ok(Async::Ready(stream)) => {
129 self.state = ClientState::Connected(stream);
130 Ok(Async::Ready(Some(Event::Online)))
131 },
132 Ok(Async::NotReady) => {
133 self.state = ClientState::Connecting(connect);
134 Ok(Async::NotReady)
135 },
136 Err(e) =>
137 Err(e),
138 }
139 },
140 ClientState::Connected(mut stream) => {
141 // Poll sink
142 match stream.poll_complete() {
143 Ok(Async::NotReady) => (),
144 Ok(Async::Ready(())) => (),
145 Err(e) =>
146 return Err(e)?,
147 };
148
149 // Poll stream
150 match stream.poll() {
151 Ok(Async::Ready(None)) => {
152 // EOF
153 self.state = ClientState::Disconnected;
154 Ok(Async::Ready(Some(Event::Disconnected)))
155 },
156 Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
157 self.state = ClientState::Connected(stream);
158 Ok(Async::Ready(Some(Event::Stanza(stanza))))
159 },
160 Ok(Async::NotReady) |
161 Ok(Async::Ready(_)) => {
162 self.state = ClientState::Connected(stream);
163 Ok(Async::NotReady)
164 },
165 Err(e) =>
166 Err(e)?,
167 }
168 },
169 }
170 }
171}
172
173impl Sink for Client {
174 type SinkItem = Element;
175 type SinkError = Error;
176
177 fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
178 match self.state {
179 ClientState::Connected(ref mut stream) =>
180 match stream.start_send(Packet::Stanza(item)) {
181 Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) =>
182 Ok(AsyncSink::NotReady(stanza)),
183 Ok(AsyncSink::NotReady(_)) =>
184 panic!("Client.start_send with stanza but got something else back"),
185 Ok(AsyncSink::Ready) => {
186 Ok(AsyncSink::Ready)
187 },
188 Err(e) =>
189 Err(e)?,
190 },
191 _ =>
192 Ok(AsyncSink::NotReady(item)),
193 }
194 }
195
196 fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
197 match self.state {
198 ClientState::Connected(ref mut stream) =>
199 stream.poll_complete()
200 .map_err(|e| e.into()),
201 _ =>
202 Ok(Async::Ready(())),
203 }
204 }
205}