gst-meet/lib-gst-meet/src/jingle.rs

use std::{collections::HashMap, fmt, net::SocketAddr};

use anyhow::{anyhow, bail, Context, Result};
use futures::stream::StreamExt;
use glib::{Cast, ObjectExt, ToValue};
use gstreamer::prelude::{ElementExt, GObjectExtManualGst, GstBinExt, PadExt};
use nice_gst_meet as nice;
use pem::Pem;
use rand::random;
use rcgen::{Certificate, CertificateParams, PKCS_ECDSA_P256_SHA256};
use ring::digest::{digest, SHA256};
use tokio::{
  net::lookup_host,
  runtime::Handle,
  sync::oneshot,
};
use tracing::{debug, error, warn};
use uuid::Uuid;
use xmpp_parsers::{
  hashes::Algo,
  iq::Iq,
  jingle::{Action, Content, Creator, Description, Jingle, Senders, Transport},
  jingle_dtls_srtp::{Fingerprint, Setup},
  jingle_ice_udp::{self, Transport as IceUdpTransport},
  jingle_rtp::{Description as RtpDescription, PayloadType, RtcpMux},
  jingle_ssma::{self, Parameter},
  Jid,
};

use crate::{
  colibri::ColibriChannel,
  conference::JitsiConference,
  source::{MediaType, Source},
  util::generate_id,
};

const DEFAULT_STUN_PORT: u16 = 3478;
const DEFAULT_TURNS_PORT: u16 = 5349;

pub(crate) struct JingleSession {
  pipeline: gstreamer::Pipeline,
  audio_sink_element: gstreamer::Element,
  video_sink_element: gstreamer::Element,
  remote_ssrc_map: HashMap<u32, Source>,
  ice_agent: nice::Agent,
  ice_stream_id: u32,
  ice_component_id: u32,
  pub(crate) accept_iq_id: Option<String>,
  pub(crate) colibri_url: Option<String>,
  pub(crate) colibri_channel: Option<ColibriChannel>,
  pipeline_state_null_rx: oneshot::Receiver<()>,
}

impl fmt::Debug for JingleSession {
  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    f.debug_struct("JingleSession").finish()
  }
}

impl JingleSession {
  pub(crate) fn pipeline(&self) -> gstreamer::Pipeline {
    self.pipeline.clone()
  }

  pub(crate) fn audio_sink_element(&self) -> gstreamer::Element {
    self.audio_sink_element.clone()
  }

  pub(crate) fn video_sink_element(&self) -> gstreamer::Element {
    self.video_sink_element.clone()
  }

  pub(crate) fn pause_all_sinks(&self) {
    if let Some(rtpbin) = self.pipeline.by_name("rtpbin") {
      rtpbin.foreach_src_pad(|_, pad| {
        let pad_name: String = pad.property("name").unwrap().get().unwrap();
        if pad_name.starts_with("recv_rtp_src_0_") {
          if let Some(peer_pad) = pad.peer() {
            if let Some(element) = peer_pad.parent_element() {
              element.set_state(gstreamer::State::Paused).unwrap();
            }
          }
        }
        true
      });
    }
  }

  pub(crate) async fn pipeline_stopped(self) -> Result<()> {
    Ok(self.pipeline_state_null_rx.await?)
  }

