jitterbuffer: signal timestamp discont
[platform/upstream/gstreamer.git] / gst / rtpmanager / gstrtpbin.c
index 64fea5b..37d6d19 100644 (file)
@@ -13,8 +13,8 @@
  *
  * You should have received a copy of the GNU Library General Public
  * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
- * Boston, MA 02111-1307, USA.
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
  */
 
 /**
  * <refsect2>
  * <title>Example pipelines</title>
  * |[
- * gst-launch udpsrc port=5000 caps="application/x-rtp, ..." ! .recv_rtp_sink_0 \
+ * gst-launch-1.0 udpsrc port=5000 caps="application/x-rtp, ..." ! .recv_rtp_sink_0 \
  *     gstrtpbin ! rtptheoradepay ! theoradec ! xvimagesink
  * ]| Receive RTP data from port 5000 and send to the session 0 in gstrtpbin.
  * |[
- * gst-launch gstrtpbin name=rtpbin \
- *         v4l2src ! ffmpegcolorspace ! ffenc_h263 ! rtph263ppay ! rtpbin.send_rtp_sink_0 \
+ * gst-launch-1.0 gstrtpbin name=rtpbin \
+ *         v4l2src ! videoconvert ! ffenc_h263 ! rtph263ppay ! rtpbin.send_rtp_sink_0 \
  *                   rtpbin.send_rtp_src_0 ! udpsink port=5000                            \
  *                   rtpbin.send_rtcp_src_0 ! udpsink port=5001 sync=false async=false    \
  *                   udpsrc port=5005 ! rtpbin.recv_rtcp_sink_0                           \
@@ -89,7 +89,7 @@
  * as soon as possible and do not participate in preroll, sync=false and
  * async=false is configured on udpsink
  * |[
- * gst-launch -v gstrtpbin name=rtpbin                                          \
+ * gst-launch-1.0 -v gstrtpbin name=rtpbin                                          \
  *     udpsrc caps="application/x-rtp,media=(string)video,clock-rate=(int)90000,encoding-name=(string)H263-1998" \
  *             port=5000 ! rtpbin.recv_rtp_sink_0                                \
  *         rtpbin. ! rtph263pdepay ! ffdec_h263 ! xvimagesink                    \
@@ -243,6 +243,7 @@ enum
 };
 
 #define DEFAULT_LATENCY_MS           200
+#define DEFAULT_DROP_ON_LATENCY      FALSE
 #define DEFAULT_SDES                 NULL
 #define DEFAULT_DO_LOST              FALSE
 #define DEFAULT_IGNORE_PT            FALSE
@@ -257,6 +258,7 @@ enum
 {
   PROP_0,
   PROP_LATENCY,
+  PROP_DROP_ON_LATENCY,
   PROP_SDES,
   PROP_DO_LOST,
   PROP_IGNORE_PT,
@@ -305,8 +307,12 @@ static GstCaps *pt_map_requested (GstElement * element, guint pt,
     GstRtpBinSession * session);
 static void payload_type_change (GstElement * element, guint pt,
     GstRtpBinSession * session);
+static void remove_recv_rtp (GstRtpBin * rtpbin, GstRtpBinSession * session);
+static void remove_recv_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session);
+static void remove_send_rtp (GstRtpBin * rtpbin, GstRtpBinSession * session);
+static void remove_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session);
 static void free_client (GstRtpBinClient * client, GstRtpBin * bin);
-static void free_stream (GstRtpBinStream * stream);
+static void free_stream (GstRtpBinStream * stream, GstRtpBin * bin);
 
 /* Manages the RTP stream for one SSRC.
  *
@@ -539,6 +545,11 @@ ssrc_demux_pad_removed (GstElement * element, guint ssrc, GstPad * pad,
     GstRtpBinSession * session)
 {
   GstRtpBinStream *stream = NULL;
+  GstRtpBin *rtpbin;
+
+  rtpbin = session->bin;
+
+  GST_RTP_BIN_LOCK (rtpbin);
 
   GST_RTP_SESSION_LOCK (session);
   if ((stream = find_stream_by_ssrc (session, ssrc)))
@@ -546,7 +557,9 @@ ssrc_demux_pad_removed (GstElement * element, guint ssrc, GstPad * pad,
   GST_RTP_SESSION_UNLOCK (session);
 
   if (stream)
-    free_stream (stream);
+    free_stream (stream, rtpbin);
+
+  GST_RTP_BIN_UNLOCK (rtpbin);
 }
 
 /* create a session with the given id.  Must be called with RTP_BIN_LOCK */
