Implement XMPP pings (client -> server)

lib-gst-meet/src/pinger.rs

@@ -1,16 +1,40 @@
-use std::convert::TryFrom;
+use std::{convert::TryFrom, time::Duration};
 
 use anyhow::{anyhow, Result};
 use async_trait::async_trait;
-use tokio::sync::mpsc;
+use tokio::{sync::mpsc, task::JoinHandle, time};
-use xmpp_parsers::{iq::Iq, Element, FullJid, Jid};
+use tracing::warn;
+use xmpp_parsers::{iq::Iq, Element, FullJid, Jid, ping::Ping};
 
-use crate::stanza_filter::StanzaFilter;
+use crate::{stanza_filter::StanzaFilter, util::generate_id};
+
+const PING_INTERVAL: Duration = Duration::from_secs(30);
 
 #[derive(Debug)]
 pub(crate) struct Pinger {
   pub(crate) jid: FullJid,
   pub(crate) tx: mpsc::Sender<Element>,
+  pub(crate) ping_task: JoinHandle<()>,
+}
+
+impl Pinger {
+  pub(crate) fn new(jid: FullJid, tx: mpsc::Sender<Element>) -> Pinger {
+    let ping_tx = tx.clone();
+    let ping_task = tokio::spawn(async move {
+      let mut interval = time::interval(PING_INTERVAL);
+      loop {
+        interval.tick().await;
+        if let Err(e) = ping_tx.send(Iq::from_get(generate_id(), Ping).into()).await {
+          warn!("failed to send XMPP ping: {:?}", e);
+        }
+      }
+    });
+    Pinger {
+      jid,
+      tx,
+      ping_task,
+    }
+  }
 }
 
 #[async_trait]

lib-gst-meet/src/xmpp/connection.rs

@@ -269,10 +269,7 @@ impl Connection {
             info!("My JID: {}", jid);
             locked_inner.jid = Some(jid.clone());
 
-            locked_inner.stanza_filters.push(Box::new(Pinger {
+            locked_inner.stanza_filters.push(Box::new(Pinger::new(jid.clone(), tx.clone())));
-              jid: jid.clone(),
-              tx: tx.clone(),
-            }));
 
             let iq = Iq::from_get(generate_id(), DiscoInfoQuery { node: None })
               .with_from(Jid::Full(jid.clone()))