  pub(crate) async fn initiate(conference: &JitsiConference, jingle: Jingle) -> Result<Self> {
    let initiator = jingle
      .initiator
      .as_ref()
      .ok_or_else(|| anyhow!("session-initiate with no initiator"))?
      .clone();

    debug!("Received Jingle session-initiate from {}", initiator);

    let mut ice_remote_candidates = None;
    let mut ice_remote_ufrag = None;
    let mut ice_remote_pwd = None;
    let mut dtls_fingerprint = None;
    let mut opus_payload_type = None;
    let mut h264_payload_type = None;
    let mut vp8_payload_type = None;
    let mut vp9_payload_type = None;
    let mut colibri_url = None;

    let mut remote_ssrc_map = HashMap::new();

    for content in &jingle.contents {
      if let Some(Description::Rtp(description)) = &content.description {
        if description.media == "audio" {
          opus_payload_type = description
            .payload_types
            .iter()
            .find(|pt| pt.name.as_deref() == Some("opus"))
            .map(|pt| pt.id);
        }
        else if description.media == "video" {
          h264_payload_type = description
            .payload_types
            .iter()
            .find(|pt| pt.name.as_deref() == Some("H264"))
            .map(|pt| pt.id);
          vp8_payload_type = description
            .payload_types
            .iter()
            .find(|pt| pt.name.as_deref() == Some("VP8"))
            .map(|pt| pt.id);
          vp9_payload_type = description
            .payload_types
            .iter()
            .find(|pt| pt.name.as_deref() == Some("VP9"))
            .map(|pt| pt.id);
        }
        else {
          continue;
        }

        for ssrc in &description.ssrcs {
          let owner = ssrc
            .info
            .as_ref()
            .context("missing ssrc-info")?
            .owner
            .clone();
          if owner == "jvb" {
            debug!("skipping ssrc (owner = jvb)");
            continue;
          }

          remote_ssrc_map.insert(
            ssrc.id.parse()?,
            Source {
              ssrc: ssrc.id.parse()?,
              participant_id: owner
                .split('/')
                .nth(1)
                .context("invalid ssrc-info owner")?
                .to_owned(),
              media_type: if description.media == "audio" {
                MediaType::Audio
              }
              else {
                MediaType::Video
              },
            },
          );
        }
      }

      if let Some(Transport::IceUdp(transport)) = &content.transport {
        if !transport.candidates.is_empty() {
          ice_remote_candidates = Some(transport.candidates.clone());
        }
        if let Some(ufrag) = &transport.ufrag {
          ice_remote_ufrag = Some(ufrag.to_owned());
        }
        if let Some(pwd) = &transport.pwd {
          ice_remote_pwd = Some(pwd.to_owned());
        }
        if let Some(fingerprint) = &transport.fingerprint {
          if fingerprint.hash != Algo::Sha_256 {
            bail!("unsupported fingerprint hash: {:?}", fingerprint.hash);
          }
          dtls_fingerprint = Some(fingerprint.value.clone());
        }
        if let Some(websocket) = &transport.web_socket {
          colibri_url = Some(websocket.url.clone());
        }
      }
    }

    if let Some(remote_fingerprint) = dtls_fingerprint {
      warn!(
        "Remote DTLS fingerprint (verification not implemented yet): {:?}",
        remote_fingerprint
      );
    }

    let mut dtls_cert_params = CertificateParams::new(vec!["gst-meet".to_owned()]);
    dtls_cert_params.alg = &PKCS_ECDSA_P256_SHA256;
    let dtls_cert = Certificate::from_params(dtls_cert_params)?;
    let dtls_cert_der = dtls_cert.serialize_der()?;
    let fingerprint = digest(&SHA256, &dtls_cert_der).as_ref().to_vec();
    let fingerprint_str =
      itertools::join(fingerprint.iter().map(|byte| format!("{:X}", byte)), ":");
    let dtls_cert_pem = pem::encode(&Pem {
      tag: "CERTIFICATE".to_string(),
      contents: dtls_cert_der,
    });
    let dtls_private_key_pem = pem::encode(&Pem {
      tag: "PRIVATE KEY".to_string(),
      contents: dtls_cert.serialize_private_key_der(),
    });
    debug!("Local DTLS certificate:\n{}", dtls_cert_pem);
    debug!("Local DTLS fingerprint: {}", fingerprint_str);

    let audio_ssrc: u32 = random();
    let video_ssrc: u32 = random();

    debug!("audio SSRC: {}", audio_ssrc);
    debug!("video SSRC: {}", video_ssrc);

    let ice_agent = nice::Agent::new(&conference.glib_main_context, nice::Compatibility::Rfc5245);
    ice_agent.set_ice_tcp(false);
    ice_agent.set_upnp(false);
    let ice_stream_id = ice_agent.add_stream(1);
    let ice_component_id = 1;

    let maybe_stun = conference
      .external_services
      .iter()
      .find(|svc| svc.r#type == "stun");

    let stun_addr = if let Some(stun) = maybe_stun {
      lookup_host(format!(
        "{}:{}",
        stun.host,
        stun.port.unwrap_or(DEFAULT_STUN_PORT)
      ))
      .await?
      .next()
    }
    else {
      None
    };
    debug!("STUN address: {:?}", stun_addr);

    if let Some((stun_addr, stun_port)) = stun_addr.map(|sa| (sa.ip().to_string(), sa.port())) {
      ice_agent.set_stun_server(Some(&stun_addr));
      ice_agent.set_stun_server_port(stun_port as u32);
    }

    let maybe_turn = conference
      .external_services
      .iter()
      .find(|svc| svc.r#type == "turns");

    if let Some(turn_server) = maybe_turn {
      let maybe_addr = lookup_host(format!(
        "{}:{}",
        turn_server.host,
        turn_server.port.unwrap_or(DEFAULT_TURNS_PORT)
      ))
      .await?
      .next();

      if let Some(addr) = maybe_addr {
        debug!("TURN address: {:?}", addr);
        ice_agent.set_relay_info(
          ice_stream_id,
          ice_component_id,
          &addr.ip().to_string(),
          addr.port() as u32,
          turn_server.username.as_deref().unwrap_or_default(),
          turn_server.password.as_deref().unwrap_or_default(),
          nice::RelayType::Tls,
        );
      }
    }

    if !ice_agent.attach_recv(
      ice_stream_id,
      ice_component_id,
      &conference.glib_main_context,
      |_, _, _, s| debug!("ICE nice_agent_attach_recv cb: {}", s),
    ) {
      warn!("nice_agent_attach_recv failed");
    }

    debug!("ice_agent={:?}", ice_agent);
    debug!("ice_stream_id={}", ice_stream_id);
    debug!("ice_component_id={}", ice_component_id);

    let (ice_local_ufrag, ice_local_pwd) = ice_agent
      .local_credentials(ice_stream_id)
      .context("no local ICE credentials")?;

    if let (Some(ufrag), Some(pwd)) = (&ice_remote_ufrag, &ice_remote_pwd) {
      debug!("setting ICE remote credentials");
      if !ice_agent.set_remote_credentials(ice_stream_id, ufrag, pwd) {
        warn!("nice_agent_set_remote_candidates failed");
      }
    }

    ice_agent.connect_candidate_gathering_done(move |_agent, candidates| {
      debug!("ICE candidate-gathering-done {:?}", candidates);
    });

    debug!("gathering ICE candidates");
    if !ice_agent.gather_candidates(ice_stream_id) {
      warn!("nice_agent_gather_candidates failed");
    }

    if let (Some(ufrag), Some(pwd), Some(remote_candidates)) =
      (&ice_remote_ufrag, &ice_remote_pwd, &ice_remote_candidates)
    {
      debug!("setting ICE remote candidates: {:?}", remote_candidates);
      let remote_candidates: Vec<_> = remote_candidates
        .iter()
        .map(|c| {
          let mut candidate = nice::Candidate::new(match c.type_ {
            jingle_ice_udp::Type::Host => nice::CandidateType::Host,
            jingle_ice_udp::Type::Prflx => nice::CandidateType::PeerReflexive,
            jingle_ice_udp::Type::Srflx => nice::CandidateType::ServerReflexive,
            jingle_ice_udp::Type::Relay => nice::CandidateType::Relayed,
          });
          candidate.set_stream_id(ice_stream_id);
          candidate.set_component_id(c.component as u32);
          candidate.set_foundation(&c.foundation);
          candidate.set_addr(SocketAddr::new(c.ip, c.port));
          candidate.set_priority(c.priority);
          candidate.set_username(ufrag);
          candidate.set_password(pwd);
          debug!("candidate: {:?}", candidate);
          candidate
        })
        .collect();
      let candidate_refs: Vec<_> = remote_candidates.iter().collect();
      let res = ice_agent.set_remote_candidates(ice_stream_id, ice_component_id, &candidate_refs);
      if res < remote_candidates.len() as i32 {
        warn!("some remote candidates failed to add: {}", res);
      }
    }

    let pipeline_spec = format!(
      r#"
        rtpbin rtp-profile=savpf name=rtpbin

        nicesrc stream={0} component={1} name=nicesrc ! dtlssrtpdec name=dtlssrtpdec connection-id=gst-meet
        dtlssrtpenc name=dtlssrtpenc connection-id=gst-meet is-client=true ! nicesink stream={0} component={1} name=nicesink

        rtpbin.send_rtp_src_0 ! dtlssrtpenc.rtp_sink_0
        rtpbin.send_rtcp_src_0 ! dtlssrtpenc.rtcp_sink_0
        rtpbin.send_rtp_src_1 ! dtlssrtpenc.rtp_sink_1
        rtpbin.send_rtcp_src_1 ! dtlssrtpenc.rtcp_sink_1

        dtlssrtpdec.rtp_src ! rtpbin.recv_rtp_sink_0
        dtlssrtpdec.rtcp_src ! rtpbin.recv_rtcp_sink_0
      "#,
      ice_stream_id, ice_component_id,
    );

    debug!("building gstreamer pipeline:\n{}", pipeline_spec);

    let pipeline = gstreamer::parse_launch(&pipeline_spec)?
      .downcast::<gstreamer::Pipeline>()
      .map_err(|_| anyhow!("pipeline did not parse as a pipeline"))?;

    let rtpbin = pipeline
      .by_name("rtpbin")
      .context("no rtpbin in pipeline")?;

    rtpbin.connect("request-pt-map", false, move |values| {
      let f = || {
        debug!("rtpbin request-pt-map {:?}", values);
        let pt = values[2].get::<u32>()? as u8;
        let maybe_caps = if Some(pt) == opus_payload_type {
          Some(gstreamer::Caps::new_simple(
            "application/x-rtp",
            &[
              ("media", &"audio"),
              ("encoding-name", &"OPUS"),
              ("clock-rate", &48000),
            ],
          ))
        }
        else if Some(pt) == h264_payload_type {
          Some(gstreamer::Caps::new_simple(
            "application/x-rtp",
            &[
              ("media", &"video"),
              ("encoding-name", &"H264"),
              ("clock-rate", &90000),
            ],
          ))
        }
        else if Some(pt) == vp8_payload_type {
          Some(gstreamer::Caps::new_simple(
            "application/x-rtp",
            &[
              ("media", &"video"),
              ("encoding-name", &"VP8"),
              ("clock-rate", &90000),
            ],
          ))
        }
        else if Some(pt) == vp9_payload_type {
          Some(gstreamer::Caps::new_simple(
            "application/x-rtp",
            &[
              ("media", &"video"),
              ("encoding-name", &"VP9"),
              ("clock-rate", &90000),
            ],
          ))
        }
        else {
          warn!("unknown payload type: {}", pt);
          None
        };
        Ok::<_, anyhow::Error>(maybe_caps)
      };
      match f() {
        Ok(Some(caps)) => {
          debug!("mapped pt to caps: {:?}", caps);
          Some(caps.to_value())
        },
        Ok(None) => None,
        Err(e) => {
          error!("handling request-pt-map: {:?}", e);
          None
        },
      }
    })?;

    let handle = Handle::current();
    let inner_ = conference.inner.clone();
    let pipeline_ = pipeline.clone();
    let rtpbin_ = rtpbin.clone();
    rtpbin.connect("pad-added", false, move |values| {
      let inner_ = inner_.clone();
      let handle = handle.clone();
      let pipeline_ = pipeline_.clone();
      let rtpbin_ = rtpbin_.clone();
      let f = move || {
        debug!("rtpbin pad-added {:?}", values);
        let pad: gstreamer::Pad = values[1].get()?;
        let pad_name: String = pad.property("name")?.get()?;
        if pad_name.starts_with("recv_rtp_src_0_") {
          let mut parts = pad_name.split('_').skip(4);
          let ssrc: u32 = parts.next().context("malformed pad name")?.parse()?;
          let pt: u8 = parts.next().context("malformed pad name")?.parse()?;
          let source = handle.block_on(async move {
            let locked_inner = inner_.lock().await;
            let jingle_session = locked_inner
              .jingle_session
              .as_ref()
              .context("not connected (no jingle session)")?;
            Ok::<_, anyhow::Error>(
              jingle_session
                .remote_ssrc_map
                .get(&ssrc)
                .context(format!("unknown ssrc: {}", ssrc))?
                .clone(),
            )
          })?;

          debug!("pad added for remote source: {:?}", source);

          let element_name = match source.media_type {
            MediaType::Audio => {
              if Some(pt) == opus_payload_type {
                "rtpopusdepay"
              }
              else {
                bail!("received audio with unsupported PT {}", pt);
              }
            },
            MediaType::Video => {
              if Some(pt) == h264_payload_type {
                "rtph264depay"
              }
              else if Some(pt) == vp8_payload_type {
                "rtpvp8depay"
              }
              else if Some(pt) == vp9_payload_type {
                "rtpvp9depay"
              }
              else {
                bail!("received video with unsupported PT {}", pt);
              }
            },
          };

          let source_element = gstreamer::ElementFactory::make(element_name, None)?;
          pipeline_
            .add(&source_element)
            .context(format!("failed to add {} to pipeline", element_name))?;
          source_element.sync_state_with_parent()?;
          debug!("created {} element", element_name);
          rtpbin_
            .link_pads(Some(&pad_name), &source_element, None)
            .context(format!(
              "failed to link rtpbin.{} to {}",
              pad_name, element_name
            ))?;
          debug!("linked rtpbin.{} to {}", pad_name, element_name);

          let src_pad = source_element
            .static_pad("src")
            .context("depayloader has no src pad")?;

          if let Some(participant_bin) =
            pipeline_.by_name(&format!("participant_{}", source.participant_id))
          {
            let sink_pad_name = match source.media_type {
              MediaType::Audio => "audio",
              MediaType::Video => "video",
            };
            if let Some(sink_pad) = participant_bin.static_pad(sink_pad_name) {
              debug!("linking depayloader to participant bin");
              src_pad.link(&sink_pad)?;
            }
            else {
              warn!(
                "no {} sink pad in {} participant bin",
                sink_pad_name, source.participant_id
              );
            }
          }
          else {
            debug!("no participant bin for {}", source.participant_id);
          }

          if !src_pad.is_linked() {
            debug!("nothing linked to {}, adding fakesink", element_name);
            let fakesink = gstreamer::ElementFactory::make("fakesink", None)?;
            pipeline_.add(&fakesink)?;
            fakesink.sync_state_with_parent()?;
            source_element.link(&fakesink)?;
          }

          gstreamer::debug_bin_to_dot_file(
            &pipeline_,
            gstreamer::DebugGraphDetails::ALL,
            &format!("ssrc-added-{}", ssrc),
          );

          Ok::<_, anyhow::Error>(())
        }
        else {
          Ok(())
        }
      };
      if let Err(e) = f() {
        error!("handling pad-added: {:?}", e);
      }
      None
    })?;

    let audio_sink_element = gstreamer::ElementFactory::make("rtpopuspay", None)?;
    audio_sink_element.set_property(
      "pt",
      opus_payload_type.context("no opus payload type in jingle session-initiate")? as u32,
    )?;
    audio_sink_element.set_property("min-ptime", 10i64 * 1000 * 1000)?;
    audio_sink_element.set_property("ssrc", audio_ssrc)?;
    pipeline.add(&audio_sink_element)?;
    audio_sink_element.link_pads(None, &rtpbin, Some("send_rtp_sink_0"))?;

    let video_sink_element = match conference.config.video_codec.as_str() {
      "h264" => {
        let element = gstreamer::ElementFactory::make("rtph264pay", None)?;
        element.set_property(
          "pt",
          h264_payload_type.context("no h264 payload type in jingle session-initiate")? as u32,
        )?;
        element.set_property_from_str("aggregate-mode", "zero-latency");
        element
      },
      "vp8" => {
        let element = gstreamer::ElementFactory::make("rtpvp8pay", None)?;
        element.set_property(
          "pt",
          vp8_payload_type.context("no vp8 payload type in jingle session-initiate")? as u32,
        )?;
        element.set_property_from_str("picture-id-mode", "15-bit");
        element
      },
      "vp9" => {
        let element = gstreamer::ElementFactory::make("rtpvp9pay", None)?;
        element.set_property(
          "pt",
          vp9_payload_type.context("no vp9 payload type in jingle session-initiate")? as u32,
        )?;
        element.set_property_from_str("picture-id-mode", "15-bit");
        element
      },
      other => bail!("unsupported video codec: {}", other),
    };
    video_sink_element.set_property("ssrc", video_ssrc)?;
    pipeline.add(&video_sink_element)?;
    video_sink_element.link_pads(None, &rtpbin, Some("send_rtp_sink_1"))?;

    let dtlssrtpdec = pipeline
      .by_name("dtlssrtpdec")
      .context("no dtlssrtpdec in pipeline")?;
    dtlssrtpdec.set_property(
      "pem",
      format!("{}\n{}", dtls_cert_pem, dtls_private_key_pem),
    )?;

    let nicesrc = pipeline
      .by_name("nicesrc")
      .context("no nicesrc in pipeline")?;
    nicesrc.set_property("agent", &ice_agent)?;

    let nicesink = pipeline
      .by_name("nicesink")
      .context("no nicesink in pipeline")?;
    nicesink.set_property("agent", &ice_agent)?;

    let bus = pipeline.bus().context("failed to get pipeline bus")?;

    let (pipeline_state_null_tx, pipeline_state_null_rx) = oneshot::channel();
    tokio::spawn(async move {
      let mut stream = bus.stream();
      while let Some(msg) = stream.next().await {
        match msg.view() {
          gstreamer::MessageView::Error(e) => {
            if let Some(d) = e.debug() {
              error!("{}", d);
            }
          },
          gstreamer::MessageView::Warning(e) => {
            if let Some(d) = e.debug() {
              warn!("{}", d);
            }
          },
          gstreamer::MessageView::StateChanged(state)
            if state.current() == gstreamer::State::Null =>
          {
            debug!("pipeline state is null");
            pipeline_state_null_tx.send(()).unwrap();
            break;
          }
          _ => {},
        }
      }
    });

    gstreamer::debug_bin_to_dot_file(
      &pipeline,
      gstreamer::DebugGraphDetails::ALL,
      "session-initiate",
    );

    let local_candidates = ice_agent.local_candidates(ice_stream_id, ice_component_id);
    debug!("local candidates: {:?}", local_candidates);

    debug!("building Jingle session-accept");
    let mut jingle_accept = Jingle::new(Action::SessionAccept, jingle.sid.clone())
      .with_initiator(
        jingle
          .initiator
          .as_ref()
          .context("jingle session-initiate with no initiator")?
          .clone(),
      )
      .with_responder(Jid::Full(conference.jid.clone()));

    for initiate_content in &jingle.contents {
      let mut description = RtpDescription::new(initiate_content.name.0.clone());

      description.payload_types = if initiate_content.name.0 == "audio" {
        vec![PayloadType::new(
          opus_payload_type.context("no opus payload type in jingle session-initiate")?,
          "opus".to_owned(),
          48000,
          2,
        )]
      }
      else {
        match conference.config.video_codec.as_str() {
          "h264" => vec![PayloadType::new(
            h264_payload_type.context("no h264 payload type in jingle session-initiate")?,
            "H264".to_owned(),
            90000,
            1,
          )],
          "vp8" => vec![PayloadType::new(
            vp8_payload_type.context("no vp8 payload type in jingle session-initiate")?,
            "VP8".to_owned(),
            90000,
            1,
          )],
          "vp9" => vec![PayloadType::new(
            vp9_payload_type.context("no vp9 payload type in jingle session-initiate")?,
            "VP9".to_owned(),
            90000,
            1,
          )],
          other => bail!("unsupported video codec: {}", other),
        }
      };

      description.rtcp_mux = Some(RtcpMux);

      let mslabel = Uuid::new_v4().to_string();
      let label = Uuid::new_v4().to_string();
      let cname = Uuid::new_v4().to_string();

      let mut ssrc = jingle_ssma::Source::new(if initiate_content.name.0 == "audio" {
        audio_ssrc.to_string()
      }
      else {
        video_ssrc.to_string()
      });
      ssrc.parameters.push(Parameter {
        name: "cname".to_owned(),
        value: Some(cname),
      });
      ssrc.parameters.push(Parameter {
        name: "msid".to_owned(),
        value: Some(format!("{} {}", mslabel, label)),
      });
      ssrc.parameters.push(Parameter {
        name: "mslabel".to_owned(),
        value: Some(mslabel),
      });
      ssrc.parameters.push(Parameter {
        name: "label".to_owned(),
        value: Some(label),
      });
      description.ssrcs = vec![ssrc];

      let mut transport = IceUdpTransport::new().with_fingerprint(Fingerprint {
        hash: Algo::Sha_256,
        setup: Some(Setup::Active),
        value: fingerprint.clone(),
        required: Some(true.to_string()),
      });
      transport.ufrag = Some(ice_local_ufrag.clone());
      transport.pwd = Some(ice_local_pwd.clone());
      transport.candidates = vec![];
      for c in &local_candidates {
        let addr = c.addr();
        let foundation = c.foundation()?;
        transport.candidates.push(jingle_ice_udp::Candidate {
          component: c.component_id() as u8,
          foundation: foundation.to_owned(),
          generation: 0,
          id: Uuid::new_v4().to_string(),
          ip: addr.ip(),
          port: addr.port(),
          priority: c.priority(),
          protocol: "udp".to_owned(),
          type_: match c.type_() {
            nice::CandidateType::Host => jingle_ice_udp::Type::Host,
            nice::CandidateType::PeerReflexive => jingle_ice_udp::Type::Prflx,
            nice::CandidateType::ServerReflexive => jingle_ice_udp::Type::Srflx,
            nice::CandidateType::Relayed => jingle_ice_udp::Type::Relay,
            other => bail!("unsupported candidate type: {:?}", other),
          },
          rel_addr: None,
          rel_port: None,
          network: None,
        });
      }

      jingle_accept = jingle_accept.add_content(
        Content::new(Creator::Responder, initiate_content.name.clone())
          .with_senders(Senders::Both)
          .with_description(description)
          .with_transport(transport),
      );
    }

    let accept_iq_id = generate_id();
    let session_accept_iq = Iq::from_set(accept_iq_id.clone(), jingle_accept)
      .with_to(Jid::Full(conference.focus_jid_in_muc()?))
      .with_from(Jid::Full(conference.jid.clone()));
    conference.xmpp_tx.send(session_accept_iq.into()).await?;

    Ok(Self {
      pipeline,
      audio_sink_element,
      video_sink_element,
      remote_ssrc_map,
      ice_agent,
      ice_stream_id,
      ice_component_id,
      accept_iq_id: Some(accept_iq_id),
      colibri_url,
      colibri_channel: None,
      pipeline_state_null_rx,
    })
  }

  pub(crate) async fn source_add(&mut self, jingle: Jingle) -> Result<()> {
    for content in &jingle.contents {
      if let Some(Description::Rtp(description)) = &content.description {
        for ssrc in &description.ssrcs {
          let owner = ssrc
            .info
            .as_ref()
            .context("missing ssrc-info")?
            .owner
            .clone();
          if owner == "jvb" {
            debug!("skipping ssrc (owner = jvb)");
            continue;
          }

          debug!("adding ssrc to remote_ssrc_map: {:?}", ssrc);
          self.remote_ssrc_map.insert(
            ssrc.id.parse()?,
            Source {
              ssrc: ssrc.id.parse()?,
              participant_id: owner
                .split('/')
                .nth(1)
                .context("invalid ssrc-info owner")?
                .to_owned(),
              media_type: if description.media == "audio" {
                MediaType::Audio
              }
              else {
                MediaType::Video
              },
            },
          );
        }
      }
    }
    Ok(())
  }
}