implement jabber

Cargo.toml

@@ -4,9 +4,12 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+futures = "0.3"
 tokio = { version = "1", features = ["full"] }
 reqwest = { version = "0.11", features = ["json"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 geo = "0.18"
 csv = "1.1"
+tokio-xmpp = "3"
+xmpp-parsers = "0.18"

src/adsb.rs

@@ -7,7 +7,7 @@ use tokio::sync::mpsc::{channel, Receiver};
 use super::location::Locations;
 
 /// ft
-const MAX_ALTITUDE: u32 = 50000; //1800;
+const MAX_ALTITUDE: u32 = 5000;
 /// s
 const STATE_TIMEOUT: u64 = 180;
 
@@ -49,9 +49,9 @@ impl Info {
         self.altitude as f64 * 0.3048
     }
 
-    pub fn get_speed_kph(&self) -> f64 {
-        self.speed as f64 * 1.852
-    }
+    // pub fn get_speed_kph(&self) -> f64 {
+    //     self.speed as f64 * 1.852
+    // }
 }
 
 #[derive(Debug, Clone)]
@@ -61,7 +61,7 @@ pub struct Event {
     pub location: Arc<String>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub enum Action {
     Appeared,
     Disappeared,
@@ -137,7 +137,7 @@ async fn fetch(url: &str) -> Result<Vec<Info>> {
     Ok(result)
 }
 
-pub async fn run(url: &'static str, locations: Locations) -> Receiver<Event> {
+pub fn run(url: &'static str, locations: Locations) -> Receiver<Event> {
     let (tx, rx) = channel(1);
 
     tokio::spawn(async move {

src/aircrafts.rs

@@ -21,7 +21,9 @@ impl Aircrafts {
         let mut data = HashMap::new();
         for result in rdr.deserialize() {
             let aircraft: Aircraft = result.unwrap();
-            data.insert(aircraft.icao24.clone(), aircraft);
+            if aircraft.registration != "" {
+                data.insert(aircraft.icao24.clone(), aircraft);
+            }
         }
         println!("Loaded aircraft database with {} entries", data.len());
 

src/jabber.rs

@@ -0,0 +1,96 @@
+use std::str::FromStr;
+use std::convert::TryFrom;
+use futures::{SinkExt, StreamExt};
+use tokio::sync::mpsc;
+use tokio_xmpp::{AsyncClient, Packet, Event};
+use xmpp_parsers::{Jid, BareJid, FullJid, Element};
+use xmpp_parsers::message::{Body, Message, MessageType};
+use xmpp_parsers::presence::{Presence, Show as PresenceShow, Type as PresenceType};
+use xmpp_parsers::muc::Muc;
+
+#[derive(Clone)]
+pub struct Handle {
+    room_jid: BareJid,
+    tx: mpsc::Sender<Packet>,
+}
+
+impl Handle {
+    pub async fn send_message(&self, msg: String) {
+        let stanza = make_room_message(self.room_jid.clone(), msg);
+        self.tx.send(Packet::Stanza(stanza)).await.unwrap();
+    }
+}
+
+pub fn run(jid: String, password: String, muc_jid: String) -> Handle {
+    let muc_jid: FullJid = match FullJid::from_str(&muc_jid) {
+        Ok(jid) => jid,
+        Err(err) => panic!("MUC Jid invalid: {:?}", err),
+    };
+    let (tx, mut rx) = mpsc::channel(1);
+    let handle = Handle {
+        room_jid: muc_jid.clone().into(),
+        tx: tx.clone(),
+    };
+
+    let client = AsyncClient::new(&jid, &password).unwrap();
+    let (mut sink, mut stream) = client.split();
+
+    tokio::task::spawn_local(async move {
+        while let Some(event) = stream.next().await {
+            match event {
+                Event::Online { .. } => {
+                    println!("XMPP client now online at {}", jid);
+                    let packet = Packet::Stanza(make_join_presence(muc_jid.clone()));
+                    tx.send(packet).await
+                        .unwrap();
+                }
+                Event::Stanza(el) => {
+                    println!("<< {:?}", el);
+                    if el.is("presence", "jabber:client") {
+                        match Presence::try_from(el) {
+                            Ok(presence) => {
+                                if presence.from == Some(Jid::Full(muc_jid.clone())) {
+                                    if presence.type_ == PresenceType::Error {
+                                        println!("Failed to enter MUC {:?}", muc_jid);
+                                    } else {
+                                        println!("Entered MUC {:?}", muc_jid);
+                                    }
+                                }
+                            },
+                            Err(err) => println!("Received invalid presence: {:?}", err),
+                        }
+                    }
+                }
+                ev => println!("<< {:?}", ev),
+                _ => {}
+            }
+        }
+
+        panic!("XMPP end")
+    });
+    tokio::task::spawn_local(async move {
+        while let Some(packet) = rx.recv().await {
+            sink.send(packet).await.unwrap();
+        }
+
+        panic!("channel end")
+    });
+
+    handle
+}
+
+pub fn make_room_message(room_jid: BareJid, text: String) -> Element {
+    let mut message = Message::new(Some(Jid::Bare(room_jid)));
+    message.type_ = MessageType::Groupchat;
+    message.bodies.insert("de".to_string(), Body(text));
+    message.into()
+}
+
+fn make_join_presence(muc_jid: FullJid) -> Element {
+    let mut presence = Presence::new(PresenceType::None)
+        .with_to(Jid::Full(muc_jid))
+        .with_show(PresenceShow::Dnd);
+    presence.set_status("de".to_string(), "Augen und Ohren nach oben".to_string());
+    presence.add_payload(Muc::new());
+    presence.into()
+}

src/main.rs

@@ -1,19 +1,73 @@
+use std::env::args;
+
 mod adsb;
 mod location;
 mod aircrafts;
+mod jabber;
+
+#[derive(Debug, Clone, Copy)]
+pub struct UsageError;
+
+impl std::fmt::Display for UsageError {
+    fn fmt(&self, fmt: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> {
+        write!(fmt, "Usage error")
+    }
+}
+
+impl std::error::Error for UsageError {}
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    let aircrafts = aircrafts::Aircrafts::load("aircraftDatabase.csv");
-    let locations = location::Locations::load("locations.json");
-    let mut events = adsb::run("https://adsb.hq.c3d2.de/data.json", locations).await;
-
-    while let Some(event) = events.recv().await {
-        println!("event: {:?}", event);
-        if let Some(aircraft) = aircrafts.find(&event.info.get_hex()) {
-            println!("aircraft: {:?}", aircraft);
+    tokio::task::LocalSet::new().run_until(async move {
+        let args: Vec<String> = args().collect();
+        if args.len() != 4 {
+            println!("Usage: {} <jid> <password> <muc>", args[0]);
+            Err(UsageError)?;
         }
-    }
+        let jid = args[1].to_owned();
+        let password = args[2].to_owned();
+        let muc = args[3].to_owned();
 
-    Ok(())
+        let aircrafts = aircrafts::Aircrafts::load("aircraftDatabase.csv");
+        let locations = location::Locations::load("locations.json");
+        let mut events = adsb::run("https://adsb.hq.c3d2.de/data.json", locations);
+
+        let jabber = jabber::run(jid, password, muc);
+
+        println!("wait for events...");
+        while let Some(event) = events.recv().await {
+            println!("event: {:?}", event);
+
+            let mut text;
+            if let Some(aircraft) = aircrafts.find(&event.info.get_hex()) {
+                println!("aircraft: {:?}", aircraft);
+                text = format!("{} {} ({} {})",
+                               aircraft.owner, aircraft.registration,
+                               aircraft.manufacturername, aircraft.model
+                );
+            } else if let Some(flight) = event.info.get_flight() {
+                text = format!("Flug {} [{}]", flight, event.info.get_hex());
+            } else {
+                text = format!("[{}]", event.info.get_hex());
+            }
+
+            text = format!("{} {}", text, match event.action {
+                adsb::Action::Appeared => "ist aufgetaucht",
+                adsb::Action::Moved => "fliegt jetzt",
+                adsb::Action::Disappeared => "ist abgetaucht.",
+            });
+            if event.action != adsb::Action::Disappeared {
+                text = format!("{} {:.0}m über {}", text,
+                               event.info.get_altitude_m(),
+                               event.location
+                );
+            }
+
+            println!(">> {}", text);
+            jabber.send_message(text).await;
+        }
+
+        let result: Result<(), Box<dyn std::error::Error + Send + Sync>> = Ok(());
+        result
+    }).await
 }