gst/realmedia/: Add first support for parsing RDT messages.

Original commit message from CVS:
* gst/realmedia/Makefile.am:
* gst/realmedia/gstrdtbuffer.c: (gst_rdt_buffer_validate_data),
(gst_rdt_buffer_validate), (gst_rdt_buffer_get_packet_count),
(read_packet_header), (gst_rdt_buffer_get_first_packet),
(gst_rdt_packet_move_to_next), (gst_rdt_packet_get_type),
(gst_rdt_packet_get_length), (gst_rdt_packet_to_buffer),
(gst_rdt_buffer_compare_seqnum), (gst_rdt_packet_data_get_seq),
(gst_rdt_packet_data_peek_data),
(gst_rdt_packet_data_get_stream_id),
(gst_rdt_packet_data_get_timestamp):
* gst/realmedia/gstrdtbuffer.h:
Add first support for parsing RDT messages.
* gst/realmedia/rdtjitterbuffer.c: (rdt_jitter_buffer_class_init),
(rdt_jitter_buffer_init), (rdt_jitter_buffer_finalize),
(rdt_jitter_buffer_new), (rdt_jitter_buffer_reset_skew),
(calculate_skew), (rdt_jitter_buffer_insert),
(rdt_jitter_buffer_pop), (rdt_jitter_buffer_peek),
(rdt_jitter_buffer_flush), (rdt_jitter_buffer_num_packets),
(rdt_jitter_buffer_get_ts_diff):
* gst/realmedia/rdtjitterbuffer.h:
Add first version of an RDT jitterbuffer.

ChangeLog

@@ -1,3 +1,28 @@
+2008-08-27  Wim Taymans  <wim.taymans@collabora.co.uk>
+
+	* gst/realmedia/Makefile.am:
+	* gst/realmedia/gstrdtbuffer.c: (gst_rdt_buffer_validate_data),
+	(gst_rdt_buffer_validate), (gst_rdt_buffer_get_packet_count),
+	(read_packet_header), (gst_rdt_buffer_get_first_packet),
+	(gst_rdt_packet_move_to_next), (gst_rdt_packet_get_type),
+	(gst_rdt_packet_get_length), (gst_rdt_packet_to_buffer),
+	(gst_rdt_buffer_compare_seqnum), (gst_rdt_packet_data_get_seq),
+	(gst_rdt_packet_data_peek_data),
+	(gst_rdt_packet_data_get_stream_id),
+	(gst_rdt_packet_data_get_timestamp):
+	* gst/realmedia/gstrdtbuffer.h:
+	Add first support for parsing RDT messages.
+
+	* gst/realmedia/rdtjitterbuffer.c: (rdt_jitter_buffer_class_init),
+	(rdt_jitter_buffer_init), (rdt_jitter_buffer_finalize),
+	(rdt_jitter_buffer_new), (rdt_jitter_buffer_reset_skew),
+	(calculate_skew), (rdt_jitter_buffer_insert),
+	(rdt_jitter_buffer_pop), (rdt_jitter_buffer_peek),
+	(rdt_jitter_buffer_flush), (rdt_jitter_buffer_num_packets),
+	(rdt_jitter_buffer_get_ts_diff):
+	* gst/realmedia/rdtjitterbuffer.h:
+	Add first version of an RDT jitterbuffer.
+
 2008-08-27  Wim Taymans  <wim.taymans@collabora.co.uk>
 
 	* gst/realmedia/rmdemux.c: (gst_rmdemux_init),

gst/realmedia/Makefile.am

@@ -2,14 +2,16 @@ plugin_LTLIBRARIES = libgstrmdemux.la
 
 libgstrmdemux_la_SOURCES = rademux.c rmdemux.c  \
 			   rmutils.c rdtdepay.c rdtmanager.c \
-			   rtspreal.c realhash.c asmrules.c
+			   rtspreal.c realhash.c asmrules.c \
+			   rdtjitterbuffer.c gstrdtbuffer.c
 
 libgstrmdemux_la_CFLAGS = $(GST_CFLAGS) $(GST_PLUGINS_BASE_CFLAGS) $(GST_BASE_CFLAGS)
 libgstrmdemux_la_LIBADD = $(GST_PLUGINS_BASE_LIBS) $(GST_BASE_LIBS) $(GST_LIBS)\
 	     -lgstrtsp-@GST_MAJORMINOR@ -lgstsdp-@GST_MAJORMINOR@
 libgstrmdemux_la_LDFLAGS = $(GST_PLUGIN_LDFLAGS)
 
-noinst_HEADERS = rademux.h rmdemux.h rmutils.h rdtdepay.h rdtmanager.h rtspreal.h realhash.h asmrules.h
+noinst_HEADERS = rademux.h rmdemux.h rmutils.h rdtdepay.h rdtmanager.h \
+		 rdtjitterbuffer.h rtspreal.h realhash.h asmrules.h gstrdtbuffer.h
 
 noinst_PROGRAMS = asmrules
 asmrules_CFLAGS = $(GST_CFLAGS) -DTEST

gst/realmedia/gstrdtbuffer.c

