<commit_before>/*
 * (C) Copyright 2015 Kurento (http://kurento.org/)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#ifdef HAVE_CONFIG_H
#  include <config.h>
#endif

#include "kmsbasertpsession.h"
#include "constants.h"
#include "kmsutils.h"
#include "sdp_utils.h"

#include "kms-core-enumtypes.h"
#include "kms-core-marshal.h"

#define GST_DEFAULT_NAME "kmsbasertpsession"
#define GST_CAT_DEFAULT kms_base_rtp_session_debug
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);

#define kms_base_rtp_session_parent_class parent_class
G_DEFINE_TYPE (KmsBaseRtpSession, kms_base_rtp_session, KMS_TYPE_SDP_SESSION);

#define BUNDLE_CONN_ADDED "bundle-conn-added"
G_DEFINE_QUARK (BUNDLE_CONN_ADDED, bundle_conn_added);

#define RTCP_DEMUX_PEER "rtcp-demux-peer"
G_DEFINE_QUARK (RTCP_DEMUX_PEER, rtcp_demux_peer);

struct _KmsBaseRTPSessionStats
{
  gboolean enabled;
  gdouble vi;
  gdouble ai;
};

enum
{
  CONNECTION_STATE_CHANGED,
  LAST_SIGNAL
};

static guint obj_signals[LAST_SIGNAL] = { 0 };

#define DEFAULT_CONNECTION_STATE KMS_CONNECTION_STATE_DISCONNECTED

enum
{
  PROP_0,
  PROP_CONNECTION_STATE
};

KmsBaseRtpSession *
kms_base_rtp_session_new (KmsBaseSdpEndpoint * ep, guint id,
    KmsIRtpSessionManager * manager)
{
  GObject *obj;
  KmsBaseRtpSession *self;

  obj = g_object_new (KMS_TYPE_BASE_RTP_SESSION, NULL);
  self = KMS_BASE_RTP_SESSION (obj);
  KMS_BASE_RTP_SESSION_CLASS (G_OBJECT_GET_CLASS (self))->post_constructor
      (self, ep, id, manager);

  return self;
}

/* Connection management begin */

KmsIRtpConnection *
kms_base_rtp_session_get_connection_by_name (KmsBaseRtpSession * self,
    const gchar * name)
{
  gpointer *conn;

  conn = g_hash_table_lookup (self->conns, name);
  if (conn == NULL) {
    return NULL;
  }

  return KMS_I_RTP_CONNECTION (conn);
}

static gchar *
kms_base_rtp_session_create_connection_name_from_handler (KmsBaseRtpSession *
    self, KmsSdpMediaHandler * handler)
{
  gchar *conn_name = NULL;
  gint gid, hid;

  g_object_get (handler, "id", &hid, NULL);

  gid = kms_sdp_agent_get_handler_group_id (KMS_SDP_SESSION (self)->agent, hid);

  if (gid >= 0) {
    conn_name =
        g_strdup_printf ("%s%" G_GINT32_FORMAT, BUNDLE_STREAM_NAME, gid);
  } else if (hid >= 0) {
    conn_name = g_strdup_printf ("%" G_GINT32_FORMAT, hid);
  } else {
    GST_ERROR_OBJECT (self, "Wrong handler");
    g_assert_not_reached ();
  }

  return conn_name;
}

KmsIRtpConnection *
kms_base_rtp_session_get_connection (KmsBaseRtpSession * self,
    KmsSdpMediaHandler * handler)
{
  gchar *name = kms_base_rtp_session_create_connection_name_from_handler (self,
      handler);
  KmsIRtpConnection *conn;

  conn = kms_base_rtp_session_get_connection_by_name (self, name);
  if (conn == NULL) {
    GST_WARNING_OBJECT (self, "Connection '%s' not found", name);
    g_free (name);
    return NULL;
  }
  g_free (name);

  return conn;
}

static KmsIRtpConnection *
kms_base_rtp_session_create_connection_default (KmsBaseRtpSession * self,
    const GstSDPMedia * media, const gchar * name,
    guint16 min_port, guint16 max_port)
{
  KmsBaseRtpSessionClass *klass =
      KMS_BASE_RTP_SESSION_CLASS (G_OBJECT_GET_CLASS (self));

  if (klass->create_connection ==
      kms_base_rtp_session_create_connection_default) {
    GST_WARNING_OBJECT (self, "%s does not reimplement 'create_connection'",
        G_OBJECT_CLASS_NAME (klass));
  }

  return NULL;
}

