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 futures::{done, Async, AsyncSink, Future, Poll, Sink, StartSend, Stream};
5use jid::{Jid, JidParseError};
6use minidom::Element;
7use std::mem::replace;
8use std::str::FromStr;
9use tokio::net::TcpStream;
10use tokio_io::{AsyncRead, AsyncWrite};
11
12use super::event::Event;
13use super::happy_eyeballs::Connecter;
14use super::xmpp_codec::Packet;
15use super::xmpp_stream;
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(
54 jid: Jid,
55 password: String,
56 server: &str,
57 port: u16,
58 ) -> impl Future<Item = XMPPStream, Error = Error> {
59 let jid1 = jid.clone();
60 let password = password;
61 done(Connecter::from_lookup(server, None, port))
62 .flatten()
63 .and_then(move |tcp_stream| {
64 xmpp_stream::XMPPStream::start(
65 tcp_stream,
66 jid1,
67 NS_JABBER_COMPONENT_ACCEPT.to_owned(),
68 )
69 })
70 .and_then(move |xmpp_stream| Self::auth(xmpp_stream, password).expect("auth"))
71 }
72
73 fn auth<S: AsyncRead + AsyncWrite>(
74 stream: xmpp_stream::XMPPStream<S>,
75 password: String,
76 ) -> Result<ComponentAuth<S>, Error> {
77 ComponentAuth::new(stream, password)
78 }
79}
80
81impl Stream for Component {
82 type Item = Event;
83 type Error = Error;
84
85 fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
86 let state = replace(&mut self.state, ComponentState::Invalid);
87
88 match state {
89 ComponentState::Invalid => Err(Error::InvalidState),
90 ComponentState::Disconnected => Ok(Async::Ready(None)),
91 ComponentState::Connecting(mut connect) => match connect.poll() {
92 Ok(Async::Ready(stream)) => {
93 self.state = ComponentState::Connected(stream);
94 Ok(Async::Ready(Some(Event::Online)))
95 }
96 Ok(Async::NotReady) => {
97 self.state = ComponentState::Connecting(connect);
98 Ok(Async::NotReady)
99 }
100 Err(e) => Err(e),
101 },
102 ComponentState::Connected(mut stream) => {
103 // Poll sink
104 match stream.poll_complete() {
105 Ok(Async::NotReady) => (),
106 Ok(Async::Ready(())) => (),
107 Err(e) => return Err(e)?,
108 };
109
110 // Poll stream
111 match stream.poll() {
112 Ok(Async::NotReady) => {
113 self.state = ComponentState::Connected(stream);
114 Ok(Async::NotReady)
115 }
116 Ok(Async::Ready(None)) => {
117 // EOF
118 self.state = ComponentState::Disconnected;
119 Ok(Async::Ready(Some(Event::Disconnected)))
120 }
121 Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
122 self.state = ComponentState::Connected(stream);
123 Ok(Async::Ready(Some(Event::Stanza(stanza))))
124 }
125 Ok(Async::Ready(_)) => {
126 self.state = ComponentState::Connected(stream);
127 Ok(Async::NotReady)
128 }
129 Err(e) => Err(e)?,
130 }
131 }
132 }
133 }
134}
135
136impl Sink for Component {
137 type SinkItem = Element;
138 type SinkError = Error;
139
140 fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
141 match self.state {
142 ComponentState::Connected(ref mut stream) => match stream
143 .start_send(Packet::Stanza(item))
144 {
145 Ok(AsyncSink::NotReady(Packet::Stanza(stanza))) => Ok(AsyncSink::NotReady(stanza)),
146 Ok(AsyncSink::NotReady(_)) => {
147 panic!("Component.start_send with stanza but got something else back")
148 }
149 Ok(AsyncSink::Ready) => Ok(AsyncSink::Ready),
150 Err(e) => Err(e)?,
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().map_err(|e| e.into())
160 }
161 _ => Ok(Async::Ready(())),
162 }
163 }
164}