Added prototypes for gst_pad_peek, gst_pad_select and gst_pad_selectv.

Original commit message from CVS:
Added prototypes for gst_pad_peek, gst_pad_select and gst_pad_selectv.
Added different scheduling methods in aggregator.
Fixed a bug in fakesrc when setting up the scheduling functions.

gst/elements/gstaggregator.c

@@ -42,6 +42,7 @@ enum {
   ARG_0,
   ARG_NUM_PADS,
   ARG_SILENT,
+  ARG_SCHED,
   /* FILL ME */
 };
 
@@ -52,6 +53,26 @@ GST_PADTEMPLATE_FACTORY (aggregator_src_factory,
   NULL			/* no caps */
 );
 
+#define GST_TYPE_AGGREGATOR_SCHED (gst_aggregator_sched_get_type())
+static GType
+gst_aggregator_sched_get_type (void)
+{
+  static GType aggregator_sched_type = 0;
+  static GEnumValue aggregator_sched[] = {
+    { AGGREGATOR_LOOP,   	"1", "Loop Based"},
+    { AGGREGATOR_LOOP_PEEK,    	"2", "Loop Based Peek"},
+    { AGGREGATOR_LOOP_SELECT,   "3", "Loop Based Select"},
+    { AGGREGATOR_CHAIN,      	"4", "Chain Based"},
+    {0, NULL, NULL},
+  };
+  if (!aggregator_sched_type) {
+    aggregator_sched_type = g_enum_register_static ("GstAggregatorSched", aggregator_sched);
+  }
+  return aggregator_sched_type;
+}
+
+#define AGGREGATOR_IS_LOOP_BASED(ag)	((ag)->sched != AGGREGATOR_CHAIN)
+
 static void 	gst_aggregator_class_init	(GstAggregatorClass *klass);
 static void 	gst_aggregator_init		(GstAggregator *aggregator);
 
@@ -63,6 +84,7 @@ static void 	gst_aggregator_get_property 	(GObject *object, guint prop_id,
 						 GValue *value, GParamSpec *pspec);
 
 static void  	gst_aggregator_chain 		(GstPad *pad, GstBuffer *buf);
+static void 	gst_aggregator_loop 		(GstElement *element);
 
 static GstElementClass *parent_class = NULL;
 //static guint gst_aggregator_signals[LAST_SIGNAL] = { 0 };
@@ -106,6 +128,9 @@ gst_aggregator_class_init (GstAggregatorClass *klass)
   g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_SILENT,
     g_param_spec_boolean ("silent", "silent", "silent",
                       FALSE, G_PARAM_READWRITE)); 
+  g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_SCHED,
+    g_param_spec_enum ("sched", "sched", "sched",
+                      GST_TYPE_AGGREGATOR_SCHED, AGGREGATOR_CHAIN, G_PARAM_READWRITE)); 
 
   gobject_class->set_property = GST_DEBUG_FUNCPTR(gst_aggregator_set_property);
   gobject_class->get_property = GST_DEBUG_FUNCPTR(gst_aggregator_get_property);
@@ -146,12 +171,38 @@ gst_aggregator_request_new_pad (GstElement *element, GstPadTemplate *templ)
   gst_pad_set_chain_function (sinkpad, gst_aggregator_chain);
   gst_element_add_pad (GST_ELEMENT (aggregator), sinkpad);
   
-  aggregator->sinkpads = g_slist_prepend (aggregator->sinkpads, sinkpad);
+  aggregator->sinkpads = g_list_prepend (aggregator->sinkpads, sinkpad);
   aggregator->numsinkpads++;
   
   return sinkpad;
 }
 
