1use futures::{future, Future, Sink, Stream};
  2use std::env::args;
  3use std::process::exit;
  4use tokio::runtime::current_thread::Runtime;
  5use tokio_xmpp::{Client, Packet};
  6use xmpp_parsers::{Jid, Element, TryFrom};
  7use xmpp_parsers::message::{Body, Message, MessageType};
  8use xmpp_parsers::presence::{Presence, Show as PresenceShow, Type as PresenceType};
  9
 10fn main() {
 11    let args: Vec<String> = args().collect();
 12    if args.len() != 3 {
 13        println!("Usage: {} <jid> <password>", args[0]);
 14        exit(1);
 15    }
 16    let jid = &args[1];
 17    let password = &args[2];
 18
 19    // tokio_core context
 20    let mut rt = Runtime::new().unwrap();
 21    // Client instance
 22    let client = Client::new(jid, password).unwrap();
 23
 24    // Make the two interfaces for sending and receiving independent
 25    // of each other so we can move one into a closure.
 26    let (sink, stream) = client.split();
 27
 28    // Create outgoing pipe
 29    let (mut tx, rx) = futures::unsync::mpsc::unbounded();
 30    rt.spawn(
 31        rx.forward(
 32            sink.sink_map_err(|_| panic!("Pipe"))
 33        )
 34            .map(|(rx, mut sink)| {
 35                drop(rx);
 36                let _ = sink.close();
 37            })
 38            .map_err(|e| {
 39                panic!("Send error: {:?}", e);
 40            })
 41    );
 42
 43    // Main loop, processes events
 44    let mut wait_for_stream_end = false;
 45    let done = stream.for_each(move |event| {
 46        if wait_for_stream_end {
 47            /* Do nothing */
 48        } else if event.is_online() {
 49            println!("Online!");
 50
 51            let presence = make_presence();
 52            tx.start_send(Packet::Stanza(presence)).unwrap();
 53        } else if let Some(message) = event
 54            .into_stanza()
 55            .and_then(|stanza| Message::try_from(stanza).ok())
 56        {
 57            match (message.from, message.bodies.get("")) {
 58                (Some(ref from), Some(ref body)) if body.0 == "die" => {
 59                    println!("Secret die command triggered by {}", from);
 60                    wait_for_stream_end = true;
 61                    tx.start_send(Packet::StreamEnd).unwrap();
 62                }
 63                (Some(ref from), Some(ref body)) => {
 64                    if message.type_ != MessageType::Error {
 65                        // This is a message we'll echo
 66                        let reply = make_reply(from.clone(), &body.0);
 67                        tx.start_send(Packet::Stanza(reply)).unwrap();
 68                    }
 69                }
 70                _ => {}
 71            }
 72        }
 73
 74        future::ok(())
 75    });
 76
 77    // Start polling `done`
 78    match rt.block_on(done) {
 79        Ok(_) => (),
 80        Err(e) => {
 81            println!("Fatal: {}", e);
 82            ()
 83        }
 84    }
 85}
 86
 87// Construct a <presence/>
 88fn make_presence() -> Element {
 89    let mut presence = Presence::new(PresenceType::None);
 90    presence.show = PresenceShow::Chat;
 91    presence
 92        .statuses
 93        .insert(String::from("en"), String::from("Echoing messages."));
 94    presence.into()
 95}
 96
 97// Construct a chat <message/>
 98fn make_reply(to: Jid, body: &str) -> Element {
 99    let mut message = Message::new(Some(to));
100    message.bodies.insert(String::new(), Body(body.to_owned()));
101    message.into()
102}