1// Copyright (c) 2019 Emmanuel Gil Peyrot <linkmauve@linkmauve.fr>
2//
3// This Source Code Form is subject to the terms of the Mozilla Public
4// License, v. 2.0. If a copy of the MPL was not distributed with this
5// file, You can obtain one at http://mozilla.org/MPL/2.0/.
6
7#![deny(bare_trait_objects)]
8
9use std::str::FromStr;
10use futures::{Future,Stream, Sink, sync::mpsc};
11use tokio_xmpp::{
12 Client as TokioXmppClient,
13 Event as TokioXmppEvent,
14 Packet,
15};
16use xmpp_parsers::{
17 caps::{compute_disco, hash_caps, Caps},
18 disco::{DiscoInfoQuery, DiscoInfoResult, Feature, Identity},
19 hashes::Algo,
20 iq::{Iq, IqType},
21 message::{Message, MessageType, Body},
22 muc::{
23 Muc,
24 user::{MucUser, Status},
25 },
26 ns,
27 presence::{Presence, Type as PresenceType},
28 pubsub::{
29 event::PubSubEvent,
30 pubsub::PubSub,
31 },
32 roster::{Roster, Item as RosterItem},
33 stanza_error::{StanzaError, ErrorType, DefinedCondition},
34 Jid, JidParseError, TryFrom,
35};
36
37mod avatar;
38
39pub type Error = tokio_xmpp::Error;
40
41#[derive(Debug)]
42pub enum ClientType {
43 Bot,
44 Pc,
45}
46
47impl Default for ClientType {
48 fn default() -> Self {
49 ClientType::Bot
50 }
51}
52
53impl ToString for ClientType {
54 fn to_string(&self) -> String {
55 String::from(
56 match self {
57 ClientType::Bot => "bot",
58 ClientType::Pc => "pc",
59 }
60 )
61 }
62}
63
64#[derive(PartialEq)]
65pub enum ClientFeature {
66 Avatars,
67 ContactList,
68}
69
70pub enum Event {
71 Online,
72 Disconnected,
73 ContactAdded(RosterItem),
74 ContactRemoved(RosterItem),
75 ContactChanged(RosterItem),
76 AvatarRetrieved(Jid, String),
77 RoomJoined(Jid),
78}
79
80#[derive(Default)]
81pub struct ClientBuilder<'a> {
82 jid: &'a str,
83 password: &'a str,
84 website: String,
85 disco: (ClientType, String),
86 features: Vec<ClientFeature>,
87}
88
89impl ClientBuilder<'_> {
90 pub fn new<'a>(jid: &'a str, password: &'a str) -> ClientBuilder<'a> {
91 ClientBuilder {
92 jid,
93 password,
94 website: String::from("https://gitlab.com/xmpp-rs/tokio-xmpp"),
95 disco: (ClientType::default(), String::from("tokio-xmpp")),
96 features: vec![],
97 }
98 }
99
100 pub fn set_client(mut self, type_: ClientType, name: &str) -> Self {
101 self.disco = (type_, String::from(name));
102 self
103 }
104
105 pub fn set_website(mut self, url: &str) -> Self {
106 self.website = String::from(url);
107 self
108 }
109
110 pub fn enable_feature(mut self, feature: ClientFeature) -> Self {
111 self.features.push(feature);
112 self
113 }
114
115 fn make_disco(&self) -> DiscoInfoResult {
116 let identities = vec![Identity::new("client", self.disco.0.to_string(),
117 "en", self.disco.1.to_string())];
118 let mut features = vec![
119 Feature::new(ns::DISCO_INFO),
120 ];
121 if self.features.contains(&ClientFeature::Avatars) {
122 features.push(Feature::new(format!("{}+notify", ns::AVATAR_METADATA)));
123 }
124 DiscoInfoResult {
125 node: None,
126 identities,
127 features,
128 extensions: vec![],
129 }
130 }
131
132 fn make_initial_presence(disco: &DiscoInfoResult, node: &str) -> Presence {
133 let caps_data = compute_disco(disco);
134 let hash = hash_caps(&caps_data, Algo::Sha_1).unwrap();
135 let caps = Caps::new(node, hash);
136
137 let mut presence = Presence::new(PresenceType::None);
138 presence.add_payload(caps);
139 presence
140 }
141
142 pub fn build(
143 self,
144 ) -> Result<(Agent, impl Stream<Item = Event, Error = tokio_xmpp::Error>), JidParseError> {
145 let client = TokioXmppClient::new(self.jid, self.password)?;
146 Ok(self.build_impl(client))
147 }
148
149 // This function is meant to be used for testing build
150 pub(crate) fn build_impl<S>(
151 self,
152 stream: S,
153 ) -> (Agent, impl Stream<Item = Event, Error = tokio_xmpp::Error>)
154 where
155 S: Stream<Item = tokio_xmpp::Event, Error = tokio_xmpp::Error>
156 + Sink<SinkItem = tokio_xmpp::Packet, SinkError = tokio_xmpp::Error>,
157 {
158 let disco = self.make_disco();
159 let node = self.website;
160 let (sender_tx, sender_rx) = mpsc::unbounded();
161
162 let client = stream;
163 let (sink, stream) = client.split();
164
165 let reader = {
166 let mut sender_tx = sender_tx.clone();
167 let jid = self.jid.to_owned();
168 stream.map(move |event| {
169 // Helper function to send an iq error.
170 let mut events = Vec::new();
171 let send_error = |to, id, type_, condition, text: &str| {
172 let error = StanzaError::new(type_, condition, "en", text);
173 let iq = Iq::from_error(id, error)
174 .with_to(to)
175 .into();
176 sender_tx.unbounded_send(Packet::Stanza(iq)).unwrap();
177 };
178
179 match event {
180 TokioXmppEvent::Online => {
181 let presence = ClientBuilder::make_initial_presence(&disco, &node).into();
182 let packet = Packet::Stanza(presence);
183 sender_tx.unbounded_send(packet)
184 .unwrap();
185 events.push(Event::Online);
186 let iq = Iq::from_get("roster", Roster { ver: None, items: vec![] })
187 .into();
188 sender_tx.unbounded_send(Packet::Stanza(iq)).unwrap();
189 }
190 TokioXmppEvent::Disconnected => {
191 events.push(Event::Disconnected);
192 }
193 TokioXmppEvent::Stanza(stanza) => {
194 if stanza.is("iq", "jabber:client") {
195 let iq = Iq::try_from(stanza).unwrap();
196 if let IqType::Get(payload) = iq.payload {
197 if payload.is("query", ns::DISCO_INFO) {
198 let query = DiscoInfoQuery::try_from(payload);
199 match query {
200 Ok(query) => {
201 let mut disco_info = disco.clone();
202 disco_info.node = query.node;
203 let iq = Iq::from_result(iq.id, Some(disco_info))
204 .with_to(iq.from.unwrap())
205 .into();
206 sender_tx.unbounded_send(Packet::Stanza(iq)).unwrap();
207 },
208 Err(err) => {
209 send_error(iq.from.unwrap(), iq.id, ErrorType::Modify, DefinedCondition::BadRequest, &format!("{}", err));
210 },
211 }
212 } else {
213 // We MUST answer unhandled get iqs with a service-unavailable error.
214 send_error(iq.from.unwrap(), iq.id, ErrorType::Cancel, DefinedCondition::ServiceUnavailable, "No handler defined for this kind of iq.");
215 }
216 } else if let IqType::Result(Some(payload)) = iq.payload {
217 if payload.is("query", ns::ROSTER) {
218 let roster = Roster::try_from(payload).unwrap();
219 for item in roster.items.into_iter() {
220 events.push(Event::ContactAdded(item));
221 }
222 } else if payload.is("pubsub", ns::PUBSUB) {
223 let pubsub = PubSub::try_from(payload).unwrap();
224 let from =
225 iq.from.clone().unwrap_or(Jid::from_str(&jid).unwrap());
226 if let PubSub::Items(items) = pubsub {
227 if items.node.0 == ns::AVATAR_DATA {
228 let new_events = avatar::handle_data_pubsub_iq(&from, &items);
229 events.extend(new_events);
230 }
231 }
232 }
233 } else if let IqType::Set(_) = iq.payload {
234 // We MUST answer unhandled set iqs with a service-unavailable error.
235 send_error(iq.from.unwrap(), iq.id, ErrorType::Cancel, DefinedCondition::ServiceUnavailable, "No handler defined for this kind of iq.");
236 }
237 } else if stanza.is("message", "jabber:client") {
238 let message = Message::try_from(stanza).unwrap();
239 let from = message.from.clone().unwrap();
240 for child in message.payloads {
241 if child.is("event", ns::PUBSUB_EVENT) {
242 let event = PubSubEvent::try_from(child).unwrap();
243 if let PubSubEvent::PublishedItems { node, items } = event {
244 if node.0 == ns::AVATAR_METADATA {
245 avatar::handle_metadata_pubsub_event(&from, &mut sender_tx, items);
246 }
247 }
248 }
249 }
250 } else if stanza.is("presence", "jabber:client") {
251 let presence = Presence::try_from(stanza).unwrap();
252 let from = presence.from.clone().unwrap();
253 for payload in presence.payloads.into_iter() {
254 let muc_user = match MucUser::try_from(payload) {
255 Ok(muc_user) => muc_user,
256 _ => continue
257 };
258 for status in muc_user.status.into_iter() {
259 if status == Status::SelfPresence {
260 events.push(Event::RoomJoined(from.clone()));
261 break;
262 }
263 }
264 }
265 } else if stanza.is("error", "http://etherx.jabber.org/streams") {
266 println!("Received a fatal stream error: {}", String::from(&stanza));
267 } else {
268 panic!("Unknown stanza: {}", String::from(&stanza));
269 }
270 }
271 }
272
273 futures::stream::iter_ok(events)
274 })
275 .flatten()
276 };
277
278 let sender = sender_rx
279 .map_err(|e| panic!("Sink error: {:?}", e))
280 .forward(sink)
281 .map(|(rx, mut sink)| {
282 drop(rx);
283 let _ = sink.close();
284 None
285 });
286
287 // TODO is this correct?
288 // Some(Error) means a real error
289 // None means the end of the sender stream and can be ignored
290 let future = reader
291 .map(Some)
292 .select(sender.into_stream())
293 .filter_map(|x| x);
294
295 let agent = Agent { sender_tx };
296
297 (agent, future)
298 }
299}
300
301pub struct Client {
302 sender_tx: mpsc::UnboundedSender<Packet>,
303 stream: Box<dyn Stream<Item = Event, Error = Error>>,
304}
305
306impl Client {
307 pub fn get_agent(&self) -> Agent {
308 Agent {
309 sender_tx: self.sender_tx.clone(),
310 }
311 }
312
313 pub fn listen(self) -> Box<dyn Stream<Item = Event, Error = Error>> {
314 self.stream
315 }
316}
317
318pub struct Agent {
319 sender_tx: mpsc::UnboundedSender<Packet>,
320}
321
322impl Agent {
323 pub fn join_room(&mut self, room: Jid, lang: &str, status: &str) {
324 let mut presence = Presence::new(PresenceType::None)
325 .with_to(Some(room))
326 .with_payloads(vec![Muc::new().into()]);
327 presence.set_status(String::from(lang), String::from(status));
328 let presence = presence.into();
329 self.sender_tx.unbounded_send(Packet::Stanza(presence))
330 .unwrap();
331 }
332
333 pub fn send_message(&mut self, recipient: Jid, type_: MessageType, lang: &str, text: &str) {
334 let mut message = Message::new(Some(recipient));
335 message.type_ = type_;
336 message.bodies.insert(String::from(lang), Body(String::from(text)));
337 let message = message.into();
338 self.sender_tx.unbounded_send(Packet::Stanza(message))
339 .unwrap();
340 }
341}