+static void
+gst_aggregator_update_functions (GstAggregator *aggregator)
+{
+  GList *pads;
+
+  if (AGGREGATOR_IS_LOOP_BASED (aggregator)) {
+    gst_element_set_loop_function (GST_ELEMENT (aggregator), GST_DEBUG_FUNCPTR (gst_aggregator_loop));
+  }
+  else {
+    gst_element_set_loop_function (GST_ELEMENT (aggregator), NULL);
+  }
+
+  pads = aggregator->sinkpads;
+  while (pads) {
+    GstPad *pad = GST_PAD (pads->data);
+		          
+    if (AGGREGATOR_IS_LOOP_BASED (aggregator)) {
+      gst_pad_set_get_function (pad, NULL);
+    }
+    else {
+      gst_element_set_loop_function (GST_ELEMENT (aggregator), NULL);
+    }
+    pads = g_list_next (pads);
+  }
+}
+
 static void
 gst_aggregator_set_property (GObject *object, guint prop_id, const GValue *value, GParamSpec *pspec)
 {
@@ -166,6 +217,10 @@ gst_aggregator_set_property (GObject *object, guint prop_id, const GValue *value
     case ARG_SILENT:
       aggregator->silent = g_value_get_boolean (value);
       break;
+    case ARG_SCHED:
+      aggregator->sched = g_value_get_enum (value);
+      gst_aggregator_update_functions (aggregator);
+      break;
     default:
       break;
   }
@@ -188,11 +243,75 @@ gst_aggregator_get_property (GObject *object, guint prop_id, GValue *value, GPar
     case ARG_SILENT:
       g_value_set_boolean (value, aggregator->silent);
       break;
+    case ARG_SCHED:
+      g_value_set_enum (value, aggregator->sched);
+      break;
     default:
       break;
   }
 }
 