static KmsIRtcpMuxConnection *
kms_base_rtp_session_create_rtcp_mux_connection_default (KmsBaseRtpSession *
    self, const gchar * name, guint16 min_port, guint16 max_port)
{
  KmsBaseRtpSessionClass *klass =
      KMS_BASE_RTP_SESSION_CLASS (G_OBJECT_GET_CLASS (self));

  if (klass->create_rtcp_mux_connection ==
      kms_base_rtp_session_create_rtcp_mux_connection_default) {
    GST_WARNING_OBJECT (self,
        "%s does not reimplement 'create_rtcp_mux_connection'",
        G_OBJECT_CLASS_NAME (klass));
  }

  return NULL;
}

static KmsIBundleConnection *
kms_base_rtp_session_create_bundle_connection_default (KmsBaseRtpSession *
    self, const gchar * name, guint16 min_port, guint16 max_port)
{
  KmsBaseRtpSessionClass *klass =
      KMS_BASE_RTP_SESSION_CLASS (G_OBJECT_GET_CLASS (self));

  if (klass->create_bundle_connection ==
      kms_base_rtp_session_create_bundle_connection_default) {
    GST_WARNING_OBJECT (self,
        "%s does not reimplement 'create_bundle_connection'",
        G_OBJECT_CLASS_NAME (klass));
  }

  return NULL;
}

static void
kms_base_rtp_session_e2e_latency_cb (GstPad * pad, KmsMediaType type,
    GstClockTimeDiff t, KmsList * mdata, gpointer user_data)
{
  KmsBaseRtpSession *self = KMS_BASE_RTP_SESSION (user_data);
  KmsListIter iter;
  gpointer key, value;
  gchar *name;

  name = gst_element_get_name (KMS_SDP_SESSION (self)->ep);

  kms_list_iter_init (&iter, mdata);
  while (kms_list_iter_next (&iter, &key, &value)) {
    gchar *id = (gchar *) key;
    StreamE2EAvgStat *stat;

    if (!g_str_has_prefix (id, name)) {
      /* This element did not add this mark to the metada */
      continue;
    }

    stat = (StreamE2EAvgStat *) value;
    stat->avg = KMS_STATS_CALCULATE_LATENCY_AVG (t, stat->avg);
  }

  g_free (name);
}

static void
kms_base_rtp_session_set_connection_stats (KmsBaseRtpSession * self,
    KmsIRtpConnection * conn)
{
  kms_i_rtp_connection_set_latency_callback (conn,
      kms_base_rtp_session_e2e_latency_cb, self);

  /* Active insertion of metadata if stats are enabled */
  kms_i_rtp_connection_collect_latency_stats (conn, self->stats_enabled);
}

KmsIRtpConnection *
kms_base_rtp_session_create_connection (KmsBaseRtpSession * self,
    KmsSdpMediaHandler * handler, GstSDPMedia * media, guint16 min_port,
    guint16 max_port)
{
  KmsBaseRtpSessionClass *base_rtp_class =
      KMS_BASE_RTP_SESSION_CLASS (G_OBJECT_GET_CLASS (self));
  gchar *name =
      kms_base_rtp_session_create_connection_name_from_handler (self, handler);
  KmsIRtpConnection *conn = NULL;

  if (name == NULL) {
    GST_WARNING_OBJECT (self, "Connection can not be created");
    goto end;
  }

  conn = kms_base_rtp_session_get_connection_by_name (self, name);
  if (conn != NULL) {
    GST_DEBUG_OBJECT (self, "Re-using connection '%s'", name);
    goto end;
  }

  if (g_str_has_prefix (name, BUNDLE_STREAM_NAME)) {    /* bundle */
    conn =
        KMS_I_RTP_CONNECTION (base_rtp_class->create_bundle_connection (self,
            name, min_port, max_port));
  } else if (gst_sdp_media_get_attribute_val (media, "rtcp-mux") != NULL) {
    conn =
        KMS_I_RTP_CONNECTION (base_rtp_class->create_rtcp_mux_connection
        (self, name, min_port, max_port));
  } else {
    conn =
        base_rtp_class->create_connection (self, media, name, min_port,
        max_port);
  }

  if (conn != NULL) {
    g_hash_table_insert (self->conns, g_strdup (name), conn);

    kms_base_rtp_session_set_connection_stats (self, conn);
  }

end:
  g_free (name);

  return conn;
}