@@ -0,0 +1,425 @@
+/* GStreamer
+ * Copyright (C) <2008> Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * 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.
+ */
+
+#include "gstrdtbuffer.h"
+
+gboolean
+gst_rdt_buffer_validate_data (guint8 * data, guint len)
+{
+  return TRUE;
+}
+
+gboolean
+gst_rdt_buffer_validate (GstBuffer * buffer)
+{
+  return TRUE;
+}
+
+guint
+gst_rdt_buffer_get_packet_count (GstBuffer * buffer)
+{
+  GstRDTPacket packet;
+  guint count;
+
+  g_return_val_if_fail (GST_IS_BUFFER (buffer), 0);
+
+  count = 0;
+  if (gst_rdt_buffer_get_first_packet (buffer, &packet)) {
+    do {
+      count++;
+    } while (gst_rdt_packet_move_to_next (&packet));
+  }
+  return count;
+}
+
+static gboolean
+read_packet_header (GstRDTPacket * packet)
+{
+  guint8 *data;
+  guint size;
+  guint offset;
+  guint length;
+  guint length_offset;
+
+  g_return_val_if_fail (packet != NULL, FALSE);
+  g_return_val_if_fail (GST_IS_BUFFER (packet->buffer), FALSE);
+
+  data = GST_BUFFER_DATA (packet->buffer);
+  size = GST_BUFFER_SIZE (packet->buffer);
+
+  offset = packet->offset;
+
+  /* check if we are at the end of the buffer, we add 3 because we also want to
+   * ensure we can read the type, which is always at offset 1 and 2 bytes long. */
+  if (offset + 3 > size)
+    return FALSE;
+
+  /* read type */
+  packet->type = GST_READ_UINT16_BE (&data[offset + 1]);
+
+  length = -1;
+  length_offset = -1;
+
+  /* figure out the length of the packet, this depends on the type */
+  if (GST_RDT_IS_DATA_TYPE (packet->type)) {
+    if (data[offset] & 0x80)
+      /* length is present */
+      length_offset = 3;
+  } else {
+    switch (packet->type) {
+      case GST_RDT_TYPE_ASMACTION:
+        if (data[offset] & 0x80)
+          length_offset = 5;
+        break;
+      case GST_RDT_TYPE_BWREPORT:
+        if (data[offset] & 0x80)
+          length_offset = 3;
+        break;
+      case GST_RDT_TYPE_ACK:
+        if (data[offset] & 0x80)
+          length_offset = 3;
+        break;
+      case GST_RDT_TYPE_RTTREQ:
+        length = 3;
+        break;
+      case GST_RDT_TYPE_RTTRESP:
+        length = 11;
+        break;
+      case GST_RDT_TYPE_CONGESTION:
+        length = 11;
+        break;
+      case GST_RDT_TYPE_STREAMEND:
+        length = 9;
+        /* total_reliable */
+        if (data[offset] & 0x80)
+          length += 2;
+        /* stream_id_expansion */
+        if ((data[offset] & 0x7c) == 0x7c)
+          length += 2;
+        /* ext_flag, FIXME, get string length */
+        if ((data[offset] & 0x1) == 0x1)
+          length += 7;
+        break;
+      case GST_RDT_TYPE_REPORT:
+        if (data[offset] & 0x80)
+          length_offset = 3;
+        break;
+      case GST_RDT_TYPE_LATENCY:
+        if (data[offset] & 0x80)
+          length_offset = 3;
+        break;
+      case GST_RDT_TYPE_INFOREQ:
+        length = 3;
+        /* request_time_ms */
+        if (data[offset] & 0x2)
+          length += 2;
+        break;
+      case GST_RDT_TYPE_INFORESP:
+        length = 3;
+        /* has_rtt_info */
+        if (data[offset] & 0x4) {
+          length += 4;
+          /* is_delayed */
+          if (data[offset] & 0x2) {
+            length += 4;
+          }
+        }
+        if (data[offset] & 0x1) {
+          /* buffer_info_count, FIXME read and skip */
+          length += 2;
+        }
+        break;
+      case GST_RDT_TYPE_AUTOBW:
+        if (data[offset] & 0x80)
+          length_offset = 3;
+        break;
+      case GST_RDT_TYPE_INVALID:
+      default:
+        goto unknown_packet;
+    }
+  }
+
+  if (length != -1) {
+    /* we have a fixed length */
+    packet->length = length;
+  } else if (length_offset != -1) {
+    /* we can read the length from an offset */
+    packet->length = GST_READ_UINT16_BE (&data[length_offset]);
+  } else {
+    /* length is remainder of packet */
+    packet->length = size - offset;
+  }
+
+  /* the length should be smaller than the remaining size */
+  if (packet->length + offset > size)
+    goto invalid_length;
+
+  return TRUE;
+
+  /* ERRORS */
+unknown_packet:
+  {
+    packet->type = GST_RDT_TYPE_INVALID;
+    return FALSE;
+  }
+invalid_length:
+  {
+    packet->type = GST_RDT_TYPE_INVALID;
+    packet->length = 0;
+    return FALSE;
+  }
+}
+
+gboolean
+gst_rdt_buffer_get_first_packet (GstBuffer * buffer, GstRDTPacket * packet)
+{
+  g_return_val_if_fail (GST_IS_BUFFER (buffer), FALSE);
+  g_return_val_if_fail (packet != NULL, FALSE);
+
+  /* init to 0 */
+  packet->buffer = buffer;
+  packet->offset = 0;
+  packet->type = GST_RDT_TYPE_INVALID;
+
+  if (!read_packet_header (packet))
+    return FALSE;
+
+  return TRUE;
+}
+
+gboolean
+gst_rdt_packet_move_to_next (GstRDTPacket * packet)
+{
+  g_return_val_if_fail (packet != NULL, FALSE);
+  g_return_val_if_fail (packet->type != GST_RDT_TYPE_INVALID, FALSE);
+  g_return_val_if_fail (GST_IS_BUFFER (packet->buffer), FALSE);
+
+  /* if we have an invalid packet, it must be the last, 
+   * return FALSE */
+  if (packet->type == GST_RDT_TYPE_INVALID)
+    goto end;
+
+  /* move to next packet */
+  packet->offset += packet->length;
+
+  /* try to read new header */
+  if (!read_packet_header (packet))
+    goto end;
+
+  return TRUE;
+
+  /* ERRORS */
+end:
+  {
+    packet->type = GST_RDT_TYPE_INVALID;
+    return FALSE;
+  }
+}
+
+GstRDTType
+gst_rdt_packet_get_type (GstRDTPacket * packet)
+{
+  g_return_val_if_fail (packet != NULL, GST_RDT_TYPE_INVALID);
+  g_return_val_if_fail (packet->type != GST_RDT_TYPE_INVALID,
+      GST_RDT_TYPE_INVALID);
+
+  return packet->type;
+}
+
+guint16
+gst_rdt_packet_get_length (GstRDTPacket * packet)
+{
+  g_return_val_if_fail (packet != NULL, 0);
+  g_return_val_if_fail (packet->type != GST_RDT_TYPE_INVALID, 0);
+
+  return packet->length;
+}
+
+GstBuffer *
+gst_rdt_packet_to_buffer (GstRDTPacket * packet)
+{
+  GstBuffer *result;
+
+  g_return_val_if_fail (packet != NULL, NULL);
+  g_return_val_if_fail (packet->type != GST_RDT_TYPE_INVALID, NULL);
+
+  result =
+      gst_buffer_create_sub (packet->buffer, packet->offset, packet->length);
+  /* timestamp applies to all packets in this buffer */
+  GST_BUFFER_TIMESTAMP (result) = GST_BUFFER_TIMESTAMP (packet->buffer);
+
+  return result;
+}
+
+gint
+gst_rdt_buffer_compare_seqnum (guint16 seqnum1, guint16 seqnum2)
+{
+  return (gint16) (seqnum2 - seqnum1);
+}
+
+guint16
+gst_rdt_packet_data_get_seq (GstRDTPacket * packet)
+{
+  guint header;
+  guint8 *bufdata;
+  guint bufsize;
+  guint16 result;
+
+  g_return_val_if_fail (packet != NULL, FALSE);
+  g_return_val_if_fail (GST_RDT_IS_DATA_TYPE (packet->type), FALSE);
+
+  bufdata = GST_BUFFER_DATA (packet->buffer);
+  bufsize = GST_BUFFER_SIZE (packet->buffer);
+
+  /* skip header bits */
+  header = packet->offset + 1;
+
+  /* read seq_no */
+  result = GST_READ_UINT16_BE (&bufdata[header]);
+
+  return result;
+}
+
+gboolean
+gst_rdt_packet_data_peek_data (GstRDTPacket * packet, guint8 ** data,
+    guint * size)
+{
+  guint header;
+  guint8 *bufdata;
+  guint bufsize;
+  gboolean length_included_flag;
+  gboolean need_reliable_flag;
+  guint8 stream_id;
+  guint8 asm_rule_number;
+
+  g_return_val_if_fail (packet != NULL, FALSE);
+  g_return_val_if_fail (GST_RDT_IS_DATA_TYPE (packet->type), FALSE);
+
+  bufdata = GST_BUFFER_DATA (packet->buffer);
+  bufsize = GST_BUFFER_SIZE (packet->buffer);
+
+  header = packet->offset;
+
+  length_included_flag = (bufdata[header] & 0x80) == 0x80;
+  need_reliable_flag = (bufdata[header] & 0x40) == 0x40;
+  stream_id = (bufdata[header] & 0x3e) >> 1;
+
+  /* skip seq_no and header bits */
+  header += 3;
+
+  if (length_included_flag) {
+    /* skip length */
+    header += 2;
+  }
+  asm_rule_number = (bufdata[header] & 0x3f);
+
+  /* skip timestamp and asm_rule_number */
+  header += 5;
+
+  if (stream_id == 0x1f) {
+    /* skip stream_id_expansion */
+    header += 2;
+  }
+  if (need_reliable_flag) {
+    /* skip total_reliable */
+    header += 2;
+  }
+  if (asm_rule_number == 63) {
+    /* skip asm_rule_number_expansion */
+    header += 2;
+  }
+
+  if (data)
+    *data = &bufdata[header];
+  if (size)
+    *size = packet->length - (header - packet->offset);
+
+  return TRUE;
+}
+
+guint16
+gst_rdt_packet_data_get_stream_id (GstRDTPacket * packet)
+{
+  guint16 result;
+  guint header;
+  gboolean length_included_flag;
+  guint8 *bufdata;
+  guint bufsize;
+
+  g_return_val_if_fail (packet != NULL, 0);
+  g_return_val_if_fail (GST_RDT_IS_DATA_TYPE (packet->type), 0);
+
+  bufdata = GST_BUFFER_DATA (packet->buffer);
+  bufsize = GST_BUFFER_SIZE (packet->buffer);
+
+  header = packet->offset;
+
+  length_included_flag = (bufdata[header] & 0x80) == 0x80;
+  result = (bufdata[header] & 0x3e) >> 1;
+  if (result == 31) {
+    /* skip seq_no and header bits */
+    header += 3;
+
+    if (length_included_flag) {
+      /* skip length */
+      header += 2;
+    }
+    /* skip asm_rule_number and timestamp */
+    header += 5;
+
+    /* stream_id_expansion */
+    result = GST_READ_UINT16_BE (&bufdata[header]);
+  }
+  return result;
+}
+
+guint32
+gst_rdt_packet_data_get_timestamp (GstRDTPacket * packet)
+{
+  guint32 result;
+  guint header;
+  gboolean length_included_flag;
+  guint8 *bufdata;
+  guint bufsize;
+
+  g_return_val_if_fail (packet != NULL, 0);
+  g_return_val_if_fail (GST_RDT_IS_DATA_TYPE (packet->type), 0);
+
+  bufdata = GST_BUFFER_DATA (packet->buffer);
+  bufsize = GST_BUFFER_SIZE (packet->buffer);
+
+  header = packet->offset;
+
+  length_included_flag = (bufdata[header] & 0x80) == 0x80;
+
+  /* skip seq_no and header bits */
+  header += 3;
+
+  if (length_included_flag) {
+    /* skip length */
+    header += 2;
+  }
+  /* skip asm_rule_number */
+  header += 1;
+
+  /* get timestamp */
+  result = GST_READ_UINT32_BE (&bufdata[header]);
+
+  return result;
+}