+static void 
+gst_aggregator_push (GstAggregator *aggregator, GstPad *pad, GstBuffer *buf, guchar *debug) 
+{
+  if (!aggregator->silent)
+    g_print("aggregator: %10.10s ******* (%s:%s)a (%d bytes, %llu) \n",
+            debug, GST_DEBUG_PAD_NAME (pad), GST_BUFFER_SIZE (buf), GST_BUFFER_TIMESTAMP (buf));
+
+  gst_pad_push (aggregator->srcpad, buf);
+}
+
+static void 
+gst_aggregator_loop (GstElement *element) 
+{
+  GstAggregator *aggregator;
+  GstBuffer *buf;
+  guchar *debug;
+
+  aggregator = GST_AGGREGATOR (element);
+
+  do {
+    if (aggregator->sched == AGGREGATOR_LOOP ||
+        aggregator->sched == AGGREGATOR_LOOP_PEEK) {
+      GList *pads = aggregator->sinkpads;
+
+      while (pads) {
+	GstPad *pad = GST_PAD (pads->data);
+	pads = g_list_next (pads);
+
+        if (aggregator->sched == AGGREGATOR_LOOP_PEEK) {
+	  buf = gst_pad_peek (pad);
+	  if (buf == NULL)
+	    continue;
+
+	  g_assert (buf == gst_pad_pull (pad));
+	  debug = "loop_peek";
+        }
+	else {
+	  buf = gst_pad_pull (pad);
+	  debug = "loop";
+	}
+	gst_aggregator_push (aggregator, pad, buf, debug);
+      }
+    }
+    else {
+      if (aggregator->sched == AGGREGATOR_LOOP_SELECT) {
+	GstPad *pad;
+
+	debug = "loop_select";
+
+	pad = gst_pad_select (aggregator->sinkpads);
+	buf = gst_pad_pull (pad);
+
+	gst_aggregator_push (aggregator, pad, buf, debug);
+      }
+      else {
+	g_assert_not_reached ();
+      }
+    }
+  } while (!GST_ELEMENT_IS_COTHREAD_STOPPING (element));
+}
+
 /**
  * gst_aggregator_chain:
  * @pad: the pad to follow
@@ -212,11 +331,7 @@ gst_aggregator_chain (GstPad *pad, GstBuffer *buf)
   aggregator = GST_AGGREGATOR (gst_pad_get_parent (pad));
   gst_trace_add_entry (NULL, 0, buf, "aggregator buffer");
 
-  if (!aggregator->silent)
+  gst_aggregator_push (aggregator, pad, buf, "chain");
-    g_print("aggregator: chain ******* (%s:%s)a (%d bytes, %llu) \n",
-               GST_DEBUG_PAD_NAME (pad), GST_BUFFER_SIZE (buf), GST_BUFFER_TIMESTAMP (buf));
-
-  gst_pad_push (aggregator->srcpad, buf);
 }
 
 gboolean

gst/elements/gstaggregator.h

@@ -33,6 +33,13 @@ extern "C" {
 
 extern GstElementDetails gst_aggregator_details;
 
+typedef enum {
+  AGGREGATOR_LOOP 		= 1,
+  AGGREGATOR_LOOP_PEEK,
+  AGGREGATOR_LOOP_SELECT,
+  AGGREGATOR_CHAIN,
+} GstAggregatorSchedType;
+
 #define GST_TYPE_AGGREGATOR \
   (gst_aggregator_get_type())
 #define GST_AGGREGATOR(obj) \
@@ -53,8 +60,10 @@ struct _GstAggregator {
   GstPad *srcpad;
 
   gboolean silent;
+  GstAggregatorSchedType sched;
+
   gint numsinkpads;
-  GSList *sinkpads;
+  GList *sinkpads;
 };
 
 struct _GstAggregatorClass {

gst/elements/gstfakesrc.c

@@ -221,17 +221,22 @@ gst_fakesrc_update_functions (GstFakeSrc *src)
 {
   GSList *pads;
 
+  if (src->loop_based) {
+    gst_element_set_loop_function (GST_ELEMENT (src), GST_DEBUG_FUNCPTR (gst_fakesrc_loop));
+  }
+  else {
+    gst_element_set_loop_function (GST_ELEMENT (src), NULL);
+  }
+
   pads = src->srcpads;
   while (pads) {
     GstPad *pad = GST_PAD (pads->data);
 
     if (src->loop_based) {
-      gst_element_set_loop_function (GST_ELEMENT (src), gst_fakesrc_loop);
       gst_pad_set_get_function (pad, NULL);
     }
     else {
-      gst_pad_set_get_function (pad, gst_fakesrc_get);
+      gst_pad_set_get_function (pad, GST_DEBUG_FUNCPTR (gst_fakesrc_get));
-      gst_element_set_loop_function (GST_ELEMENT (src), NULL);
     }
     pads = g_slist_next (pads);
   }

gst/gstpad.c

@@ -1566,6 +1566,72 @@ gst_pad_pullregion (GstPad *pad, GstRegionType type, guint64 offset, guint64 len
 }
 #endif
 
+/**
+ * gst_pad_peek:
+ * @pad: the pad to peek
+ *
+ * Peek for a buffer from the peer pad.
+ *
+ * Returns: a from the peer pad or NULL if the peer has no buffer.
+ */
+GstBuffer*
+gst_pad_peek (GstPad *pad)
+{
+  g_return_val_if_fail (GST_PAD_DIRECTION (pad) == GST_PAD_SINK, NULL);
+
+  return GST_RPAD_BUFPEN (GST_RPAD_PEER (pad));
+}
+
+/**
+ * gst_pad_select:
+ * @padlist: A list of pads 
+ *
+ * Wait for a buffer on the list of pads.
+ *
+ * Returns: The pad that has a buffer available, use 
+ * #gst_pad_pull to get the buffer.
+ */
+GstPad*
+gst_pad_select (GList *padlist)
+{
+  // FIXME implement me
+  return NULL;
+}
+
+/**
+ * gst_pad_selectv:
+ * @pad: The first pad to perform the select on 
+ * @...: More pads
+ *
+ * Wait for a buffer on the given of pads.
+ *
+ * Returns: The pad that has a buffer available, use 
+ * #gst_pad_pull to get the buffer.
+ */
+GstPad*
+gst_pad_selectv (GstPad *pad, ...)
+{
+  GstPad *result;
+  GList *padlist = NULL;
+  va_list var_args;
+
+  if (pad == NULL)
+    return NULL;
+
+  va_start (var_args, pad);
+
+  while (pad) {
+    padlist = g_list_prepend (padlist, pad);
+    pad = va_arg (var_args, GstPad *);
+  }
+  result = gst_pad_select (padlist);
+  g_list_free (padlist);
+
+  va_end (var_args);
+  
+  return result;
+}
+
 /************************************************************************
  *
  * templates

gst/gstpad.h

@@ -287,7 +287,6 @@ name (void)                                     \
 
 #define GST_PADTEMPLATE_GET(fact) (fact)()
 
-
 GType			gst_pad_get_type		(void);
 GType			gst_real_pad_get_type		(void);
 GType			gst_ghost_pad_get_type		(void);
@@ -362,6 +361,9 @@ NULL )
 (((GstRealPad *)(pad))->peer->pullregionfunc)((GstPad *)(((GstRealPad *)(pad))->peer),(type),(offset),(len)) : \
 NULL )
 #endif
+GstBuffer*		gst_pad_peek			(GstPad *pad);
+GstPad*			gst_pad_select			(GList *padlist);
+GstPad*			gst_pad_selectv			(GstPad *pad, ...);
 
 #define			gst_pad_eos(pad)		(GST_RPAD_EOSFUNC(GST_RPAD_PEER(pad))(GST_PAD(GST_RPAD_PEER(pad))))
 gboolean		gst_pad_set_eos			(GstPad *pad);
@@ -373,8 +375,7 @@ void			gst_pad_load_and_connect	(xmlNodePtr self, GstObject *parent);
 
 
 /* ghostpads */