/* Connection management end */

/* Start Transport Send begin */

static gboolean
ssrcs_are_mapped (GstElement * ssrcdemux,
    guint32 local_ssrc, guint32 remote_ssrc)
{
  GstElement *rtcpdemux =
      g_object_get_qdata (G_OBJECT (ssrcdemux), rtcp_demux_peer_quark ());
  guint local_ssrc_pair;

  g_signal_emit_by_name (rtcpdemux, "get-local-rr-ssrc-pair", remote_ssrc,
      &local_ssrc_pair);

  return ((local_ssrc != SSRC_INVALID) && (local_ssrc_pair == local_ssrc));
}

static void
kms_base_rtp_session_link_pads (GstPad *src, GstPad *sink)
{
  // Link without the GstPad hierarchy checks.
  GstPadLinkReturn ret = gst_pad_link_full (src, sink,
      GST_PAD_LINK_CHECK_DEFAULT & ~GST_PAD_LINK_CHECK_HIERARCHY);

  if (ret != GST_PAD_LINK_OK) {
    GST_ERROR ("Error linking pads, src: %" GST_PTR_FORMAT
               ", sink: %" GST_PTR_FORMAT ", reason: '%s'",
        src, sink, gst_pad_link_get_name (ret));
  }
}

static void
rtp_ssrc_demux_new_ssrc_pad (GstElement * ssrcdemux, guint ssrc, GstPad * pad,
    KmsBaseRtpSession * self)
{
  const gchar *rtp_pad_name = GST_OBJECT_NAME (pad);
  gchar *rtcp_pad_name;
  const GstSDPMedia *media;
  GstPad *src, *sink;

  GST_DEBUG_OBJECT (self, "pad: %" GST_PTR_FORMAT " ssrc: %" G_GUINT32_FORMAT,
      pad, ssrc);

  KMS_SDP_SESSION_LOCK (self);

  if (self->remote_audio_ssrc == ssrc
      || ssrcs_are_mapped (ssrcdemux, self->local_audio_ssrc, ssrc)) {
    media = self->audio_neg;
  } else if (self->remote_video_ssrc == ssrc
      || ssrcs_are_mapped (ssrcdemux, self->local_video_ssrc, ssrc)) {
    media = self->video_neg;
  } else {
    if (!kms_i_rtp_session_manager_custom_ssrc_management (self->manager, self,
            ssrcdemux, ssrc, pad)) {
      GST_ERROR_OBJECT (pad, "SSRC %" G_GUINT32_FORMAT " not matching.", ssrc);
    }
    goto end;
  }

  /* RTP */
  sink = kms_i_rtp_session_manager_request_rtp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (pad, sink);
  g_object_unref (sink);

  /* RTCP */
  rtcp_pad_name = g_strconcat ("rtcp_", rtp_pad_name, NULL);
  src = gst_element_get_static_pad (ssrcdemux, rtcp_pad_name);
  g_free (rtcp_pad_name);
  sink = kms_i_rtp_session_manager_request_rtcp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

end:
  KMS_SDP_SESSION_UNLOCK (self);
}

static void
kms_base_rtp_session_add_gst_bundle_elements (KmsBaseRtpSession * self,
    KmsIRtpConnection * conn, const GstSDPMedia * media, gboolean active)
{
  gboolean added;
  GstElement *ssrcdemux;
  GstElement *rtcpdemux;        /* FIXME: Useful for local and remote ssrcs mapping */
  GstPad *src, *sink;

  if (GPOINTER_TO_UINT (g_object_get_qdata (G_OBJECT (conn),
              bundle_conn_added_quark ()))) {
    GST_DEBUG_OBJECT (self, "Connection configured");
    return;
  }

  g_object_set_qdata (G_OBJECT (conn), bundle_conn_added_quark (),
      GUINT_TO_POINTER (TRUE));

  g_object_get (conn, "added", &added, NULL);
  if (!added) {
    kms_i_rtp_connection_add (conn, GST_BIN (self), active);
  }

  ssrcdemux = gst_element_factory_make ("rtpssrcdemux", NULL);
  rtcpdemux = gst_element_factory_make ("rtcpdemux", NULL);

  g_object_set_qdata_full (G_OBJECT (ssrcdemux), rtcp_demux_peer_quark (),
      g_object_ref (rtcpdemux), g_object_unref);
  g_signal_connect (ssrcdemux, "new-ssrc-pad",
      G_CALLBACK (rtp_ssrc_demux_new_ssrc_pad), self);

  kms_i_rtp_connection_sink_sync_state_with_parent (conn);
  gst_bin_add_many (GST_BIN (self), ssrcdemux, rtcpdemux, NULL);

  /* RTP */
  src = kms_i_rtp_connection_request_rtp_src (conn);
  sink = gst_element_get_static_pad (ssrcdemux, "sink");
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

  /* RTCP */
  src = kms_i_rtp_connection_request_rtcp_src (conn);
  sink = gst_element_get_static_pad (rtcpdemux, "sink");
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);
  gst_element_link_pads (rtcpdemux, "rtcp_src", ssrcdemux, "rtcp_sink");

  gst_element_sync_state_with_parent_target_state (ssrcdemux);
  gst_element_sync_state_with_parent_target_state (rtcpdemux);

  kms_i_rtp_connection_src_sync_state_with_parent (conn);
}