gst/realmedia/gstrdtbuffer.h

@@ -0,0 +1,118 @@
+/* GStreamer
+ * Copyright (C) <2008> Wim Taymans <wim.taymans@gmail.com>
+ *
+ * gstrdtbuffer.h: various helper functions to manipulate buffers
+ *     with RDT payload.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * 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.
+ */
+
+#ifndef __GST_RDTBUFFER_H__
+#define __GST_RDTBUFFER_H__
+
+#include <gst/gst.h>
+
+G_BEGIN_DECLS
+
+/** 
+ * GstRDTType:
+ * @GST_RDT_TYPE_INVALID:
+ * @GST_RDT_TYPE_ASMACTION:
+ * @GST_RDT_TYPE_ACK:
+ * @GST_RDT_TYPE_RTTREQ:
+ * @GST_RDT_TYPE_RTTRESP:
+ * @GST_RDT_TYPE_CONGESTION:
+ * @GST_RDT_TYPE_STREAMEND:
+ * @GST_RDT_TYPE_LATENCY:
+ * @GST_RDT_TYPE_INFOREQ:
+ * @GST_RDT_TYPE_INFORESP:
+ * @GST_RDT_TYPE_AUTOBW:
+ *
+ * Different RDT packet types.
+ */
+typedef enum
+{
+  GST_RDT_TYPE_INVALID     = 0xffff,
+  GST_RDT_TYPE_ASMACTION   = 0xff00,
+  GST_RDT_TYPE_BWREPORT    = 0xff01,
+  GST_RDT_TYPE_ACK         = 0xff02,
+  GST_RDT_TYPE_RTTREQ      = 0xff03,
+  GST_RDT_TYPE_RTTRESP     = 0xff04,
+  GST_RDT_TYPE_CONGESTION  = 0xff05,
+  GST_RDT_TYPE_STREAMEND   = 0xff06,
+  GST_RDT_TYPE_REPORT      = 0xff07,
+  GST_RDT_TYPE_LATENCY     = 0xff08,
+  GST_RDT_TYPE_INFOREQ     = 0xff09,
+  GST_RDT_TYPE_INFORESP    = 0xff0a,
+  GST_RDT_TYPE_AUTOBW      = 0xff0b
+} GstRDTType;
+
+/**
+ * GST_RDT_IS_DATA_TYPE:
+ * @t: the #GstRDTType to check
+ * 
+ * Check if @t is a data packet type.
+ */
+#define GST_RDT_IS_DATA_TYPE(t) ((t) < 0xff00)
+
+typedef struct _GstRDTPacket GstRDTPacket;
+
+/**
+ * GstRDTPacket:
+ * @buffer: pointer to RDT buffer
+ * @offset: offset of packet in buffer data
+ *
+ * Data structure that points to a packet at @offset in @buffer. 
+ * The size of the structure is made public to allow stack allocations.
+ */
+struct _GstRDTPacket
+{ 
+  GstBuffer   *buffer;
+  guint        offset;
+  
+  /*< private >*/
+  GstRDTType   type;         /* type of current packet */
+  guint16      length;       /* length of current packet in bytes */
+};
+
+/* validate buffers */
+gboolean        gst_rdt_buffer_validate_data      (guint8 *data, guint len);
+gboolean        gst_rdt_buffer_validate           (GstBuffer *buffer);
+
+/* retrieving packets */
+guint           gst_rdt_buffer_get_packet_count   (GstBuffer *buffer);
+gboolean        gst_rdt_buffer_get_first_packet   (GstBuffer *buffer, GstRDTPacket *packet);
+gboolean        gst_rdt_packet_move_to_next       (GstRDTPacket *packet);
+
+/* working with packets */
+GstRDTType      gst_rdt_packet_get_type           (GstRDTPacket *packet);
+guint16         gst_rdt_packet_get_length         (GstRDTPacket *packet);
+GstBuffer*      gst_rdt_packet_to_buffer          (GstRDTPacket *packet);
+
+
+/* data packets */
+guint16         gst_rdt_packet_data_get_seq       (GstRDTPacket *packet);
+gboolean        gst_rdt_packet_data_peek_data     (GstRDTPacket *packet, guint8 **data, guint *size);
+guint16         gst_rdt_packet_data_get_stream_id (GstRDTPacket *packet);
+guint32         gst_rdt_packet_data_get_timestamp (GstRDTPacket *packet);
+
+/* utils */
+gint            gst_rdt_buffer_compare_seqnum     (guint16 seqnum1, guint16 seqnum2);
+
+G_END_DECLS
+
+#endif /* __GST_RDTBUFFER_H__ */
+

