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, Stream, Poll, Async, Sink, StartSend, AsyncSink};
9use minidom::Element;
10use jid::{Jid, JidParseError};
11use sasl::common::{Credentials, ChannelBinding};
12
13use super::xmpp_codec::Packet;
14use super::xmpp_stream;
15use super::starttls::{NS_XMPP_TLS, StartTlsClient};
16use super::happy_eyeballs::Connecter;
17
18mod auth;
19use self::auth::ClientAuth;
20mod bind;
21use self::bind::ClientBind;
22mod event;
23pub use self::event::Event as ClientEvent;
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 Box::new(
57 Connecter::from_lookup(handle, &jid.domain, "_xmpp-client._tcp", 5222)
58 .expect("Connector::from_lookup")
59 .and_then(move |tcp_stream|
60 xmpp_stream::XMPPStream::start(tcp_stream, jid1, NS_JABBER_CLIENT.to_owned())
61 .map_err(|e| format!("{}", e))
62 ).and_then(|xmpp_stream| {
63 if Self::can_starttls(&xmpp_stream) {
64 Self::starttls(xmpp_stream)
65 } else {
66 panic!("No STARTTLS")
67 }
68 }).and_then(|tls_stream|
69 XMPPStream::start(tls_stream, jid2, NS_JABBER_CLIENT.to_owned())
70 .map_err(|e| format!("{}", e))
71 ).and_then(move |xmpp_stream| {
72 Self::auth(xmpp_stream, username, password).expect("auth")
73 }).and_then(|xmpp_stream| {
74 Self::bind(xmpp_stream)
75 }).and_then(|xmpp_stream| {
76 println!("Bound to {}", xmpp_stream.jid);
77 Ok(xmpp_stream)
78 })
79 )
80 }
81
82 fn can_starttls<S>(stream: &xmpp_stream::XMPPStream<S>) -> bool {
83 stream.stream_features
84 .get_child("starttls", NS_XMPP_TLS)
85 .is_some()
86 }
87
88 fn starttls<S: AsyncRead + AsyncWrite>(stream: xmpp_stream::XMPPStream<S>) -> StartTlsClient<S> {
89 StartTlsClient::from_stream(stream)
90 }
91
92 fn auth<S: AsyncRead + AsyncWrite>(stream: xmpp_stream::XMPPStream<S>, username: String, password: String) -> Result<ClientAuth<S>, String> {
93 let creds = Credentials::default()
94 .with_username(username)
95 .with_password(password)
96 .with_channel_binding(ChannelBinding::None);
97 ClientAuth::new(stream, creds)
98 }
99
100 fn bind<S: AsyncWrite>(stream: xmpp_stream::XMPPStream<S>) -> ClientBind<S> {
101 ClientBind::new(stream)
102 }
103}
104
105impl Stream for Client {
106 type Item = ClientEvent;
107 type Error = String;
108
109 fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
110 let state = replace(&mut self.state, ClientState::Invalid);
111
112 match state {
113 ClientState::Invalid =>
114 Err("invalid client state".to_owned()),
115 ClientState::Disconnected =>
116 Ok(Async::Ready(None)),
117 ClientState::Connecting(mut connect) => {
118 match connect.poll() {
119 Ok(Async::Ready(stream)) => {
120 self.state = ClientState::Connected(stream);
121 Ok(Async::Ready(Some(ClientEvent::Online)))
122 },
123 Ok(Async::NotReady) => {
124 self.state = ClientState::Connecting(connect);
125 Ok(Async::NotReady)
126 },
127 Err(e) =>
128 Err(e),
129 }
130 },
131 ClientState::Connected(mut stream) => {
132 match stream.poll() {
133 Ok(Async::Ready(None)) => {
134 // EOF
135 self.state = ClientState::Disconnected;
136 Ok(Async::Ready(Some(ClientEvent::Disconnected)))
137 },
138 Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
139 self.state = ClientState::Connected(stream);
140 Ok(Async::Ready(Some(ClientEvent::Stanza(stanza))))
141 },
142 Ok(Async::NotReady) |
143 Ok(Async::Ready(_)) => {
144 self.state = ClientState::Connected(stream);
145 Ok(Async::NotReady)
146 },
147 Err(e) =>
148 Err(e.description().to_owned()),
149 }
150 },
151 }
152 }
153}
154
155impl Sink for Client {
156 type SinkItem = Element;
157 type SinkError = String;
158
159 fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
160 match self.state {
161 ClientState::Connected(ref mut stream) =>
162 match stream.start_send(Packet::Stanza(item)) {
163 Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) =>
164 Ok(AsyncSink::NotReady(stanza)),
165 Ok(AsyncSink::NotReady(_)) =>
166 panic!("Client.start_send with stanza but got something else back"),
167 Ok(AsyncSink::Ready) => {
168 Ok(AsyncSink::Ready)
169 },
170 Err(e) =>
171 Err(e.description().to_owned()),
172 },
173 _ =>
174 Ok(AsyncSink::NotReady(item)),
175 }
176 }
177
178 fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
179 match self.state {
180 ClientState::Connected(ref mut stream) =>
181 stream.poll_complete()
182 .map_err(|e| e.description().to_owned()),
183 _ =>
184 Ok(Async::Ready(())),
185 }
186 }
187}