static void
kms_base_rtp_session_link_gst_connection_sink (KmsBaseRtpSession * self,
    KmsIRtpConnection * conn, const GstSDPMedia * media)
{
  GstPad *src, *sink;

  /* RTP */
  src = kms_i_rtp_session_manager_request_rtp_src (self->manager, self, media);
  sink = kms_i_rtp_connection_request_rtp_sink (conn);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

  /* RTCP */
  src = kms_i_rtp_session_manager_request_rtcp_src (self->manager, self, media);
  sink = kms_i_rtp_connection_request_rtcp_sink (conn);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);
}

static void
kms_base_rtp_session_link_gst_connection_src (KmsBaseRtpSession * self,
    KmsIRtpConnection * conn, const GstSDPMedia * media)
{
  GstPad *src, *sink;

  /* RTP */
  src = kms_i_rtp_connection_request_rtp_src (conn);
  sink = kms_i_rtp_session_manager_request_rtp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

  /* RTCP */
  src = kms_i_rtp_connection_request_rtcp_src (conn);
  sink = kms_i_rtp_session_manager_request_rtcp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);
}

static void
kms_base_rtp_session_add_gst_rtcp_mux_elements (KmsBaseRtpSession * self,
    KmsIRtpConnection * conn, const GstSDPMedia * media, gboolean active)
{
  /* FIXME: Useful for local and remote ssrcs mapping */
  GstElement *rtcpdemux = gst_element_factory_make ("rtcpdemux", NULL);
  GstPad *src, *sink;

  kms_i_rtp_connection_add (conn, GST_BIN (self), active);
  kms_i_rtp_connection_sink_sync_state_with_parent (conn);
  gst_bin_add (GST_BIN (self), rtcpdemux);

  /* RTP */
  src = kms_i_rtp_connection_request_rtp_src (conn);
  sink = kms_i_rtp_session_manager_request_rtp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

  /* RTCP */
  src = gst_element_get_static_pad (rtcpdemux, "rtcp_src");
  sink = kms_i_rtp_session_manager_request_rtcp_sink (self->manager, self, media);
  g_object_unref (src);
  g_object_unref (sink);

  src = kms_i_rtp_connection_request_rtcp_src (conn);
  sink = gst_element_get_static_pad (rtcpdemux, "sink");
  kms_base_rtp_session_link_pads (src, sink);<commit_msg>rtp: use rtcpdemux::new-ssrc-pad signal to demux based on Media SSRC<commit_after>/*
 * (C) Copyright 2015 Kurento (http://kurento.org/)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#ifdef HAVE_CONFIG_H
#  include <config.h>
#endif

#include "kmsbasertpsession.h"
#include "constants.h"
#include "kmsutils.h"
#include "sdp_utils.h"

#include "kms-core-enumtypes.h"
#include "kms-core-marshal.h"

#define GST_DEFAULT_NAME "kmsbasertpsession"
#define GST_CAT_DEFAULT kms_base_rtp_session_debug
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);

#define kms_base_rtp_session_parent_class parent_class
G_DEFINE_TYPE (KmsBaseRtpSession, kms_base_rtp_session, KMS_TYPE_SDP_SESSION);

#define BUNDLE_CONN_ADDED "bundle-conn-added"
G_DEFINE_QUARK (BUNDLE_CONN_ADDED, bundle_conn_added);

#define RTCP_DEMUX_PEER "rtcp-demux-peer"
G_DEFINE_QUARK (RTCP_DEMUX_PEER, rtcp_demux_peer);

struct _KmsBaseRTPSessionStats
{
  gboolean enabled;
  gdouble vi;
  gdouble ai;
};

enum
{
  CONNECTION_STATE_CHANGED,
  LAST_SIGNAL
};

static guint obj_signals[LAST_SIGNAL] = { 0 };

#define DEFAULT_CONNECTION_STATE KMS_CONNECTION_STATE_DISCONNECTED

enum
{
  PROP_0,
  PROP_CONNECTION_STATE
};

KmsBaseRtpSession *
kms_base_rtp_session_new (KmsBaseSdpEndpoint * ep, guint id,
    KmsIRtpSessionManager * manager)
{
  GObject *obj;
  KmsBaseRtpSession *self;

  obj = g_object_new (KMS_TYPE_BASE_RTP_SESSION, NULL);
  self = KMS_BASE_RTP_SESSION (obj);
  KMS_BASE_RTP_SESSION_CLASS (G_OBJECT_GET_CLASS (self))->post_constructor
      (self, ep, id, manager);

  return self;
}

/* Connection management begin */

KmsIRtpConnection *
kms_base_rtp_session_get_connection_by_name (KmsBaseRtpSession * self,
    const gchar * name)
{
  gpointer *conn;

  conn = g_hash_table_lookup (self->conns, name);
  if (conn == NULL) {
    return NULL;
  }

  return KMS_I_RTP_CONNECTION (conn);
}

static gchar *
kms_base_rtp_session_create_connection_name_from_handler (KmsBaseRtpSession *
    self, KmsSdpMediaHandler * handler)
{
  gchar *conn_name = NULL;
  gint gid, hid;

  g_object_get (handler, "id", &hid, NULL);

  gid = kms_sdp_agent_get_handler_group_id (KMS_SDP_SESSION (self)->agent, hid);

  if (gid >= 0) {
    conn_name =
        g_strdup_printf ("%s%" G_GINT32_FORMAT, BUNDLE_STREAM_NAME, gid);
  } else if (hid >= 0) {
    conn_name = g_strdup_printf ("%" G_GINT32_FORMAT, hid);
  } else {
    GST_ERROR_OBJECT (self, "Wrong handler");
    g_assert_not_reached ();
  }

  return conn_name;
}

KmsIRtpConnection *
kms_base_rtp_session_get_connection (KmsBaseRtpSession * self,
    KmsSdpMediaHandler * handler)
{
  gchar *name = kms_base_rtp_session_create_connection_name_from_handler (self,
      handler);
  KmsIRtpConnection *conn;

  conn = kms_base_rtp_session_get_connection_by_name (self, name);
  if (conn == NULL) {
    GST_WARNING_OBJECT (self, "Connection '%s' not found", name);
    g_free (name);
    return NULL;
  }
  g_free (name);

  return conn;
}

static KmsIRtpConnection *
kms_base_rtp_session_create_connection_default (KmsBaseRtpSession * self,
    const GstSDPMedia * media, const gchar * name,
    guint16 min_port, guint16 max_port)
{
  KmsBaseRtpSessionClass *klass =
      KMS_BASE_RTP_SESSION_CLASS (G_OBJECT_GET_CLASS (self));

  if (klass->create_connection ==
      kms_base_rtp_session_create_connection_default) {
    GST_WARNING_OBJECT (self, "%s does not reimplement 'create_connection'",
        G_OBJECT_CLASS_NAME (klass));
  }

  return NULL;
}

static KmsIRtcpMuxConnection *
kms_base_rtp_session_create_rtcp_mux_connection_default (KmsBaseRtpSession *
    self, const gchar * name, guint16 min_port, guint16 max_port)
{
  KmsBaseRtpSessionClass *klass =
      KMS_BASE_RTP_SESSION_CLASS (G_OBJECT_GET_CLASS (self));

  if (klass->create_rtcp_mux_connection ==
      kms_base_rtp_session_create_rtcp_mux_connection_default) {
    GST_WARNING_OBJECT (self,
        "%s does not reimplement 'create_rtcp_mux_connection'",
        G_OBJECT_CLASS_NAME (klass));
  }

  return NULL;
}

static KmsIBundleConnection *
kms_base_rtp_session_create_bundle_connection_default (KmsBaseRtpSession *
    self, const gchar * name, guint16 min_port, guint16 max_port)
{
  KmsBaseRtpSessionClass *klass =
      KMS_BASE_RTP_SESSION_CLASS (G_OBJECT_GET_CLASS (self));

  if (klass->create_bundle_connection ==
      kms_base_rtp_session_create_bundle_connection_default) {
    GST_WARNING_OBJECT (self,
        "%s does not reimplement 'create_bundle_connection'",
        G_OBJECT_CLASS_NAME (klass));
  }

  return NULL;
}

static void
kms_base_rtp_session_e2e_latency_cb (GstPad * pad, KmsMediaType type,
    GstClockTimeDiff t, KmsList * mdata, gpointer user_data)
{
  KmsBaseRtpSession *self = KMS_BASE_RTP_SESSION (user_data);
  KmsListIter iter;
  gpointer key, value;
  gchar *name;

  name = gst_element_get_name (KMS_SDP_SESSION (self)->ep);

  kms_list_iter_init (&iter, mdata);
  while (kms_list_iter_next (&iter, &key, &value)) {
    gchar *id = (gchar *) key;
    StreamE2EAvgStat *stat;

    if (!g_str_has_prefix (id, name)) {
      /* This element did not add this mark to the metada */
      continue;
    }

    stat = (StreamE2EAvgStat *) value;
    stat->avg = KMS_STATS_CALCULATE_LATENCY_AVG (t, stat->avg);
  }

  g_free (name);
}

static void
kms_base_rtp_session_set_connection_stats (KmsBaseRtpSession * self,
    KmsIRtpConnection * conn)
{
  kms_i_rtp_connection_set_latency_callback (conn,
      kms_base_rtp_session_e2e_latency_cb, self);

  /* Active insertion of metadata if stats are enabled */
  kms_i_rtp_connection_collect_latency_stats (conn, self->stats_enabled);
}

KmsIRtpConnection *
kms_base_rtp_session_create_connection (KmsBaseRtpSession * self,
    KmsSdpMediaHandler * handler, GstSDPMedia * media, guint16 min_port,
    guint16 max_port)
{
  KmsBaseRtpSessionClass *base_rtp_class =
      KMS_BASE_RTP_SESSION_CLASS (G_OBJECT_GET_CLASS (self));
  gchar *name =
      kms_base_rtp_session_create_connection_name_from_handler (self, handler);
  KmsIRtpConnection *conn = NULL;

  if (name == NULL) {
    GST_WARNING_OBJECT (self, "Connection can not be created");
    goto end;
  }

  conn = kms_base_rtp_session_get_connection_by_name (self, name);
  if (conn != NULL) {
    GST_DEBUG_OBJECT (self, "Re-using connection '%s'", name);
    goto end;
  }

  if (g_str_has_prefix (name, BUNDLE_STREAM_NAME)) {    /* bundle */
    conn =
        KMS_I_RTP_CONNECTION (base_rtp_class->create_bundle_connection (self,
            name, min_port, max_port));
  } else if (gst_sdp_media_get_attribute_val (media, "rtcp-mux") != NULL) {
    conn =
        KMS_I_RTP_CONNECTION (base_rtp_class->create_rtcp_mux_connection
        (self, name, min_port, max_port));
  } else {
    conn =
        base_rtp_class->create_connection (self, media, name, min_port,
        max_port);
  }

  if (conn != NULL) {
    g_hash_table_insert (self->conns, g_strdup (name), conn);

    kms_base_rtp_session_set_connection_stats (self, conn);
  }

end:
  g_free (name);

  return conn;
}

/* Connection management end */

/* Start Transport Send begin */

static gboolean
ssrcs_are_mapped (GstElement * ssrcdemux,
    guint32 local_ssrc, guint32 remote_ssrc)
{
  GstElement *rtcpdemux =
      g_object_get_qdata (G_OBJECT (ssrcdemux), rtcp_demux_peer_quark ());
  guint local_ssrc_pair;

  g_signal_emit_by_name (rtcpdemux, "get-local-rr-ssrc-pair", remote_ssrc,
      &local_ssrc_pair);

  return ((local_ssrc != SSRC_INVALID) && (local_ssrc_pair == local_ssrc));
}

static void
kms_base_rtp_session_link_pads (GstPad *src, GstPad *sink)
{
  // Link without the GstPad hierarchy checks.
  GstPadLinkReturn ret = gst_pad_link_full (src, sink,
      GST_PAD_LINK_CHECK_DEFAULT & ~GST_PAD_LINK_CHECK_HIERARCHY);

  if (ret != GST_PAD_LINK_OK) {
    GST_ERROR ("Error linking pads, src: %" GST_PTR_FORMAT
               ", sink: %" GST_PTR_FORMAT ", reason: '%s'",
        src, sink, gst_pad_link_get_name (ret));
  }
}

static void
on_rtcpdemux_new_ssrc_pad (GstElement *rtcpdemux,
    guint ssrc,
    GstPad *pad,
    KmsBaseRtpSession *self)
{
  GST_DEBUG_OBJECT (self, "Local SSRC: %u, pad: %" GST_PTR_FORMAT, ssrc, pad);

  KMS_SDP_SESSION_LOCK (self);

  // Link the new pad with the appropriate sink from GstBin.
  if (self->local_audio_ssrc == ssrc) {
    GstPad *sink = kms_i_rtp_session_manager_request_rtcp_sink (self->manager,
        self, self->audio_neg);
    kms_base_rtp_session_link_pads (pad, sink);
    gst_object_unref (sink);
  } else if (self->local_video_ssrc == ssrc) {
    GstPad *sink = kms_i_rtp_session_manager_request_rtcp_sink (self->manager,
        self, self->video_neg);
    kms_base_rtp_session_link_pads (pad, sink);
    gst_object_unref (sink);
  } else {
    GST_ERROR_OBJECT (self,
        "Local SSRC %u in RTCP doesn't match any SDP media; DATA WILL BE DROPPED",
        ssrc);

    // Cannot identify which local media corresponds to the SSRC.
    // The pad cannot be left without linking, so discard into a fakesink.
    GstElement *fakesink = gst_element_factory_make ("fakesink", NULL);
    gst_bin_add (GST_BIN (self), fakesink);
    gst_element_sync_state_with_parent_target_state (fakesink);
    GstPad *sink = gst_element_get_static_pad (fakesink, "sink");
    kms_base_rtp_session_link_pads (pad, sink);
    gst_object_unref(sink);
  }

  KMS_SDP_SESSION_UNLOCK (self);
}

static void
rtp_ssrc_demux_new_ssrc_pad (GstElement * ssrcdemux, guint ssrc, GstPad * pad,
    KmsBaseRtpSession * self)
{
  const gchar *rtp_pad_name = GST_OBJECT_NAME (pad);
  gchar *rtcp_pad_name;
  const GstSDPMedia *media;
  GstPad *src, *sink;

  GST_DEBUG_OBJECT (self, "Remote SSRC: %u, pad: %" GST_PTR_FORMAT, ssrc, pad);

  KMS_SDP_SESSION_LOCK (self);

  if (self->remote_audio_ssrc == ssrc
      || ssrcs_are_mapped (ssrcdemux, self->local_audio_ssrc, ssrc)) {
    media = self->audio_neg;
  } else if (self->remote_video_ssrc == ssrc
      || ssrcs_are_mapped (ssrcdemux, self->local_video_ssrc, ssrc)) {
    media = self->video_neg;
  } else {
    if (!kms_i_rtp_session_manager_custom_ssrc_management (self->manager, self,
            ssrcdemux, ssrc, pad)) {
      GST_ERROR_OBJECT (self,
          "Remote SSRC %u in RTCP doesn't match any SDP media",
          ssrc);
    }
    goto end;
  }

  /* RTP */
  sink = kms_i_rtp_session_manager_request_rtp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (pad, sink);
  g_object_unref (sink);

  /* RTCP */
  rtcp_pad_name = g_strconcat ("rtcp_", rtp_pad_name, NULL);
  src = gst_element_get_static_pad (ssrcdemux, rtcp_pad_name);
  g_free (rtcp_pad_name);
  sink = kms_i_rtp_session_manager_request_rtcp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

end:
  KMS_SDP_SESSION_UNLOCK (self);
}

static void
kms_base_rtp_session_add_gst_bundle_elements (KmsBaseRtpSession * self,
    KmsIRtpConnection * conn, const GstSDPMedia * media, gboolean active)
{
  gboolean added;
  GstElement *rtcpdemux;
  GstElement *ssrcdemux;
  GstPad *src, *sink;

  if (GPOINTER_TO_UINT (g_object_get_qdata (G_OBJECT (conn),
              bundle_conn_added_quark ()))) {
    GST_DEBUG_OBJECT (self, "Connection configured");
    return;
  }

  g_object_set_qdata (G_OBJECT (conn), bundle_conn_added_quark (),
      GUINT_TO_POINTER (TRUE));

  g_object_get (conn, "added", &added, NULL);
  if (!added) {
    kms_i_rtp_connection_add (conn, GST_BIN (self), active);
  }

  rtcpdemux = gst_element_factory_make ("rtcpdemux", NULL);

  g_signal_connect (rtcpdemux, "new-ssrc-pad",
      G_CALLBACK (on_rtcpdemux_new_ssrc_pad), self);

  ssrcdemux = gst_element_factory_make ("rtpssrcdemux", NULL);

  g_object_set_qdata_full (G_OBJECT (ssrcdemux), rtcp_demux_peer_quark (),
      g_object_ref (rtcpdemux), g_object_unref);
  g_signal_connect (ssrcdemux, "new-ssrc-pad",
      G_CALLBACK (rtp_ssrc_demux_new_ssrc_pad), self);

  kms_i_rtp_connection_sink_sync_state_with_parent (conn);
  gst_bin_add_many (GST_BIN (self), ssrcdemux, rtcpdemux, NULL);

  /* RTP */

  src = kms_i_rtp_connection_request_rtp_src (conn);
  sink = gst_element_get_static_pad (ssrcdemux, "sink");
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

  /* RTCP */

  src = kms_i_rtp_connection_request_rtcp_src (conn);
  sink = gst_element_get_static_pad (rtcpdemux, "sink");
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

  gst_element_link_pads (rtcpdemux, "rtcp_src", ssrcdemux, "rtcp_sink");

  gst_element_sync_state_with_parent_target_state (rtcpdemux);
  gst_element_sync_state_with_parent_target_state (ssrcdemux);

  kms_i_rtp_connection_src_sync_state_with_parent (conn);
}

static void
kms_base_rtp_session_link_gst_connection_sink (KmsBaseRtpSession * self,
    KmsIRtpConnection * conn, const GstSDPMedia * media)
{
  GstPad *src, *sink;

  /* RTP */
  src = kms_i_rtp_session_manager_request_rtp_src (self->manager, self, media);
  sink = kms_i_rtp_connection_request_rtp_sink (conn);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

  /* RTCP */
  src = kms_i_rtp_session_manager_request_rtcp_src (self->manager, self, media);
  sink = kms_i_rtp_connection_request_rtcp_sink (conn);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);
}