@@ -628,11 +641,10 @@ no_demux:
   }
 }
 
+/* called with RTP_BIN_LOCK */
 static void
 free_session (GstRtpBinSession * sess, GstRtpBin * bin)
 {
-  GSList *client_walk;
-
   GST_DEBUG_OBJECT (bin, "freeing session %p", sess);
 
   gst_element_set_locked_state (sess->demux, TRUE);
@@ -641,66 +653,15 @@ free_session (GstRtpBinSession * sess, GstRtpBin * bin)
   gst_element_set_state (sess->demux, GST_STATE_NULL);
   gst_element_set_state (sess->session, GST_STATE_NULL);
 
-  if (sess->recv_rtp_sink != NULL) {
-    gst_element_release_request_pad (sess->session, sess->recv_rtp_sink);
-    gst_object_unref (sess->recv_rtp_sink);
-  }
-  if (sess->recv_rtp_src != NULL)
-    gst_object_unref (sess->recv_rtp_src);
-  if (sess->recv_rtcp_sink != NULL) {
-    gst_element_release_request_pad (sess->session, sess->recv_rtcp_sink);
-    gst_object_unref (sess->recv_rtcp_sink);
-  }
-  if (sess->sync_src != NULL)
-    gst_object_unref (sess->sync_src);
-  if (sess->send_rtp_sink != NULL) {
-    gst_element_release_request_pad (sess->session, sess->send_rtp_sink);
-    gst_object_unref (sess->send_rtp_sink);
-  }
-  if (sess->send_rtp_src != NULL)
-    gst_object_unref (sess->send_rtp_src);
-  if (sess->send_rtcp_src != NULL) {
-    gst_element_release_request_pad (sess->session, sess->send_rtcp_src);
-    gst_object_unref (sess->send_rtcp_src);
-  }
+  remove_recv_rtp (bin, sess);
+  remove_recv_rtcp (bin, sess);
+  remove_send_rtp (bin, sess);
+  remove_rtcp (bin, sess);
 
   gst_bin_remove (GST_BIN_CAST (bin), sess->session);
   gst_bin_remove (GST_BIN_CAST (bin), sess->demux);
 
-  /* remove any references in bin->clients to the streams in sess->streams */
-  client_walk = bin->clients;
-  while (client_walk) {
-    GSList *client_node = client_walk;
-    GstRtpBinClient *client = (GstRtpBinClient *) client_node->data;
-    GSList *stream_walk = client->streams;
-
-    while (stream_walk) {
-      GSList *stream_node = stream_walk;
-      GstRtpBinStream *stream = (GstRtpBinStream *) stream_node->data;
-      GSList *inner_walk;
-
-      stream_walk = g_slist_next (stream_walk);
-
-      for (inner_walk = sess->streams; inner_walk;
-          inner_walk = g_slist_next (inner_walk)) {
-        if ((GstRtpBinStream *) inner_walk->data == stream) {
-          client->streams = g_slist_delete_link (client->streams, stream_node);
-          --client->nstreams;
-          break;
-        }
-      }
-    }
-    client_walk = g_slist_next (client_walk);
-
-    g_assert ((client->streams && client->nstreams > 0) || (!client->streams
-            && client->streams == 0));
-    if (client->nstreams == 0) {
-      free_client (client, bin);
-      bin->clients = g_slist_delete_link (bin->clients, client_node);
-    }
-  }
-
-  g_slist_foreach (sess->streams, (GFunc) free_stream, NULL);
+  g_slist_foreach (sess->streams, (GFunc) free_stream, bin);
   g_slist_free (sess->streams);
 
   g_mutex_clear (&sess->lock);
@@ -950,7 +911,7 @@ get_current_times (GstRtpBin * bin, GstClockTime * running_time,
     clock_time = gst_clock_get_time (clock);
 
     if (bin->use_pipeline_clock) {
-      ntpns = clock_time;
+      ntpns = clock_time - base_time;
     } else {
       GTimeVal current;
 
@@ -979,7 +940,7 @@ get_current_times (GstRtpBin * bin, GstClockTime * running_time,
 
 static void
 stream_set_ts_offset (GstRtpBin * bin, GstRtpBinStream * stream,
-    gint64 ts_offset)
+    gint64 ts_offset, gboolean check)
 {
   gint64 prev_ts_offset;
 
@@ -995,18 +956,20 @@ stream_set_ts_offset (GstRtpBin * bin, GstRtpBinStream * stream,
         "ts-offset %" G_GINT64_FORMAT ", prev %" G_GINT64_FORMAT
         ", diff: %" G_GINT64_FORMAT, ts_offset, prev_ts_offset, diff);
 
-    /* only change diff when it changed more than 4 milliseconds. This
-     * compensates for rounding errors in NTP to RTP timestamp
-     * conversions */
-    if (ABS (diff) > 4 * GST_MSECOND) {
-      if (ABS (diff) < (3 * GST_SECOND)) {
-        g_object_set (stream->buffer, "ts-offset", ts_offset, NULL);
-      } else {
+    if (check) {
+      /* only change diff when it changed more than 4 milliseconds. This
+       * compensates for rounding errors in NTP to RTP timestamp
+       * conversions */
+      if (ABS (diff) < 4 * GST_MSECOND) {
+        GST_DEBUG_OBJECT (bin, "offset too small, ignoring");
+        return;
+      }
+      if (ABS (diff) > (3 * GST_SECOND)) {
         GST_WARNING_OBJECT (bin, "offset unusually large, ignoring");
+        return;
       }
-    } else {
-      GST_DEBUG_OBJECT (bin, "offset too small, ignoring");
     }
+    g_object_set (stream->buffer, "ts-offset", ts_offset, NULL);
   }
   GST_DEBUG_OBJECT (bin, "stream SSRC %08x, delta %" G_GINT64_FORMAT,
       stream->ssrc, ts_offset);
@@ -1123,7 +1086,7 @@ gst_rtp_bin_associate (GstRtpBin * bin, GstRtpBinStream * stream, guint8 len,
     /* combine to get the final diff to apply to the running_time */
     stream->rt_delta = rtdiff - ntpdiff;
 
-    stream_set_ts_offset (bin, stream, stream->rt_delta);
+    stream_set_ts_offset (bin, stream, stream->rt_delta, FALSE);
   } else {
     gint64 min, rtp_min, clock_base = stream->clock_base;
     gboolean all_sync, use_rtp;
@@ -1275,7 +1238,7 @@ gst_rtp_bin_associate (GstRtpBin * bin, GstRtpBinStream * stream, guint8 len,
       else
         ts_offset = ostream->rt_delta - min;
 
-      stream_set_ts_offset (bin, ostream, ts_offset);
+      stream_set_ts_offset (bin, ostream, ts_offset, TRUE);
     }
   }
   return;
@@ -1441,6 +1404,7 @@ create_stream (GstRtpBinSession * session, guint32 ssrc)
 
   /* configure latency and packet lost */
   g_object_set (buffer, "latency", rtpbin->latency_ms, NULL);
+  g_object_set (buffer, "drop-on-latency", rtpbin->drop_on_latency, NULL);
   g_object_set (buffer, "do-lost", rtpbin->do_lost, NULL);
   g_object_set (buffer, "mode", rtpbin->buffer_mode, NULL);
 
@@ -1487,12 +1451,13 @@ no_demux:
   }
 }
 
+/* called with RTP_BIN_LOCK */
 static void
-free_stream (GstRtpBinStream * stream)
+free_stream (GstRtpBinStream * stream, GstRtpBin * bin)
 {
-  GstRtpBinSession *session;
+  GSList *clients, *next_client;
 
-  session = stream->session;
+  GST_DEBUG_OBJECT (bin, "freeing stream %p", stream);
 
   if (stream->demux) {
     g_signal_handler_disconnect (stream->demux, stream->demux_newpad_sig);
@@ -1516,10 +1481,33 @@ free_stream (GstRtpBinStream * stream)
   if (stream->demux)
     g_signal_handler_disconnect (stream->demux, stream->demux_padremoved_sig);
 
-  gst_bin_remove (GST_BIN_CAST (session->bin), stream->buffer);
+  gst_bin_remove (GST_BIN_CAST (bin), stream->buffer);
   if (stream->demux)
-    gst_bin_remove (GST_BIN_CAST (session->bin), stream->demux);
+    gst_bin_remove (GST_BIN_CAST (bin), stream->demux);
+
+  for (clients = bin->clients; clients; clients = next_client) {
+    GstRtpBinClient *client = (GstRtpBinClient *) clients->data;
+    GSList *streams, *next_stream;
+
+    next_client = g_slist_next (clients);
 
+    for (streams = client->streams; streams; streams = next_stream) {
+      GstRtpBinStream *ostream = (GstRtpBinStream *) streams->data;
+
+      next_stream = g_slist_next (streams);
+
+      if (ostream == stream) {
+        client->streams = g_slist_delete_link (client->streams, streams);
+        /* If this was the last stream belonging to this client,
+         * clean up the client. */
+        if (--client->nstreams == 0) {
+          bin->clients = g_slist_delete_link (bin->clients, clients);
+          free_client (client, bin);
+          break;
+        }
+      }
+    }
+  }
   g_free (stream);
 }
 
@@ -1566,6 +1554,12 @@ gst_rtp_bin_class_init (GstRtpBinClass * klass)
           G_MAXUINT, DEFAULT_LATENCY_MS,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
+  g_object_class_install_property (gobject_class, PROP_DROP_ON_LATENCY,
+      g_param_spec_boolean ("drop-on-latency",
+          "Drop buffers when maximum latency is reached",
+          "Tells the jitterbuffer to never exceed the given latency in size",
+          DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
   /**
    * GstRtpBin::request-pt-map:
    * @rtpbin: the object which received the signal
@@ -1790,7 +1784,7 @@ gst_rtp_bin_class_init (GstRtpBinClass * klass)
 
   g_object_class_install_property (gobject_class, PROP_USE_PIPELINE_CLOCK,
       g_param_spec_boolean ("use-pipeline-clock", "Use pipeline clock",
-          "Use the pipeline clock to set the NTP time in the RTCP SR messages",
+          "Use the pipeline running-time to set the NTP time in the RTCP SR messages",
           DEFAULT_USE_PIPELINE_CLOCK,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
   /**
@@ -1807,9 +1801,11 @@ gst_rtp_bin_class_init (GstRtpBinClass * klass)
   /**
    * GstRtpBin::ntp-sync:
    *
-   * Synchronize received streams to the NTP clock. When the NTP clock is shared
-   * between the receivers and the senders (such as when using ntpd) this option
-   * can be used to synchronize receivers on multiple machines.
+   * Set the NTP time from the sender reports as the running-time on the
+   * buffers. When both the sender and receiver have sychronized
+   * running-time, i.e. when the clock and base-time is shared
+   * between the receivers and the and the senders, this option can be
+   * used to synchronize receivers on multiple machines.
    *
    * Since: 0.10.21
    */
@@ -1865,7 +1861,7 @@ gst_rtp_bin_class_init (GstRtpBinClass * klass)
   gst_element_class_add_pad_template (gstelement_class,
       gst_static_pad_template_get (&rtpbin_send_rtp_src_template));
 
-  gst_element_class_set_details_simple (gstelement_class, "RTP Bin",
+  gst_element_class_set_static_metadata (gstelement_class, "RTP Bin",
       "Filter/Network/RTP",
       "Real-Time Transport Protocol bin",
       "Wim Taymans <wim.taymans@gmail.com>");
@@ -1891,6 +1887,7 @@ gst_rtp_bin_init (GstRtpBin * rtpbin)
 
   rtpbin->latency_ms = DEFAULT_LATENCY_MS;
   rtpbin->latency_ns = DEFAULT_LATENCY_MS * GST_MSECOND;
+  rtpbin->drop_on_latency = DEFAULT_DROP_ON_LATENCY;
   rtpbin->do_lost = DEFAULT_DO_LOST;
   rtpbin->ignore_pt = DEFAULT_IGNORE_PT;
   rtpbin->ntp_sync = DEFAULT_NTP_SYNC;
@@ -1901,7 +1898,7 @@ gst_rtp_bin_init (GstRtpBin * rtpbin)
   rtpbin->use_pipeline_clock = DEFAULT_USE_PIPELINE_CLOCK;
 
   /* some default SDES entries */
-  cname = g_strdup_printf ("user%u@x-%u.net", g_random_int (), g_random_int ());
+  cname = g_strdup_printf ("user%u@host-%x", g_random_int (), g_random_int ());
   rtpbin->sdes = gst_structure_new ("application/x-rtp-source-sdes",
       "cname", G_TYPE_STRING, cname, "tool", G_TYPE_STRING, "GStreamer", NULL);
   g_free (cname);
@@ -1914,14 +1911,12 @@ gst_rtp_bin_dispose (GObject * object)
 
   rtpbin = GST_RTP_BIN (object);
 
+  GST_RTP_BIN_LOCK (rtpbin);
   GST_DEBUG_OBJECT (object, "freeing sessions");
   g_slist_foreach (rtpbin->sessions, (GFunc) free_session, rtpbin);
   g_slist_free (rtpbin->sessions);
   rtpbin->sessions = NULL;
-  GST_DEBUG_OBJECT (object, "freeing clients");
-  g_slist_foreach (rtpbin->clients, (GFunc) free_client, rtpbin);
-  g_slist_free (rtpbin->clients);
-  rtpbin->clients = NULL;
+  GST_RTP_BIN_UNLOCK (rtpbin);
 
   G_OBJECT_CLASS (parent_class)->dispose (object);
 }
@@ -1997,6 +1992,14 @@ gst_rtp_bin_set_property (GObject * object, guint prop_id,
       /* propagate the property down to the jitterbuffer */
       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin, "latency", value);
       break;
+    case PROP_DROP_ON_LATENCY:
+      GST_RTP_BIN_LOCK (rtpbin);
+      rtpbin->drop_on_latency = g_value_get_boolean (value);
+      GST_RTP_BIN_UNLOCK (rtpbin);
+      /* propagate the property down to the jitterbuffer */
+      gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin,
+          "drop-on-latency", value);
+      break;
     case PROP_SDES:
       gst_rtp_bin_set_sdes_struct (rtpbin, g_value_get_boxed (value));
       break;
@@ -2063,6 +2066,11 @@ gst_rtp_bin_get_property (GObject * object, guint prop_id,
       g_value_set_uint (value, rtpbin->latency_ms);
       GST_RTP_BIN_UNLOCK (rtpbin);
       break;
+    case PROP_DROP_ON_LATENCY:
+      GST_RTP_BIN_LOCK (rtpbin);
+      g_value_set_boolean (value, rtpbin->drop_on_latency);
+      GST_RTP_BIN_UNLOCK (rtpbin);
+      break;
     case PROP_SDES:
       g_value_take_boxed (value, gst_rtp_bin_get_sdes_struct (rtpbin));
       break;