gst/realmedia/rdtjitterbuffer.c

@@ -0,0 +1,531 @@
+/* GStreamer
+ * Copyright (C) <2007> Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * 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.
+ */
+#include <string.h>
+#include <stdlib.h>
+
+#include "rdtjitterbuffer.h"
+#include "gstrdtbuffer.h"
+
+GST_DEBUG_CATEGORY_STATIC (rdt_jitter_buffer_debug);
+#define GST_CAT_DEFAULT rdt_jitter_buffer_debug
+
+#define MAX_WINDOW	RDT_JITTER_BUFFER_MAX_WINDOW
+#define MAX_TIME	(2 * GST_SECOND)
+
+/* signals and args */
+enum
+{
+  LAST_SIGNAL
+};
+
+enum
+{
+  PROP_0
+};
+
+/* GObject vmethods */
+static void rdt_jitter_buffer_finalize (GObject * object);
+
+/* static guint rdt_jitter_buffer_signals[LAST_SIGNAL] = { 0 }; */
+
+G_DEFINE_TYPE (RDTJitterBuffer, rdt_jitter_buffer, G_TYPE_OBJECT);
+
+static void
+rdt_jitter_buffer_class_init (RDTJitterBufferClass * klass)
+{
+  GObjectClass *gobject_class;
+
+  gobject_class = (GObjectClass *) klass;
+
+  gobject_class->finalize = rdt_jitter_buffer_finalize;
+
+  GST_DEBUG_CATEGORY_INIT (rdt_jitter_buffer_debug, "rdtjitterbuffer", 0,
+      "RDT Jitter Buffer");
+}
+
+static void
+rdt_jitter_buffer_init (RDTJitterBuffer * jbuf)
+{
+  jbuf->packets = g_queue_new ();
+
+  rdt_jitter_buffer_reset_skew (jbuf);
+}
+
+static void
+rdt_jitter_buffer_finalize (GObject * object)
+{
+  RDTJitterBuffer *jbuf;
+
+  jbuf = RDT_JITTER_BUFFER_CAST (object);
+
+  rdt_jitter_buffer_flush (jbuf);
+  g_queue_free (jbuf->packets);
+
+  G_OBJECT_CLASS (rdt_jitter_buffer_parent_class)->finalize (object);
+}
+
+/**
+ * rdt_jitter_buffer_new:
+ *
+ * Create an #RDTJitterBuffer.
+ *
+ * Returns: a new #RDTJitterBuffer. Use g_object_unref() after usage.
+ */
+RDTJitterBuffer *
+rdt_jitter_buffer_new (void)
+{
+  RDTJitterBuffer *jbuf;
+
+  jbuf = g_object_new (RDT_TYPE_JITTER_BUFFER, NULL);
+
+  return jbuf;
+}
+
+void
+rdt_jitter_buffer_reset_skew (RDTJitterBuffer * jbuf)
+{
+  jbuf->base_time = -1;
+  jbuf->base_rtptime = -1;
+  jbuf->ext_rtptime = -1;
+  jbuf->window_pos = 0;
+  jbuf->window_filling = TRUE;
+  jbuf->window_min = 0;
+  jbuf->skew = 0;
+  jbuf->prev_send_diff = -1;
+}
+
+/* For the clock skew we use a windowed low point averaging algorithm as can be
+ * found in http://www.grame.fr/pub/TR-050601.pdf. The idea is that the jitter is
+ * composed of:
+ *
+ *  J = N + n
+ *
+ *   N   : a constant network delay.
+ *   n   : random added noise. The noise is concentrated around 0
+ *
+ * In the receiver we can track the elapsed time at the sender with:
+ *
+ *  send_diff(i) = (Tsi - Ts0);
+ *
+ *   Tsi : The time at the sender at packet i
+ *   Ts0 : The time at the sender at the first packet
+ *
+ * This is the difference between the RDT timestamp in the first received packet
+ * and the current packet.
+ *
+ * At the receiver we have to deal with the jitter introduced by the network.
+ *
+ *  recv_diff(i) = (Tri - Tr0)
+ *
+ *   Tri : The time at the receiver at packet i
+ *   Tr0 : The time at the receiver at the first packet
+ *
+ * Both of these values contain a jitter Ji, a jitter for packet i, so we can
+ * write:
+ *
+ *  recv_diff(i) = (Cri + D + ni) - (Cr0 + D + n0))
+ *
+ *    Cri    : The time of the clock at the receiver for packet i
+ *    D + ni : The jitter when receiving packet i
+ *
+ * We see that the network delay is irrelevant here as we can elliminate D:
+ *
+ *  recv_diff(i) = (Cri + ni) - (Cr0 + n0))
+ *
+ * The drift is now expressed as:
+ *
+ *  Drift(i) = recv_diff(i) - send_diff(i);
+ *
+ * We now keep the W latest values of Drift and find the minimum (this is the
+ * one with the lowest network jitter and thus the one which is least affected
+ * by it). We average this lowest value to smooth out the resulting network skew.
+ *
+ * Both the window and the weighting used for averaging influence the accuracy
+ * of the drift estimation. Finding the correct parameters turns out to be a
+ * compromise between accuracy and inertia. 
+ *
+ * We use a 2 second window or up to 512 data points, which is statistically big
+ * enough to catch spikes (FIXME, detect spikes).
+ * We also use a rather large weighting factor (125) to smoothly adapt. During
+ * startup, when filling the window, we use a parabolic weighting factor, the
+ * more the window is filled, the faster we move to the detected possible skew.
+ *
+ * Returns: @time adjusted with the clock skew.
+ */
+static GstClockTime
+calculate_skew (RDTJitterBuffer * jbuf, guint32 rtptime, GstClockTime time,
+    guint32 clock_rate)
+{
+  guint64 ext_rtptime;
+  guint64 send_diff, recv_diff;
+  gint64 delta;
+  gint64 old;
+  gint pos, i;
+  GstClockTime gstrtptime, out_time;
+
+  //ext_rtptime = gst_rtp_buffer_ext_timestamp (&jbuf->ext_rtptime, rtptime);
+  ext_rtptime = rtptime;
+
+  gstrtptime = gst_util_uint64_scale_int (ext_rtptime, GST_SECOND, clock_rate);
+
+again:
+  /* first time, lock on to time and gstrtptime */
+  if (jbuf->base_time == -1)
+    jbuf->base_time = time;
+  if (jbuf->base_rtptime == -1)
+    jbuf->base_rtptime = gstrtptime;
+
+  if (gstrtptime >= jbuf->base_rtptime)
+    send_diff = gstrtptime - jbuf->base_rtptime;
+  else {
+    /* elapsed time at sender, timestamps can go backwards and thus be smaller
+     * than our base time, take a new base time in that case. */
+    GST_DEBUG ("backward timestamps at server, taking new base time");
+    jbuf->base_rtptime = gstrtptime;
+    jbuf->base_time = time;
+    send_diff = 0;
+  }
+
+  GST_DEBUG ("extrtp %" G_GUINT64_FORMAT ", gstrtp %" GST_TIME_FORMAT ", base %"
+      GST_TIME_FORMAT ", send_diff %" GST_TIME_FORMAT, ext_rtptime,
+      GST_TIME_ARGS (gstrtptime), GST_TIME_ARGS (jbuf->base_rtptime),
+      GST_TIME_ARGS (send_diff));
+
+  if (jbuf->prev_send_diff != -1 && time != -1) {
+    gint64 delta_diff;
+
+    if (send_diff > jbuf->prev_send_diff)
+      delta_diff = send_diff - jbuf->prev_send_diff;
+    else
+      delta_diff = jbuf->prev_send_diff - send_diff;
+
+    /* server changed rtp timestamps too quickly, reset skew detection and start
+     * again. This value is sortof arbitrary and can be a bad measurement up if
+     * there are many packets missing because then we get a big gap that is
+     * unrelated to a timestamp switch. */
+    if (delta_diff > GST_SECOND) {
+      GST_DEBUG ("delta changed too quickly %" GST_TIME_FORMAT " reset skew",
+          GST_TIME_ARGS (delta_diff));
+      rdt_jitter_buffer_reset_skew (jbuf);
+      goto again;
+    }
+  }
+  jbuf->prev_send_diff = send_diff;
+
+  /* we don't have an arrival timestamp so we can't do skew detection. we
+   * should still apply a timestamp based on RDT timestamp and base_time */
+  if (time == -1)
+    goto no_skew;
+
+  /* elapsed time at receiver, includes the jitter */
+  recv_diff = time - jbuf->base_time;
+
+  GST_DEBUG ("time %" GST_TIME_FORMAT ", base %" GST_TIME_FORMAT ", recv_diff %"
+      GST_TIME_FORMAT, GST_TIME_ARGS (time), GST_TIME_ARGS (jbuf->base_time),
+      GST_TIME_ARGS (recv_diff));
+
+  /* measure the diff */
+  delta = ((gint64) recv_diff) - ((gint64) send_diff);
+
+  pos = jbuf->window_pos;
+
+  if (jbuf->window_filling) {
+    /* we are filling the window */
+    GST_DEBUG ("filling %d, delta %" G_GINT64_FORMAT, pos, delta);
+    jbuf->window[pos++] = delta;
+    /* calc the min delta we observed */
+    if (pos == 1 || delta < jbuf->window_min)
+      jbuf->window_min = delta;
+
+    if (send_diff >= MAX_TIME || pos >= MAX_WINDOW) {
+      jbuf->window_size = pos;
+
+      /* window filled */
+      GST_DEBUG ("min %" G_GINT64_FORMAT, jbuf->window_min);
+
+      /* the skew is now the min */
+      jbuf->skew = jbuf->window_min;
+      jbuf->window_filling = FALSE;
+    } else {
+      gint perc_time, perc_window, perc;
+
+      /* figure out how much we filled the window, this depends on the amount of
+       * time we have or the max number of points we keep. */
+      perc_time = send_diff * 100 / MAX_TIME;
+      perc_window = pos * 100 / MAX_WINDOW;
+      perc = MAX (perc_time, perc_window);
+
+      /* make a parabolic function, the closer we get to the MAX, the more value
+       * we give to the scaling factor of the new value */
+      perc = perc * perc;
+
+      /* quickly go to the min value when we are filling up, slowly when we are
+       * just starting because we're not sure it's a good value yet. */
+      jbuf->skew =
+          (perc * jbuf->window_min + ((10000 - perc) * jbuf->skew)) / 10000;
+      jbuf->window_size = pos + 1;
+    }
+  } else {
+    /* pick old value and store new value. We keep the previous value in order
+     * to quickly check if the min of the window changed */
+    old = jbuf->window[pos];
+    jbuf->window[pos++] = delta;
+
+    if (delta <= jbuf->window_min) {
+      /* if the new value we inserted is smaller or equal to the current min,
+       * it becomes the new min */
+      jbuf->window_min = delta;
+    } else if (old == jbuf->window_min) {
+      gint64 min = G_MAXINT64;
+
+      /* if we removed the old min, we have to find a new min */
+      for (i = 0; i < jbuf->window_size; i++) {
+        /* we found another value equal to the old min, we can stop searching now */
+        if (jbuf->window[i] == old) {
+          min = old;
+          break;
+        }
+        if (jbuf->window[i] < min)
+          min = jbuf->window[i];
+      }
+      jbuf->window_min = min;
+    }
+    /* average the min values */
+    jbuf->skew = (jbuf->window_min + (124 * jbuf->skew)) / 125;
+    GST_DEBUG ("delta %" G_GINT64_FORMAT ", new min: %" G_GINT64_FORMAT,
+        delta, jbuf->window_min);
+  }
+  /* wrap around in the window */
+  if (pos >= jbuf->window_size)
+    pos = 0;
+  jbuf->window_pos = pos;
+
+no_skew:
+  /* the output time is defined as the base timestamp plus the RDT time
+   * adjusted for the clock skew .*/
+  out_time = jbuf->base_time + send_diff + jbuf->skew;
+
+  GST_DEBUG ("skew %" G_GINT64_FORMAT ", out %" GST_TIME_FORMAT,
+      jbuf->skew, GST_TIME_ARGS (out_time));
+
+  return out_time;
+}
+
+/**
+ * rdt_jitter_buffer_insert:
+ * @jbuf: an #RDTJitterBuffer
+ * @buf: a buffer
+ * @time: a running_time when this buffer was received in nanoseconds
+ * @clock_rate: the clock-rate of the payload of @buf
+ * @tail: TRUE when the tail element changed.
+ *
+ * Inserts @buf into the packet queue of @jbuf. The sequence number of the
+ * packet will be used to sort the packets. This function takes ownerhip of
+ * @buf when the function returns %TRUE.
+ * @buf should have writable metadata when calling this function.
+ *
+ * Returns: %FALSE if a packet with the same number already existed.
+ */
+gboolean
+rdt_jitter_buffer_insert (RDTJitterBuffer * jbuf, GstBuffer * buf,
+    GstClockTime time, guint32 clock_rate, gboolean * tail)
+{
+  GList *list;
+  guint32 rtptime;
+  guint16 seqnum;
+  GstRDTPacket packet;
+  gboolean more;
+
+  g_return_val_if_fail (jbuf != NULL, FALSE);
+  g_return_val_if_fail (buf != NULL, FALSE);
+
+  more = gst_rdt_buffer_get_first_packet (buf, &packet);
+  /* programmer error */
+  g_return_val_if_fail (more == TRUE, FALSE);
+
+  seqnum = gst_rdt_packet_data_get_seq (&packet);
+
+  /* loop the list to skip strictly smaller seqnum buffers */
+  for (list = jbuf->packets->head; list; list = g_list_next (list)) {
+    guint16 qseq;
+    gint gap;
+
+    more =
+        gst_rdt_buffer_get_first_packet (GST_BUFFER_CAST (list->data), &packet);
+    /* programmer error */
+    g_return_val_if_fail (more == TRUE, FALSE);
+
+    qseq = gst_rdt_packet_data_get_seq (&packet);
+
+    /* compare the new seqnum to the one in the buffer */
+    gap = gst_rdt_buffer_compare_seqnum (seqnum, qseq);
+
+    /* we hit a packet with the same seqnum, notify a duplicate */
+    if (G_UNLIKELY (gap == 0))
+      goto duplicate;
+
+    /* seqnum > qseq, we can stop looking */
+    if (G_LIKELY (gap < 0))
+      break;
+  }
+
+  /* do skew calculation by measuring the difference between rtptime and the
+   * receive time, this function will retimestamp @buf with the skew corrected
+   * running time. */
+  //rtptime = gst_rtp_buffer_get_timestamp (buf);
+  rtptime = 0;
+  if (clock_rate) {
+    time = calculate_skew (jbuf, rtptime, time, clock_rate);
+    GST_BUFFER_TIMESTAMP (buf) = time;
+  }
+
+  if (list)
+    g_queue_insert_before (jbuf->packets, list, buf);
+  else
+    g_queue_push_tail (jbuf->packets, buf);
+
+  /* tail was changed when we did not find a previous packet, we set the return
+   * flag when requested. */
+  if (tail)
+    *tail = (list == NULL);
+
+  return TRUE;
+
+  /* ERRORS */
+duplicate:
+  {
+    GST_WARNING ("duplicate packet %d found", (gint) seqnum);
+    return FALSE;
+  }
+}
+
+/**
+ * rdt_jitter_buffer_pop:
+ * @jbuf: an #RDTJitterBuffer
+ *
+ * Pops the oldest buffer from the packet queue of @jbuf. The popped buffer will
+ * have its timestamp adjusted with the incomming running_time and the detected
+ * clock skew.
+ *
+ * Returns: a #GstBuffer or %NULL when there was no packet in the queue.
+ */
+GstBuffer *
+rdt_jitter_buffer_pop (RDTJitterBuffer * jbuf)
+{
+  GstBuffer *buf;
+
+  g_return_val_if_fail (jbuf != NULL, FALSE);
+
+  buf = g_queue_pop_tail (jbuf->packets);
+
+  return buf;
+}
+
+/**
+ * rdt_jitter_buffer_peek:
+ * @jbuf: an #RDTJitterBuffer
+ *
+ * Peek the oldest buffer from the packet queue of @jbuf. Register a callback
+ * with rdt_jitter_buffer_set_tail_changed() to be notified when an older packet
+ * was inserted in the queue.
+ *
+ * Returns: a #GstBuffer or %NULL when there was no packet in the queue.
+ */
+GstBuffer *
+rdt_jitter_buffer_peek (RDTJitterBuffer * jbuf)
+{
+  GstBuffer *buf;
+
+  g_return_val_if_fail (jbuf != NULL, FALSE);
+
+  buf = g_queue_peek_tail (jbuf->packets);
+
+  return buf;
+}
+
+/**
+ * rdt_jitter_buffer_flush:
+ * @jbuf: an #RDTJitterBuffer
+ *
+ * Flush all packets from the jitterbuffer.
+ */
+void
+rdt_jitter_buffer_flush (RDTJitterBuffer * jbuf)
+{
+  GstBuffer *buffer;
+
+  g_return_if_fail (jbuf != NULL);
+
+  while ((buffer = g_queue_pop_head (jbuf->packets)))
+    gst_buffer_unref (buffer);
+}
+
+/**
+ * rdt_jitter_buffer_num_packets:
+ * @jbuf: an #RDTJitterBuffer
+ *
+ * Get the number of packets currently in "jbuf.
+ *
+ * Returns: The number of packets in @jbuf.
+ */
+guint
+rdt_jitter_buffer_num_packets (RDTJitterBuffer * jbuf)
+{
+  g_return_val_if_fail (jbuf != NULL, 0);
+
+  return jbuf->packets->length;
+}
+
+/**
+ * rdt_jitter_buffer_get_ts_diff:
+ * @jbuf: an #RDTJitterBuffer
+ *
+ * Get the difference between the timestamps of first and last packet in the
+ * jitterbuffer.
+ *
+ * Returns: The difference expressed in the timestamp units of the packets.
+ */
+guint32
+rdt_jitter_buffer_get_ts_diff (RDTJitterBuffer * jbuf)
+{
+  guint64 high_ts, low_ts;
+  GstBuffer *high_buf, *low_buf;
+  guint32 result;
+
+  g_return_val_if_fail (jbuf != NULL, 0);
+
+  high_buf = g_queue_peek_head (jbuf->packets);
+  low_buf = g_queue_peek_tail (jbuf->packets);
+
+  if (!high_buf || !low_buf || high_buf == low_buf)
+    return 0;
+
+  //high_ts = gst_rtp_buffer_get_timestamp (high_buf);
+  //low_ts = gst_rtp_buffer_get_timestamp (low_buf);
+  high_ts = 0;
+  low_ts = 0;
+
+  /* it needs to work if ts wraps */
+  if (high_ts >= low_ts) {
+    result = (guint32) (high_ts - low_ts);
+  } else {
+    result = (guint32) (high_ts + G_MAXUINT32 + 1 - low_ts);
+  }
+  return result;
+}

