*** plugins/elements/gsttee.c	Tue Jan 12 16:35:17 2010
--- /home/user/Development/videoconf/data/bug/gstxvimage/gsttee_prototype.c	Wed Dec 29 13:28:47 2010
***************
*** 33,38 ****
--- 33,39 ----
  #endif
  
  #include "gsttee.h"
+ #include "gst/gst.h"
  
  #include <string.h>
  
***************
*** 45,50 ****
--- 46,52 ----
  #define GST_CAT_DEFAULT gst_tee_debug
  
  #define GST_TYPE_TEE_PULL_MODE (gst_tee_pull_mode_get_type())
+ 
  static GType
  gst_tee_pull_mode_get_type (void)
  {
***************
*** 73,78 ****
--- 75,86 ----
  #define DEFAULT_PROP_LAST_MESSAGE	NULL
  #define DEFAULT_PULL_MODE		GST_TEE_PULL_MODE_NEVER
  
+ #define GST_TEE_QOS_OBSCURED_RATE_SCALE  100
+ #define GST_TEE_QOS_OBSCURED_DELAY_COUNT 3
+ #define DO_RUNNING_AVG(avg,val,size)     (((val) + ((size)-1) * (avg)) / (size))
+ /* generic running average, this has a neutral window size */
+ #define UPDATE_RUNNING_AVG(avg,val)      DO_RUNNING_AVG(avg,val,8)
+ 
  enum
  {
    PROP_0,
***************
*** 102,107 ****
--- 110,119 ----
  {
    gboolean pushed;
    GstFlowReturn result;
+   GstClockTime last_buftime;
+   gint avg_rate;
+   gint qos_rate;
+   gint qos_count;
    gboolean removed;
  } PushData;
  
***************
*** 126,132 ****
  static gboolean gst_tee_src_activate_pull (GstPad * pad, gboolean active);
  static GstFlowReturn gst_tee_src_get_range (GstPad * pad, guint64 offset,
      guint length, GstBuffer ** buf);
! 
  
  static void
  gst_tee_base_init (gpointer g_class)
--- 138,145 ----
  static gboolean gst_tee_src_activate_pull (GstPad * pad, gboolean active);
  static GstFlowReturn gst_tee_src_get_range (GstPad * pad, guint64 offset,
      guint length, GstBuffer ** buf);
! static gboolean gst_tee_src_event (GstPad * pad, GstEvent * event);
! static gboolean gst_tee_qos_filter (PushData * pdata);
  
  static void
  gst_tee_base_init (gpointer g_class)
***************
*** 282,287 ****
--- 295,304 ----
    data->pushed = FALSE;
    data->result = GST_FLOW_NOT_LINKED;
    data->removed = FALSE;
+   data->avg_rate = 0;
+   data->qos_rate = -1;
+   data->qos_count = GST_TEE_QOS_OBSCURED_DELAY_COUNT;
+   data->last_buftime = GST_CLOCK_TIME_NONE;
    g_object_set_qdata_full (G_OBJECT (srcpad), push_data, data, g_free);
  
    GST_OBJECT_UNLOCK (tee);
***************
*** 311,316 ****
--- 328,335 ----
        GST_DEBUG_FUNCPTR (gst_tee_src_check_get_range));
    gst_pad_set_getrange_function (srcpad,
        GST_DEBUG_FUNCPTR (gst_tee_src_get_range));
+   gst_pad_set_event_function (srcpad, 
+       GST_DEBUG_FUNCPTR (gst_tee_src_event));
    gst_element_add_pad (GST_ELEMENT_CAST (tee), srcpad);
  
    return srcpad;
***************
*** 652,657 ****
--- 671,706 ----
    data->result = GST_FLOW_NOT_LINKED;
  }
  
+ static gboolean gst_tee_qos_filter (PushData * pdata)
+ {
+   gint qos_rate = -1, avg_rate = 0;
+   GstClockTime current = GST_CLOCK_TIME_NONE;
+ 
+   qos_rate = g_atomic_int_get (&pdata->qos_rate);
+   if (qos_rate >= 0) {
+     /* we have to reduce rate for qos_obscured event */
+     current = gst_util_get_timestamp ();
+     if (GST_CLOCK_TIME_IS_VALID (pdata->last_buftime)) {
+       gdouble duration = GST_CLOCK_DIFF (pdata->last_buftime, current);
+       gdouble rate = (gdouble)GST_SECOND / duration;  
+       /* update current rate */
+       avg_rate = UPDATE_RUNNING_AVG (pdata->avg_rate, (gint)(rate * GST_TEE_QOS_OBSCURED_RATE_SCALE));
+     }
+   }
+   if (qos_rate < 0 || avg_rate <= qos_rate) {
+     /* the buffer should be passed through */
+     pdata->avg_rate = avg_rate;
+     pdata->last_buftime = current;
+     /* each qos_obscured event can be in effect for some time */
+     if (g_atomic_int_dec_and_test (&pdata->qos_count))       
+       g_atomic_int_set (&pdata->qos_rate, -1);
+     return FALSE;
+   }
+   
+   /* the buffer should be filtered to reduce rate */
+   return TRUE;
+ }
+ 
  static GstFlowReturn
  gst_tee_handle_data (GstTee * tee, gpointer data, gboolean is_list)
  {
***************
*** 677,686 ****
  
      if (pad == tee->pull_pad) {
        ret = GST_FLOW_OK;
-     } else if (is_list) {
-       ret = gst_pad_push_list (pad, GST_BUFFER_LIST_CAST (data));
      } else {
!       ret = gst_pad_push (pad, GST_BUFFER_CAST (data));
      }
      return ret;
    }
