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