static void
kms_base_rtp_session_link_gst_connection_src (KmsBaseRtpSession * self,
    KmsIRtpConnection * conn, const GstSDPMedia * media)
{
  GstPad *src, *sink;

  /* RTP */
  src = kms_i_rtp_connection_request_rtp_src (conn);
  sink = kms_i_rtp_session_manager_request_rtp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

  /* RTCP */
  src = kms_i_rtp_connection_request_rtcp_src (conn);
  sink = kms_i_rtp_session_manager_request_rtcp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);
}

static void
kms_base_rtp_session_add_gst_rtcp_mux_elements (KmsBaseRtpSession * self,
    KmsIRtpConnection * conn, const GstSDPMedia * media, gboolean active)
{
  GstElement *rtcpdemux;
  GstPad *src, *sink;

  rtcpdemux = gst_element_factory_make ("rtcpdemux", NULL);

  g_signal_connect (rtcpdemux, "new-ssrc-pad",
      G_CALLBACK (on_rtcpdemux_new_ssrc_pad), self);

  kms_i_rtp_connection_add (conn, GST_BIN (self), active);
  kms_i_rtp_connection_sink_sync_state_with_parent (conn);
  gst_bin_add (GST_BIN (self), rtcpdemux);

  /* RTP */

  src = kms_i_rtp_connection_request_rtp_src (conn);
  sink =
      kms_i_rtp_session_manager_request_rtp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

  /* RTCP */

  src = kms_i_rtp_connection_request_rtcp_src (conn);
  sink = gst_element_get_static_pad (rtcpdemux, "sink");
  kms_base_rtp_session_link_pads (src, sink);
  g_object_unref (src);
  g_object_unref (sink);

  src = gst_element_get_static_pad (rtcpdemux, "rtcp_src");
  sink =
      kms_i_rtp_session_manager_request_rtcp_sink (self->manager, self, media);
  kms_base_rtp_session_link_pads (src, sink);
{'fraction_non_alphanumeric': 0.09815458678791121, 'fraction_numerical': 0.00280823749665686, 'mean_word_length': 4.7948469585432, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 73, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6071379', 'n_tokens_mistral': 11873, 'n_tokens_neox': 11739, 'n_words': 2229}