--- 726,744 ----
  
      if (pad == tee->pull_pad) {
        ret = GST_FLOW_OK;
      } else {
!       PushData *pdata = g_object_get_qdata ((GObject *) pad, push_data);
!       g_assert (pdata != NULL);
!       /* whether such buffer should be filtered to reduce rate for qos_obscured event */
!       if (!gst_tee_qos_filter (pdata)) {
!         if (is_list) {
!           ret = gst_pad_push_list (pad, GST_BUFFER_LIST_CAST (data));
!         } else {
!           ret = gst_pad_push (pad, GST_BUFFER_CAST (data));
!         }
!       } else {
!         ret = GST_FLOW_OK;
!       }
      }
      return ret;
    }
***************
*** 705,727 ****
      g_assert (pdata != NULL);
  
      if (G_LIKELY (!pdata->pushed)) {
        /* not yet pushed, release lock and start pushing */
        gst_object_ref (pad);
        GST_OBJECT_UNLOCK (tee);
  
        GST_LOG_OBJECT (tee, "Starting to push %s %p",
            is_list ? "list" : "buffer", data);
! 
!       ret = gst_tee_do_push (tee, pad, data, is_list);
! 
!       GST_LOG_OBJECT (tee, "Pushing item %p yielded result %s", data,
!           gst_flow_get_name (ret));
  
        GST_OBJECT_LOCK (tee);
        /* keep track of which pad we pushed and the result value. We need to do
         * this before we release the refcount on the pad, the PushData is
         * destroyed when the last ref of the pad goes away. */
!       pdata->pushed = TRUE;
        pdata->result = ret;
        gst_object_unref (pad);
      } else {
--- 763,797 ----
      g_assert (pdata != NULL);
  
      if (G_LIKELY (!pdata->pushed)) {
+       gboolean filtered;
+ 
        /* not yet pushed, release lock and start pushing */
        gst_object_ref (pad);
        GST_OBJECT_UNLOCK (tee);
  
        GST_LOG_OBJECT (tee, "Starting to push %s %p",
            is_list ? "list" : "buffer", data);
!       
!       /* whether such buffer should be filtered to reduce rate for qos_obscured event */
!       if (!gst_tee_qos_filter (pdata)) {
!         ret = gst_tee_do_push (tee, pad, data, is_list);
!       /* keep track of which pad we pushed and the result value. We need to do
!        * this before we release the refcount on the pad, the PushData is
!        * destroyed when the last ref of the pad goes away. */
!         filtered = FALSE;
!         /*GST_LOG_OBJECT (tee,*//*fprintf(stderr, "Pushing item %p yielded result %s\n", data,
!             gst_flow_get_name (ret));*/
!       } else {
!         filtered = TRUE;
!         ret = GST_FLOW_OK;
!         /*GST_LOG_OBJECT (tee, *//*fprintf(stderr, "Filtering item %p ", data);*/
!       }
  
        GST_OBJECT_LOCK (tee);
        /* keep track of which pad we pushed and the result value. We need to do
         * this before we release the refcount on the pad, the PushData is
         * destroyed when the last ref of the pad goes away. */
!       pdata->pushed = !filtered;
        pdata->result = ret;
        gst_object_unref (pad);
      } else {
***************
*** 991,993 ****
--- 1061,1101 ----
  
    return ret;
  }
+ 
+ static gboolean
+ gst_tee_src_event (GstPad * pad, GstEvent * event)
+ {
+   GstTee *tee;
+   gboolean result; 
+ 
+   tee = GST_TEE (gst_pad_get_parent_element (pad));
+   g_return_val_if_fail (tee != NULL, FALSE);
+ 
+   switch (GST_EVENT_TYPE (event)) {
+     case GST_EVENT_QOS_OBSCURED:
+     {
+       PushData * pdata;
+       gdouble rate;
+       gint qos_rate;
+       pdata = g_object_get_qdata (G_OBJECT (pad), push_data);
+       g_assert (pdata != NULL);
+       gst_event_parse_qos_obscured (event, &rate);
+       qos_rate = g_atomic_int_get (&pdata->qos_rate);
+       if (rate >= 0.0) {
+         gint target_rate = (int)(rate * GST_TEE_QOS_OBSCURED_RATE_SCALE);
+         g_atomic_int_set (&pdata->qos_count, GST_TEE_QOS_OBSCURED_DELAY_COUNT);
+         if (qos_rate != target_rate)
+           g_atomic_int_set (&pdata->qos_rate, target_rate);
+       }
+       result = GST_FLOW_OK;
+       break;
+     }
+     default: 
+       result = gst_pad_push_event (tee->sinkpad, event);
+   }
+   
+   gst_object_unref (tee);
+ 
+   return result;
+ }
+ 
