jitterbuffer: signal timestamp discont
[platform/upstream/gstreamer.git] / gst / rtpmanager / gstrtpbin.c
index bcc904a..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.
  */
 
 /**
@@ -312,7 +312,7 @@ 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.
  *
@@ -545,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)))
@@ -552,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 */
@@ -638,8 +645,6 @@ no_demux:
 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);
@@ -656,40 +661,7 @@ free_session (GstRtpBinSession * sess, GstRtpBin * bin)
   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);
@@ -939,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;
 
@@ -968,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;
 
@@ -984,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);
@@ -1112,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;
@@ -1264,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;
@@ -1477,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);
@@ -1506,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);
 }
 
@@ -1786,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));
   /**
@@ -1803,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
    */
@@ -1916,10 +1916,6 @@ gst_rtp_bin_dispose (GObject * object)
   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);