gst/realmedia/rdtjitterbuffer.h

@@ -0,0 +1,91 @@
+/* GStreamer
+ * Copyright (C) <2007> Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * 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.
+ */
+
+#ifndef __RDT_JITTER_BUFFER_H__
+#define __RDT_JITTER_BUFFER_H__
+
+#include <gst/gst.h>
+
+typedef struct _RDTJitterBuffer RDTJitterBuffer;
+typedef struct _RDTJitterBufferClass RDTJitterBufferClass;
+
+#define RDT_TYPE_JITTER_BUFFER             (rdt_jitter_buffer_get_type())
+#define RDT_JITTER_BUFFER(src)             (G_TYPE_CHECK_INSTANCE_CAST((src),RDT_TYPE_JITTER_BUFFER,RDTJitterBuffer))
+#define RDT_JITTER_BUFFER_CLASS(klass)     (G_TYPE_CHECK_CLASS_CAST((klass),RDT_TYPE_JITTER_BUFFER,RDTJitterBufferClass))
+#define RDT_IS_JITTER_BUFFER(src)          (G_TYPE_CHECK_INSTANCE_TYPE((src),RDT_TYPE_JITTER_BUFFER))
+#define RDT_IS_JITTER_BUFFER_CLASS(klass)  (G_TYPE_CHECK_CLASS_TYPE((klass),RDT_TYPE_JITTER_BUFFER))
+#define RDT_JITTER_BUFFER_CAST(src)        ((RDTJitterBuffer *)(src))
+
+/**
+ * RTPTailChanged:
+ * @jbuf: an #RDTJitterBuffer
+ * @user_data: user data specified when registering
+ *
+ * This callback will be called when the tail buffer of @jbuf changed.
+ */
+typedef void (*RTPTailChanged) (RDTJitterBuffer *jbuf, gpointer user_data);
+
+#define RDT_JITTER_BUFFER_MAX_WINDOW 512
+/**
+ * RDTJitterBuffer:
+ *
+ * A JitterBuffer in the #RTPSession
+ */
+struct _RDTJitterBuffer {
+  GObject        object;
+
+  GQueue        *packets;
+
+  /* for calculating skew */
+  GstClockTime   base_time;
+  GstClockTime   base_rtptime;
+  guint64        ext_rtptime;
+  gint64         window[RDT_JITTER_BUFFER_MAX_WINDOW];
+  guint          window_pos;
+  guint          window_size;
+  gboolean       window_filling;
+  gint64         window_min;
+  gint64         skew;
+  gint64         prev_send_diff;
+};
+
+struct _RDTJitterBufferClass {
+  GObjectClass   parent_class;
+};
+
+GType rdt_jitter_buffer_get_type (void);
+
+/* managing lifetime */
+RDTJitterBuffer*      rdt_jitter_buffer_new              (void);
+
+void                  rdt_jitter_buffer_reset_skew       (RDTJitterBuffer *jbuf);
+
+gboolean              rdt_jitter_buffer_insert           (RDTJitterBuffer *jbuf, GstBuffer *buf,
+		                                          GstClockTime time,
+		                                          guint32 clock_rate,
+		                                          gboolean *tail);
+GstBuffer *           rdt_jitter_buffer_peek             (RDTJitterBuffer *jbuf);
+GstBuffer *           rdt_jitter_buffer_pop              (RDTJitterBuffer *jbuf);
+
+void                  rdt_jitter_buffer_flush            (RDTJitterBuffer *jbuf);
+
+guint                 rdt_jitter_buffer_num_packets      (RDTJitterBuffer *jbuf);
+guint32               rdt_jitter_buffer_get_ts_diff      (RDTJitterBuffer *jbuf);
+
+#endif /* __RDT_JITTER_BUFFER_H__ */