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 futures::{Future, Stream, Poll, Async, Sink, StartSend, AsyncSink};
8use minidom::Element;
9use jid::{Jid, JidParseError};
10
11use super::xmpp_codec::Packet;
12use super::xmpp_stream;
13use super::happy_eyeballs::Connecter;
14use super::event::Event;
15
16mod auth;
17use self::auth::ComponentAuth;
18
19pub struct Component {
20 pub jid: Jid,
21 state: ComponentState,
22}
23
24type XMPPStream = xmpp_stream::XMPPStream<TcpStream>;
25const NS_JABBER_COMPONENT_ACCEPT: &str = "jabber:component:accept";
26
27enum ComponentState {
28 Invalid,
29 Disconnected,
30 Connecting(Box<Future<Item=XMPPStream, Error=String>>),
31 Connected(XMPPStream),
32}
33
34impl Component {
35 pub fn new(jid: &str, password: &str, server: &str, port: u16, handle: Handle) -> Result<Self, JidParseError> {
36 let jid = try!(Jid::from_str(jid));
37 let password = password.to_owned();
38 let connect = Self::make_connect(jid.clone(), password, server, port, handle);
39 Ok(Component {
40 jid,
41 state: ComponentState::Connecting(connect),
42 })
43 }
44
45 fn make_connect(jid: Jid, password: String, server: &str, port: u16, handle: Handle) -> Box<Future<Item=XMPPStream, Error=String>> {
46 let jid1 = jid.clone();
47 let password = password;
48 Box::new(
49 Connecter::from_lookup(handle, server, "_xmpp-component._tcp", port)
50 .expect("Connector::from_lookup")
51 .and_then(move |tcp_stream| {
52 xmpp_stream::XMPPStream::start(tcp_stream, jid1, NS_JABBER_COMPONENT_ACCEPT.to_owned())
53 .map_err(|e| format!("{}", e))
54 }).and_then(move |xmpp_stream| {
55 Self::auth(xmpp_stream, password).expect("auth")
56 }).and_then(|xmpp_stream| {
57 println!("Bound to {}", xmpp_stream.jid);
58 Ok(xmpp_stream)
59 })
60 )
61 }
62
63 fn auth<S: AsyncRead + AsyncWrite>(stream: xmpp_stream::XMPPStream<S>, password: String) -> Result<ComponentAuth<S>, String> {
64 ComponentAuth::new(stream, password)
65 }
66}
67
68impl Stream for Component {
69 type Item = Event;
70 type Error = String;
71
72 fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
73 let state = replace(&mut self.state, ComponentState::Invalid);
74
75 match state {
76 ComponentState::Invalid =>
77 Err("invalid client state".to_owned()),
78 ComponentState::Disconnected =>
79 Ok(Async::Ready(None)),
80 ComponentState::Connecting(mut connect) => {
81 match connect.poll() {
82 Ok(Async::Ready(stream)) => {
83 self.state = ComponentState::Connected(stream);
84 Ok(Async::Ready(Some(Event::Online)))
85 },
86 Ok(Async::NotReady) => {
87 self.state = ComponentState::Connecting(connect);
88 Ok(Async::NotReady)
89 },
90 Err(e) =>
91 Err(e),
92 }
93 },
94 ComponentState::Connected(mut stream) => {
95 // Poll sink
96 match stream.poll_complete() {
97 Ok(Async::NotReady) => (),
98 Ok(Async::Ready(())) => (),
99 Err(e) =>
100 return Err(e.description().to_owned()),
101 };
102
103 // Poll stream
104 match stream.poll() {
105 Ok(Async::NotReady) => {
106 self.state = ComponentState::Connected(stream);
107 Ok(Async::NotReady)
108 },
109 Ok(Async::Ready(None)) => {
110 // EOF
111 self.state = ComponentState::Disconnected;
112 Ok(Async::Ready(Some(Event::Disconnected)))
113 },
114 Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
115 self.state = ComponentState::Connected(stream);
116 Ok(Async::Ready(Some(Event::Stanza(stanza))))
117 },
118 Ok(Async::Ready(_)) => {
119 self.state = ComponentState::Connected(stream);
120 Ok(Async::NotReady)
121 },
122 Err(e) =>
123 Err(e.description().to_owned()),
124 }
125 },
126 }
127 }
128}
129
130impl Sink for Component {
131 type SinkItem = Element;
132 type SinkError = String;
133
134 fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
135 match self.state {
136 ComponentState::Connected(ref mut stream) =>
137 match stream.start_send(Packet::Stanza(item)) {
138 Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) =>
139 Ok(AsyncSink::NotReady(stanza)),
140 Ok(AsyncSink::NotReady(_)) =>
141 panic!("Component.start_send with stanza but got something else back"),
142 Ok(AsyncSink::Ready) => {
143 Ok(AsyncSink::Ready)
144 },
145 Err(e) =>
146 Err(e.description().to_owned()),
147 },
148 _ =>
149 Ok(AsyncSink::NotReady(item)),
150 }
151 }
152
153 fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
154 match &mut self.state {
155 &mut ComponentState::Connected(ref mut stream) =>
156 stream.poll_complete()
157 .map_err(|e| e.description().to_owned()),
158 _ =>
159 Ok(Async::Ready(())),
160 }
161 }
162}