-GstPad *		gst_ghost_pad_new		(gchar *name,GstPad *pad);
+GstPad*			gst_ghost_pad_new		(gchar *name,GstPad *pad);
-
 
 
 /* templates and factories */

gst/gstscheduler.h

@@ -70,6 +70,7 @@ struct _GstSchedule {
   void (*unlock_element)	(GstSchedule *sched, GstElement *element);
   void (*pad_connect)		(GstSchedule *sched, GstPad *srcpad, GstPad *sinkpad);
   void (*pad_disconnect)	(GstSchedule *sched, GstPad *srcpad, GstPad *sinkpad);
+  void (*pad_select)		(GstSchedule *sched, GList *padlist);
   gboolean (*iterate)		(GstSchedule *sched);
 };
 

plugins/elements/gstaggregator.c

@@ -42,6 +42,7 @@ enum {
   ARG_0,
   ARG_NUM_PADS,
   ARG_SILENT,
+  ARG_SCHED,
   /* FILL ME */
 };
 
@@ -52,6 +53,26 @@ GST_PADTEMPLATE_FACTORY (aggregator_src_factory,
   NULL			/* no caps */
 );
 
+#define GST_TYPE_AGGREGATOR_SCHED (gst_aggregator_sched_get_type())
+static GType
+gst_aggregator_sched_get_type (void)
+{
+  static GType aggregator_sched_type = 0;
+  static GEnumValue aggregator_sched[] = {
+    { AGGREGATOR_LOOP,   	"1", "Loop Based"},
+    { AGGREGATOR_LOOP_PEEK,    	"2", "Loop Based Peek"},
+    { AGGREGATOR_LOOP_SELECT,   "3", "Loop Based Select"},
+    { AGGREGATOR_CHAIN,      	"4", "Chain Based"},
+    {0, NULL, NULL},
+  };
+  if (!aggregator_sched_type) {
+    aggregator_sched_type = g_enum_register_static ("GstAggregatorSched", aggregator_sched);
+  }
+  return aggregator_sched_type;
+}
+
+#define AGGREGATOR_IS_LOOP_BASED(ag)	((ag)->sched != AGGREGATOR_CHAIN)
+
 static void 	gst_aggregator_class_init	(GstAggregatorClass *klass);
 static void 	gst_aggregator_init		(GstAggregator *aggregator);
 
@@ -63,6 +84,7 @@ static void 	gst_aggregator_get_property 	(GObject *object, guint prop_id,
 						 GValue *value, GParamSpec *pspec);
 
 static void  	gst_aggregator_chain 		(GstPad *pad, GstBuffer *buf);
