1//! Components in XMPP are services/gateways that are logged into an
2//! XMPP server under a JID consisting of just a domain name. They are
3//! allowed to use any user and resource identifiers in their stanzas.
4use std::mem::replace;
5use std::str::FromStr;
6use tokio::net::TcpStream;
7use tokio_io::{AsyncRead, AsyncWrite};
8use futures::{Future, Stream, Poll, Async, Sink, StartSend, AsyncSink, done};
9use minidom::Element;
10use jid::{Jid, JidParseError};
11
12use super::xmpp_codec::Packet;
13use super::xmpp_stream;
14use super::happy_eyeballs::Connecter;
15use super::event::Event;
16use super::Error;
17
18mod auth;
19use self::auth::ComponentAuth;
20
21/// Component connection to an XMPP server
22pub struct Component {
23 /// The component's Jabber-Id
24 pub jid: Jid,
25 state: ComponentState,
26}
27
28type XMPPStream = xmpp_stream::XMPPStream<TcpStream>;
29const NS_JABBER_COMPONENT_ACCEPT: &str = "jabber:component:accept";
30
31enum ComponentState {
32 Invalid,
33 Disconnected,
34 Connecting(Box<Future<Item=XMPPStream, Error=Error>>),
35 Connected(XMPPStream),
36}
37
38impl Component {
39 /// Start a new XMPP component
40 ///
41 /// Start polling the returned instance so that it will connect
42 /// and yield events.
43 pub fn new(jid: &str, password: &str, server: &str, port: u16) -> Result<Self, JidParseError> {
44 let jid = Jid::from_str(jid)?;
45 let password = password.to_owned();
46 let connect = Self::make_connect(jid.clone(), password, server, port);
47 Ok(Component {
48 jid,
49 state: ComponentState::Connecting(Box::new(connect)),
50 })
51 }
52
53 fn make_connect(jid: Jid, password: String, server: &str, port: u16) -> impl Future<Item=XMPPStream, Error=Error> {
54 let jid1 = jid.clone();
55 let password = password;
56 done(Connecter::from_lookup(server, "_xmpp-component._tcp", port))
57 .and_then(|connecter| connecter)
58 .map_err(Error::Connection)
59 .and_then(move |tcp_stream| {
60 xmpp_stream::XMPPStream::start(tcp_stream, jid1, NS_JABBER_COMPONENT_ACCEPT.to_owned())
61 }).and_then(move |xmpp_stream| {
62 Self::auth(xmpp_stream, password).expect("auth")
63 })
64 }
65
66 fn auth<S: AsyncRead + AsyncWrite>(stream: xmpp_stream::XMPPStream<S>, password: String) -> Result<ComponentAuth<S>, Error> {
67 ComponentAuth::new(stream, password)
68 }
69}
70
71impl Stream for Component {
72 type Item = Event;
73 type Error = Error;
74
75 fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
76 let state = replace(&mut self.state, ComponentState::Invalid);
77
78 match state {
79 ComponentState::Invalid =>
80 Err(Error::InvalidState),
81 ComponentState::Disconnected =>
82 Ok(Async::Ready(None)),
83 ComponentState::Connecting(mut connect) => {
84 match connect.poll() {
85 Ok(Async::Ready(stream)) => {
86 self.state = ComponentState::Connected(stream);
87 Ok(Async::Ready(Some(Event::Online)))
88 },
89 Ok(Async::NotReady) => {
90 self.state = ComponentState::Connecting(connect);
91 Ok(Async::NotReady)
92 },
93 Err(e) =>
94 Err(e),
95 }
96 },
97 ComponentState::Connected(mut stream) => {
98 // Poll sink
99 match stream.poll_complete() {
100 Ok(Async::NotReady) => (),
101 Ok(Async::Ready(())) => (),
102 Err(e) =>
103 return Err(e)?,
104 };
105
106 // Poll stream
107 match stream.poll() {
108 Ok(Async::NotReady) => {
109 self.state = ComponentState::Connected(stream);
110 Ok(Async::NotReady)
111 },
112 Ok(Async::Ready(None)) => {
113 // EOF
114 self.state = ComponentState::Disconnected;
115 Ok(Async::Ready(Some(Event::Disconnected)))
116 },
117 Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
118 self.state = ComponentState::Connected(stream);
119 Ok(Async::Ready(Some(Event::Stanza(stanza))))
120 },
121 Ok(Async::Ready(_)) => {
122 self.state = ComponentState::Connected(stream);
123 Ok(Async::NotReady)
124 },
125 Err(e) =>
126 Err(e)?,
127 }
128 },
129 }
130 }
131}
132
133impl Sink for Component {
134 type SinkItem = Element;
135 type SinkError = Error;
136
137 fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
138 match self.state {
139 ComponentState::Connected(ref mut stream) =>
140 match stream.start_send(Packet::Stanza(item)) {
141 Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) =>
142 Ok(AsyncSink::NotReady(stanza)),
143 Ok(AsyncSink::NotReady(_)) =>
144 panic!("Component.start_send with stanza but got something else back"),
145 Ok(AsyncSink::Ready) => {
146 Ok(AsyncSink::Ready)
147 },
148 Err(e) =>
149 Err(e)?,
150 },
151 _ =>
152 Ok(AsyncSink::NotReady(item)),
153 }
154 }
155
156 fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
157 match &mut self.state {
158 &mut ComponentState::Connected(ref mut stream) =>
159 stream.poll_complete()
160 .map_err(|e| e.into()),
161 _ =>
162 Ok(Async::Ready(())),
163 }
164 }
165}