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;
18
19mod auth;
20use self::auth::ClientAuth;
21mod bind;
22use self::bind::ClientBind;
23mod event;
24pub use self::event::Event as ClientEvent;
25
26pub struct Client {
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=String>>),
38 Connected(XMPPStream),
39}
40
41impl Client {
42 pub fn new(jid: &str, password: &str, handle: Handle) -> Result<Self, JidParseError> {
43 let jid = try!(Jid::from_str(jid));
44 let password = password.to_owned();
45 let connect = Self::make_connect(jid.clone(), password.clone(), handle);
46 Ok(Client {
47 jid,
48 state: ClientState::Connecting(connect),
49 })
50 }
51
52 fn make_connect(jid: Jid, password: String, handle: Handle) -> Box<Future<Item=XMPPStream, Error=String>> {
53 let username = jid.node.as_ref().unwrap().to_owned();
54 let jid1 = jid.clone();
55 let jid2 = jid.clone();
56 let password = password;
57 let domain = match idna::domain_to_ascii(&jid.domain) {
58 Ok(domain) =>
59 domain,
60 Err(e) =>
61 return Box::new(future::err(format!("{:?}", e))),
62 };
63 Box::new(
64 Connecter::from_lookup(handle, &domain, "_xmpp-client._tcp", 5222)
65 .expect("Connector::from_lookup")
66 .and_then(move |tcp_stream|
67 xmpp_stream::XMPPStream::start(tcp_stream, jid1, NS_JABBER_CLIENT.to_owned())
68 .map_err(|e| format!("{}", e))
69 ).and_then(|xmpp_stream| {
70 if Self::can_starttls(&xmpp_stream) {
71 Ok(Self::starttls(xmpp_stream))
72 } else {
73 Err("No STARTTLS".to_owned())
74 }
75 }).and_then(|starttls|
76 starttls
77 ).and_then(|tls_stream|
78 XMPPStream::start(tls_stream, jid2, NS_JABBER_CLIENT.to_owned())
79 .map_err(|e| format!("{}", e))
80 ).and_then(move |xmpp_stream| {
81 Self::auth(xmpp_stream, username, password).expect("auth")
82 }).and_then(|xmpp_stream| {
83 Self::bind(xmpp_stream)
84 }).and_then(|xmpp_stream| {
85 println!("Bound to {}", xmpp_stream.jid);
86 Ok(xmpp_stream)
87 })
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>, String> {
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 = ClientEvent;
116 type Error = String;
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("invalid client state".to_owned()),
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(ClientEvent::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 match stream.poll() {
142 Ok(Async::Ready(None)) => {
143 // EOF
144 self.state = ClientState::Disconnected;
145 Ok(Async::Ready(Some(ClientEvent::Disconnected)))
146 },
147 Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
148 self.state = ClientState::Connected(stream);
149 Ok(Async::Ready(Some(ClientEvent::Stanza(stanza))))
150 },
151 Ok(Async::NotReady) |
152 Ok(Async::Ready(_)) => {
153 self.state = ClientState::Connected(stream);
154 Ok(Async::NotReady)
155 },
156 Err(e) =>
157 Err(e.description().to_owned()),
158 }
159 },
160 }
161 }
162}
163
164impl Sink for Client {
165 type SinkItem = Element;
166 type SinkError = String;
167
168 fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
169 match self.state {
170 ClientState::Connected(ref mut stream) =>
171 match stream.start_send(Packet::Stanza(item)) {
172 Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) =>
173 Ok(AsyncSink::NotReady(stanza)),
174 Ok(AsyncSink::NotReady(_)) =>
175 panic!("Client.start_send with stanza but got something else back"),
176 Ok(AsyncSink::Ready) => {
177 Ok(AsyncSink::Ready)
178 },
179 Err(e) =>
180 Err(e.description().to_owned()),
181 },
182 _ =>
183 Ok(AsyncSink::NotReady(item)),
184 }
185 }
186
187 fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
188 match self.state {
189 ClientState::Connected(ref mut stream) =>
190 stream.poll_complete()
191 .map_err(|e| e.description().to_owned()),
192 _ =>
193 Ok(Async::Ready(())),
194 }
195 }
196}