+static void 	gst_aggregator_loop 		(GstElement *element);
 
 static GstElementClass *parent_class = NULL;
 //static guint gst_aggregator_signals[LAST_SIGNAL] = { 0 };
@@ -106,6 +128,9 @@ gst_aggregator_class_init (GstAggregatorClass *klass)
   g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_SILENT,
     g_param_spec_boolean ("silent", "silent", "silent",
                       FALSE, G_PARAM_READWRITE)); 
+  g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_SCHED,
+    g_param_spec_enum ("sched", "sched", "sched",
+                      GST_TYPE_AGGREGATOR_SCHED, AGGREGATOR_CHAIN, G_PARAM_READWRITE)); 
 
   gobject_class->set_property = GST_DEBUG_FUNCPTR(gst_aggregator_set_property);
   gobject_class->get_property = GST_DEBUG_FUNCPTR(gst_aggregator_get_property);
@@ -146,12 +171,38 @@ gst_aggregator_request_new_pad (GstElement *element, GstPadTemplate *templ)
   gst_pad_set_chain_function (sinkpad, gst_aggregator_chain);
   gst_element_add_pad (GST_ELEMENT (aggregator), sinkpad);
   
-  aggregator->sinkpads = g_slist_prepend (aggregator->sinkpads, sinkpad);
+  aggregator->sinkpads = g_list_prepend (aggregator->sinkpads, sinkpad);
   aggregator->numsinkpads++;
   
   return sinkpad;
 }
 
+static void
+gst_aggregator_update_functions (GstAggregator *aggregator)
+{
+  GList *pads;
+
+  if (AGGREGATOR_IS_LOOP_BASED (aggregator)) {
+    gst_element_set_loop_function (GST_ELEMENT (aggregator), GST_DEBUG_FUNCPTR (gst_aggregator_loop));
+  }
+  else {
+    gst_element_set_loop_function (GST_ELEMENT (aggregator), NULL);
+  }
+
+  pads = aggregator->sinkpads;
+  while (pads) {
+    GstPad *pad = GST_PAD (pads->data);
+		          
+    if (AGGREGATOR_IS_LOOP_BASED (aggregator)) {
+      gst_pad_set_get_function (pad, NULL);
+    }
+    else {
+      gst_element_set_loop_function (GST_ELEMENT (aggregator), NULL);
+    }
+    pads = g_list_next (pads);
+  }
+}
+
 static void
 gst_aggregator_set_property (GObject *object, guint prop_id, const GValue *value, GParamSpec *pspec)
 {
@@ -166,6 +217,10 @@ gst_aggregator_set_property (GObject *object, guint prop_id, const GValue *value
     case ARG_SILENT:
       aggregator->silent = g_value_get_boolean (value);
       break;
+    case ARG_SCHED:
+      aggregator->sched = g_value_get_enum (value);
+      gst_aggregator_update_functions (aggregator);
+      break;
     default:
       break;
   }
@@ -188,11 +243,75 @@ gst_aggregator_get_property (GObject *object, guint prop_id, GValue *value, GPar
     case ARG_SILENT:
       g_value_set_boolean (value, aggregator->silent);
       break;
+    case ARG_SCHED:
+      g_value_set_enum (value, aggregator->sched);
+      break;
     default:
       break;
   }
 }
 
