Code shared from the Rust Playground

playground.rs

use std::time::Duration;
use std::env;
use anyhow::{bail, Context, Result};
#[cfg(target_os = "macos")]
use cocoa::appkit::NSApplication;
use colibri::{ColibriMessage, Constraints, VideoType};
use glib::ObjectExt;
use gstreamer::{
  prelude::{ElementExt, GstBinExt},
  GhostPad,
};
use gstreamer_app::{
  AppSink
};
use http::Uri;
use lib_gst_meet::{
  init_tracing, Authentication, Connection, JitsiConference, JitsiConferenceConfig, MediaType,
};
use structopt::StructOpt;
use tokio::{signal::ctrl_c, task, time::timeout};
use tracing::{error, info, trace, warn};

#[derive(Debug, Clone, StructOpt)]
#[structopt(
  name = "ripplecreate-bot",
  about = "Connect a GStreamer pipeline to a Jitsi Meet conference."
)]
struct Opt {
  web_socket_url: String,
  xmpp_domain: Option<String>,
  room_name: String,
  muc_domain: Option<String>,
  focus_jid: Option<String>,
  xmpp_username: Option<String>,
  xmpp_password: Option<String>,
  video_codec: String,
  nick: String,
  region: Option<String>,
  send_pipeline: Option<String>,
  recv_pipeline: Option<String>,
  recv_pipeline_participant_template: Option<String>,
  select_endpoints: Option<String>,
  last_n: Option<u16>,
  send_video_height: u16,
  recv_video_scale_width: u16,
  recv_video_scale_height: u16,
  buffer_size: u32,
  start_bitrate: Option<u32>,
  stereo: Option<bool>,
  verbose: u8,
  #[cfg(feature = "tls-insecure")]
  tls_insecure: bool,
  #[cfg(feature = "log-rtp")]
  log_rtp: bool,
  #[cfg(feature = "log-rtp")]
  log_rtcp: bool,
}


#[cfg(not(target_os = "macos"))]
#[tokio::main]
async fn main() -> Result<()> {
  main_inner().await
}


#[cfg(target_os = "macos")]
fn main() {
  // GStreamer requires an NSApp event loop in order for osxvideosink etc to work.
  let app = unsafe { cocoa::appkit::NSApp() };

  let rt = tokio::runtime::Builder::new_multi_thread()
    .enable_all()
    .build()
    .unwrap();

  rt.spawn(async move {
    if let Err(e) = main_inner().await {
      error!("fatal: {:?}", e);
    }
    unsafe {
      cocoa::appkit::NSApp().stop_(cocoa::base::nil);
    }
    std::process::exit(0);
  });

  unsafe {
    app.run();
  }
}


fn init_gstreamer() -> Result<()> {
  trace!("starting gstreamer init");
  gstreamer::init()?;
  trace!("finished gstreamer init");
  Ok(())
}



async fn main_inner() -> Result<()> {

  let args: Vec<String> = env::args().collect();

  let opt = Opt {
    web_socket_url: String::from("wss://meet.ripplecreate.sg/xmpp-websocket"),
    xmpp_domain: <Option<String> as Default>::default(),
    room_name: String::from(&args[1]),
    muc_domain: <Option<String> as Default>::default(),
    focus_jid: <Option<String> as Default>::default(),
    xmpp_username: Some("admin".to_owned()),
    xmpp_password: Some("abc@123".to_owned()),
    video_codec: String::from("vp9"),
    nick: String::from("ripplecreate bot"),
    region: <Option<String> as Default>::default(),
    send_pipeline: <Option<String> as Default>::default(),
    recv_pipeline: <Option<String> as Default>::default(),
    recv_pipeline_participant_template: <Option<String> as Default>::default(),
    select_endpoints: <Option<String> as Default>::default(),
    last_n: None,
    send_video_height: 720,
    recv_video_scale_width: 1280,
    recv_video_scale_height: 720,
    buffer_size: 200,
    start_bitrate: None,
    stereo: <Option<bool> as Default>::default(),
    verbose: 0

  };
  
  init_tracing(match opt.verbose {
    0 => tracing::Level::INFO,
    1 => tracing::Level::DEBUG,
    _ => tracing::Level::TRACE,
  });
  glib::log_set_default_handler(glib::rust_log_handler);

  init_gstreamer()?;

  let web_socket_url: Uri = opt.web_socket_url.parse()?;

  let xmpp_domain = opt
    .xmpp_domain
    .as_deref()
    .or_else(|| web_socket_url.host())
    .context("invalid WebSocket URL")?;

  let (connection, background) = Connection::new(
    &opt.web_socket_url,
    xmpp_domain,
    Authentication::Plain {
      username: String::from("admin"),
      password: String::from("abc@123"),
    },
    #[cfg(feature = "tls-insecure")]
    opt.tls_insecure,
    #[cfg(not(feature = "tls-insecure"))]
    false,
  )
  .await
  .context("failed to build connection")?;

  tokio::spawn(background);

  connection.connect().await?;

  let room_jid = format!(
    "{}@{}",
    opt.room_name,
    opt
      .muc_domain
      .clone()
      .unwrap_or_else(|| { format!("conference.{}", xmpp_domain) }),
  );

  let focus_jid = opt
    .focus_jid
    .clone()
    .unwrap_or_else(|| format!("focus@auth.{}/focus", xmpp_domain));

  let Opt {
    nick,
    region,
    video_codec,
    recv_pipeline_participant_template,
    send_video_height,
    recv_video_scale_height,
    recv_video_scale_width,
    buffer_size,
    start_bitrate,
    stereo,
    #[cfg(feature = "log-rtp")]
    log_rtp,
    #[cfg(feature = "log-rtp")]
    log_rtcp,
    ..
  } = opt;

  let config = JitsiConferenceConfig {
    muc: room_jid.parse()?,
    focus: focus_jid.parse()?,
    nick,
    region,
    video_codec,
    extra_muc_features: vec![],
    start_bitrate: start_bitrate.unwrap_or(800),
    stereo: stereo.unwrap_or_default(),
    recv_video_scale_height,
    recv_video_scale_width,
    buffer_size,
    #[cfg(feature = "log-rtp")]
    log_rtp,
    #[cfg(feature = "log-rtp")]
    log_rtcp,
  };

  let main_loop = glib::MainLoop::new(None, false);

  let conference = JitsiConference::join(connection, main_loop.context(), config)
    .await
    .context("failed to join conference")?;

  conference
    .set_send_resolution(send_video_height.into())
    .await;
    

  conference
    .on_participant_left(move |_conference, participant| {
      Box::pin(async move {
        info!("Participant left: {:?}", participant);
        Ok(())
      })
    })
    .await;

  conference
    .set_pipeline_state(gstreamer::State::Playing)
    .await?;

  let conference_ = conference.clone();
  let main_loop_ = main_loop.clone();
  tokio::spawn(async move {
    ctrl_c().await.unwrap();

    info!("Exiting...");

    match timeout(Duration::from_secs(10), conference_.leave()).await {
      Ok(Ok(_)) => {},
      Ok(Err(e)) => warn!("Error leaving conference: {:?}", e),
      Err(_) => warn!("Timed out leaving conference"),
    }

    main_loop_.quit();
  });

  task::spawn_blocking(move || main_loop.run()).await?;

  Ok(())
}