+static void 
+gst_aggregator_push (GstAggregator *aggregator, GstPad *pad, GstBuffer *buf, guchar *debug) 
+{
+  if (!aggregator->silent)
+    g_print("aggregator: %10.10s ******* (%s:%s)a (%d bytes, %llu) \n",
+            debug, GST_DEBUG_PAD_NAME (pad), GST_BUFFER_SIZE (buf), GST_BUFFER_TIMESTAMP (buf));
+
+  gst_pad_push (aggregator->srcpad, buf);
+}
+
+static void 
+gst_aggregator_loop (GstElement *element) 
+{
+  GstAggregator *aggregator;
+  GstBuffer *buf;
+  guchar *debug;
+
+  aggregator = GST_AGGREGATOR (element);
+
+  do {
+    if (aggregator->sched == AGGREGATOR_LOOP ||
+        aggregator->sched == AGGREGATOR_LOOP_PEEK) {
+      GList *pads = aggregator->sinkpads;
+
+      while (pads) {
+	GstPad *pad = GST_PAD (pads->data);
+	pads = g_list_next (pads);
+
+        if (aggregator->sched == AGGREGATOR_LOOP_PEEK) {
+	  buf = gst_pad_peek (pad);
+	  if (buf == NULL)
+	    continue;
+
+	  g_assert (buf == gst_pad_pull (pad));
+	  debug = "loop_peek";
+        }
+	else {
+	  buf = gst_pad_pull (pad);
+	  debug = "loop";
+	}
+	gst_aggregator_push (aggregator, pad, buf, debug);
+      }
+    }
+    else {
+      if (aggregator->sched == AGGREGATOR_LOOP_SELECT) {
+	GstPad *pad;
+
+	debug = "loop_select";
+
+	pad = gst_pad_select (aggregator->sinkpads);
+	buf = gst_pad_pull (pad);
+
+	gst_aggregator_push (aggregator, pad, buf, debug);
+      }
+      else {
+	g_assert_not_reached ();
+      }
+    }
+  } while (!GST_ELEMENT_IS_COTHREAD_STOPPING (element));
+}
+
 /**
  * gst_aggregator_chain:
  * @pad: the pad to follow
@@ -212,11 +331,7 @@ gst_aggregator_chain (GstPad *pad, GstBuffer *buf)
   aggregator = GST_AGGREGATOR (gst_pad_get_parent (pad));
   gst_trace_add_entry (NULL, 0, buf, "aggregator buffer");
 
-  if (!aggregator->silent)
+  gst_aggregator_push (aggregator, pad, buf, "chain");
-    g_print("aggregator: chain ******* (%s:%s)a (%d bytes, %llu) \n",
-               GST_DEBUG_PAD_NAME (pad), GST_BUFFER_SIZE (buf), GST_BUFFER_TIMESTAMP (buf));
-
-  gst_pad_push (aggregator->srcpad, buf);
 }
 
 gboolean

plugins/elements/gstaggregator.h

@@ -33,6 +33,13 @@ extern "C" {
 
 extern GstElementDetails gst_aggregator_details;
 
+typedef enum {
+  AGGREGATOR_LOOP 		= 1,
+  AGGREGATOR_LOOP_PEEK,
+  AGGREGATOR_LOOP_SELECT,
+  AGGREGATOR_CHAIN,
+} GstAggregatorSchedType;
+
 #define GST_TYPE_AGGREGATOR \
   (gst_aggregator_get_type())
 #define GST_AGGREGATOR(obj) \
@@ -53,8 +60,10 @@ struct _GstAggregator {
   GstPad *srcpad;
 
   gboolean silent;
+  GstAggregatorSchedType sched;
+
   gint numsinkpads;
-  GSList *sinkpads;
+  GList *sinkpads;
 };
 
 struct _GstAggregatorClass {

plugins/elements/gstfakesrc.c

@@ -221,17 +221,22 @@ gst_fakesrc_update_functions (GstFakeSrc *src)
 {
   GSList *pads;
 
+  if (src->loop_based) {
+    gst_element_set_loop_function (GST_ELEMENT (src), GST_DEBUG_FUNCPTR (gst_fakesrc_loop));
+  }
+  else {
+    gst_element_set_loop_function (GST_ELEMENT (src), NULL);
+  }
+
   pads = src->srcpads;
   while (pads) {
     GstPad *pad = GST_PAD (pads->data);
 
     if (src->loop_based) {
-      gst_element_set_loop_function (GST_ELEMENT (src), gst_fakesrc_loop);
       gst_pad_set_get_function (pad, NULL);
     }
     else {
-      gst_pad_set_get_function (pad, gst_fakesrc_get);
+      gst_pad_set_get_function (pad, GST_DEBUG_FUNCPTR (gst_fakesrc_get));
-      gst_element_set_loop_function (GST_ELEMENT (src), NULL);
     }
     pads = g_slist_next (pads);
   }