gst/playback/: Better error recovery. Added configurable preroll queue size. Faster detection of no-more-pads.

Original commit message from CVS:
* gst/playback/gstdecodebin.c: (gst_decode_bin_get_type),
(gst_decode_bin_class_init), (gst_decode_bin_is_dynamic),
(gst_decode_bin_factory_filter), (compare_ranks), (print_feature),
(gst_decode_bin_init), (gst_decode_bin_dispose),
(find_compatibles), (close_pad_link), (try_to_link_1), (new_pad),
(no_more_pads), (close_link), (type_found),
(gst_decode_bin_set_property), (gst_decode_bin_get_property),
(gst_decode_bin_change_state), (plugin_init):
* gst/playback/gstplaybasebin.c: (gst_play_base_bin_get_type),
(gst_play_base_bin_class_init), (gst_play_base_bin_init),
(gst_play_base_bin_dispose), (queue_overrun),
(gen_preroll_element), (remove_prerolls), (unknown_type),
(no_more_pads), (new_stream), (setup_source),
(gst_play_base_bin_set_property), (gst_play_base_bin_get_property),
(play_base_eos), (gst_play_base_bin_change_state),
(gst_play_base_bin_add_element),
(gst_play_base_bin_remove_element),
(gst_play_base_bin_mute_stream), (gst_play_base_bin_link_stream),
(gst_play_base_bin_unlink_stream),
(gst_play_base_bin_get_streaminfo):
* gst/playback/gstplaybasebin.h:
Better error recovery. Added configurable preroll queue size. Faster
detection of no-more-pads.

ChangeLog

@@ -1,3 +1,29 @@
+2004-07-16  Wim Taymans  <wim@fluendo.com>
+
+	* gst/playback/gstdecodebin.c: (gst_decode_bin_get_type),
+	(gst_decode_bin_class_init), (gst_decode_bin_is_dynamic),
+	(gst_decode_bin_factory_filter), (compare_ranks), (print_feature),
+	(gst_decode_bin_init), (gst_decode_bin_dispose),
+	(find_compatibles), (close_pad_link), (try_to_link_1), (new_pad),
+	(no_more_pads), (close_link), (type_found),
+	(gst_decode_bin_set_property), (gst_decode_bin_get_property),
+	(gst_decode_bin_change_state), (plugin_init):
+	* gst/playback/gstplaybasebin.c: (gst_play_base_bin_get_type),
+	(gst_play_base_bin_class_init), (gst_play_base_bin_init),
+	(gst_play_base_bin_dispose), (queue_overrun),
+	(gen_preroll_element), (remove_prerolls), (unknown_type),
+	(no_more_pads), (new_stream), (setup_source),
+	(gst_play_base_bin_set_property), (gst_play_base_bin_get_property),
+	(play_base_eos), (gst_play_base_bin_change_state),
+	(gst_play_base_bin_add_element),
+	(gst_play_base_bin_remove_element),
+	(gst_play_base_bin_mute_stream), (gst_play_base_bin_link_stream),
+	(gst_play_base_bin_unlink_stream),
+	(gst_play_base_bin_get_streaminfo):
+	* gst/playback/gstplaybasebin.h:
+	Better error recovery. Added configurable preroll queue size. Faster
+	detection of no-more-pads.
+
 2004-07-16  Wim Taymans  <wim@fluendo.com>
 
 	* gst-libs/gst/video/video.h:

gst/playback/gstdecodebin.c

@@ -56,7 +56,7 @@ struct _GstDecodeBin
   GstElement *typefind;
 
   gboolean threaded;
-  gboolean dynamic;
+  GList *dynamics;
 
   GList *factories;
   gint numpads;
@@ -71,6 +71,7 @@ struct _GstDecodeBinClass
   GstBinClass parent_class;
 
   void (*new_stream) (GstElement * element, GstPad * pad, gboolean last);
+  void (*unknown_type) (GstElement * element, GstCaps * caps);
 };
 
 /* props */
@@ -84,9 +85,19 @@ enum
 enum
 {
   SIGNAL_NEW_STREAM,
+  SIGNAL_UNKNOWN_TYPE,
   LAST_SIGNAL
 };
 
+typedef struct
+{
+  gint np_sig_id;
+  gint nmp_sig_id;
+  GstElement *element;
+  GstDecodeBin *decode_bin;
+}
+GstDynamic;
+
 static void gst_decode_bin_class_init (GstDecodeBinClass * klass);
 static void gst_decode_bin_init (GstDecodeBin * decode_bin);
 static void gst_decode_bin_dispose (GObject * object);
@@ -168,6 +179,10 @@ gst_decode_bin_class_init (GstDecodeBinClass * klass)
       G_STRUCT_OFFSET (GstDecodeBinClass, new_stream), NULL, NULL,
       gst_marshal_VOID__OBJECT_POINTER, G_TYPE_NONE, 2, G_TYPE_OBJECT,
       G_TYPE_BOOLEAN);
+  gst_decode_bin_signals[SIGNAL_UNKNOWN_TYPE] =
+      g_signal_new ("unknown-type", G_TYPE_FROM_CLASS (klass),
+      G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstDecodeBinClass, unknown_type),
+      NULL, NULL, gst_marshal_VOID__OBJECT, G_TYPE_NONE, 1, GST_TYPE_CAPS);
 
   gobject_klass->dispose = GST_DEBUG_FUNCPTR (gst_decode_bin_dispose);
 
@@ -180,6 +195,12 @@ gst_decode_bin_class_init (GstDecodeBinClass * klass)
       GST_DEBUG_FUNCPTR (gst_decode_bin_change_state);
 }
 
+static gboolean
+gst_decode_bin_is_dynamic (GstDecodeBin * decode_bin)
+{
+  return decode_bin->dynamics != NULL;
+}
+
 static gboolean
 gst_decode_bin_factory_filter (GstPluginFeature * feature,
     GstDecodeBin * decode_bin)
@@ -238,22 +259,32 @@ gst_decode_bin_init (GstDecodeBin * decode_bin)
       g_signal_connect (G_OBJECT (decode_bin->typefind), "have_type",
       G_CALLBACK (type_found), decode_bin);
 
+  decode_bin->dynamics = NULL;
 }
 
 static void
 gst_decode_bin_dispose (GObject * object)
 {
   GstDecodeBin *decode_bin;
+  GList *dyns;
 
   decode_bin = GST_DECODE_BIN (object);
 
-  g_signal_handlers_disconnect_matched (G_OBJECT (decode_bin->typefind),
+  g_signal_handler_disconnect (G_OBJECT (decode_bin->typefind),
-      G_SIGNAL_MATCH_ID, decode_bin->have_type_id, 0, NULL, NULL, NULL);
+      decode_bin->have_type_id);
 
   gst_bin_remove (GST_BIN (decode_bin), decode_bin->typefind);
 
   g_list_free (decode_bin->factories);
 
+  for (dyns = decode_bin->dynamics; dyns; dyns = g_list_next (dyns)) {
+    GstDynamic *dynamic = (GstDynamic *) dyns->data;
+
+    g_free (dynamic);
+  }
+  g_list_free (decode_bin->dynamics);
+  decode_bin->dynamics = NULL;
+
   if (G_OBJECT_CLASS (parent_class)->dispose) {
     G_OBJECT_CLASS (parent_class)->dispose (object);
   }
@@ -298,6 +329,12 @@ close_pad_link (GstElement * element, GstPad * pad, GstCaps * caps,
   GstStructure *structure;
   const gchar *mimetype;
 
+  if (gst_caps_is_empty (caps)) {
+    g_signal_emit (G_OBJECT (decode_bin),
+        gst_decode_bin_signals[SIGNAL_UNKNOWN_TYPE], 0, caps);
+    return;
+  }
+
   structure = gst_caps_get_structure (caps, 0);
   mimetype = gst_structure_get_name (structure);
 
@@ -306,6 +343,7 @@ close_pad_link (GstElement * element, GstPad * pad, GstCaps * caps,
       g_str_has_prefix (mimetype, "audio/x-raw")) {
     gchar *padname;
     GstPad *ghost;
+    gboolean dynamic;
 
     padname = g_strdup_printf ("src%d", decode_bin->numpads);
     decode_bin->numpads++;
@@ -314,10 +352,11 @@ close_pad_link (GstElement * element, GstPad * pad, GstCaps * caps,
 
     ghost = gst_element_get_pad (GST_ELEMENT (decode_bin), padname);
 
+    dynamic = gst_decode_bin_is_dynamic (decode_bin);
+
     /* our own signal with an extra flag that this is the only pad */
     g_signal_emit (G_OBJECT (decode_bin),
-        gst_decode_bin_signals[SIGNAL_NEW_STREAM], 0,
+        gst_decode_bin_signals[SIGNAL_NEW_STREAM], 0, ghost, !dynamic);
-        ghost, !decode_bin->dynamic);
 
     g_free (padname);
     return;
@@ -326,10 +365,8 @@ close_pad_link (GstElement * element, GstPad * pad, GstCaps * caps,
   /* then continue plugging */
   to_try = find_compatibles (decode_bin, caps);
   if (to_try == NULL) {
-    gchar *capsstr = gst_caps_to_string (caps);
+    g_signal_emit (G_OBJECT (decode_bin),
-
+        gst_decode_bin_signals[SIGNAL_UNKNOWN_TYPE], 0, caps);
-    g_warning ("don't know how to handle %s", capsstr);
-    g_free (capsstr);
     return;
   }
 
@@ -353,6 +390,7 @@ try_to_link_1 (GstDecodeBin * decode_bin, GstPad * pad, GList * factories)
     if (element == NULL)
       continue;
 
+    GST_DEBUG ("adding %s\n", gst_element_get_name (element));
     gst_bin_add (GST_BIN (decode_bin), element);
     decode_bin->elements = g_list_prepend (decode_bin->elements, element);
 
@@ -380,9 +418,27 @@ try_to_link_1 (GstDecodeBin * decode_bin, GstPad * pad, GList * factories)
 }
 
 static void
-new_pad (GstElement * element, GstPad * pad, GstDecodeBin * decode_bin)
+new_pad (GstElement * element, GstPad * pad, GstDynamic * dynamic)
 {
-  close_pad_link (element, pad, gst_pad_get_caps (pad), decode_bin);
+  close_pad_link (element, pad, gst_pad_get_caps (pad), dynamic->decode_bin);
+}
+
+static void
+no_more_pads (GstElement * element, GstDynamic * dynamic)
+{
+  GstDecodeBin *decode_bin = dynamic->decode_bin;
+
+  GST_DEBUG ("decodebin: no more pads\n");
+
+  g_signal_handler_disconnect (G_OBJECT (dynamic->element), dynamic->np_sig_id);
+  g_signal_handler_disconnect (G_OBJECT (dynamic->element),
+      dynamic->nmp_sig_id);
+
+  decode_bin->dynamics = g_list_remove (decode_bin->dynamics, dynamic);
+  g_free (dynamic);
+
+  if (decode_bin->dynamics == NULL)
+    gst_element_no_more_pads (GST_ELEMENT (decode_bin));
 }
 
 static void
@@ -427,9 +483,17 @@ close_link (GstElement * element, GstDecodeBin * decode_bin)
     }
   }
   if (dynamic) {
-    g_signal_connect (G_OBJECT (element), "new_pad",
+    GstDynamic *dyn;
-        G_CALLBACK (new_pad), decode_bin);
+
-    decode_bin->dynamic = TRUE;
+    dyn = g_new0 (GstDynamic, 1);
+    dyn->np_sig_id = g_signal_connect (G_OBJECT (element), "new-pad",
+        G_CALLBACK (new_pad), dyn);
+    dyn->nmp_sig_id = g_signal_connect (G_OBJECT (element), "no-more-pads",
+        G_CALLBACK (no_more_pads), dyn);
+    dyn->element = element;
+    dyn->decode_bin = decode_bin;
+
+    decode_bin->dynamics = g_list_prepend (decode_bin->dynamics, dyn);
   }
 
   for (pads = to_connect; pads; pads = g_list_next (pads)) {
@@ -444,12 +508,13 @@ static void
 type_found (GstElement * typefind, guint probability, GstCaps * caps,
     GstDecodeBin * decode_bin)
 {
-  decode_bin->dynamic = FALSE;
+  gboolean dynamic;
 
   close_pad_link (typefind, gst_element_get_pad (typefind, "src"), caps,
       decode_bin);
 
-  if (decode_bin->dynamic == FALSE) {
+  dynamic = gst_decode_bin_is_dynamic (decode_bin);
+  if (dynamic == FALSE) {
     gst_element_no_more_pads (GST_ELEMENT (decode_bin));
   }
 }
@@ -514,6 +579,7 @@ gst_decode_bin_change_state (GstElement * element)
     case GST_STATE_NULL_TO_READY:
       decode_bin->numpads = 0;
       decode_bin->threaded = FALSE;
+      decode_bin->dynamics = NULL;
       break;
     case GST_STATE_READY_TO_PAUSED:
     case GST_STATE_PAUSED_TO_PLAYING:

gst/playback/gstplaybasebin.c

@@ -27,6 +27,8 @@
 GST_DEBUG_CATEGORY_STATIC (gst_play_base_bin_debug);
 #define GST_CAT_DEFAULT gst_play_base_bin_debug
 
+#define DEFAULT_QUEUE_SIZE  (3 * GST_SECOND)
+
 /* props */
 enum
 {
@@ -34,6 +36,7 @@ enum
   ARG_URI,
   ARG_THREADED,
   ARG_NSTREAMS,
+  ARG_QUEUE_SIZE,
   ARG_STREAMINFO,
 };
 
@@ -113,6 +116,10 @@ gst_play_base_bin_class_init (GstPlayBaseBinClass * klass)
   g_object_class_install_property (gobject_klass, ARG_NSTREAMS,
       g_param_spec_int ("nstreams", "NStreams", "number of streams",
           0, G_MAXINT, 0, G_PARAM_READABLE));
+  g_object_class_install_property (gobject_klass, ARG_QUEUE_SIZE,
+      g_param_spec_uint64 ("queue-size", "Queue size",
+          "Size of internal queues in nanoseconds", 0, G_MAXINT64,
+          DEFAULT_QUEUE_SIZE, G_PARAM_READABLE));
   g_object_class_install_property (gobject_klass, ARG_STREAMINFO,
       g_param_spec_pointer ("stream-info", "Stream info", "List of streaminfo",
           G_PARAM_READABLE));
@@ -161,6 +168,7 @@ gst_play_base_bin_init (GstPlayBaseBin * play_base_bin)
   play_base_bin->preroll_lock = g_mutex_new ();
   play_base_bin->preroll_cond = g_cond_new ();
   play_base_bin->preroll_elems = NULL;
+  play_base_bin->queue_size = DEFAULT_QUEUE_SIZE;
 
   GST_FLAG_SET (play_base_bin, GST_BIN_SELF_SCHEDULABLE);
 }
@@ -184,8 +192,6 @@ queue_overrun (GstElement * element, GstPlayBaseBin * play_base_bin)
   g_mutex_lock (play_base_bin->preroll_lock);
   g_cond_signal (play_base_bin->preroll_cond);
   g_mutex_unlock (play_base_bin->preroll_lock);
-  //g_signal_handlers_disconnect_by_func(G_OBJECT (element),
-//                G_CALLBACK (queue_overrun), play_base_bin);
 }
 
 static GstElement *
@@ -196,6 +202,9 @@ gen_preroll_element (GstPlayBaseBin * play_base_bin, GstPad * pad)
 
   name = g_strdup_printf ("preroll_%s", gst_pad_get_name (pad));
   element = gst_element_factory_make ("queue", name);
+  g_object_set (G_OBJECT (element), "max-size-buffers", 0, NULL);
+  g_object_set (G_OBJECT (element), "max-size-bytes", 0, NULL);
+  g_object_set (G_OBJECT (element), "max-size-time", 3 * GST_SECOND, NULL);
   g_signal_connect (G_OBJECT (element), "overrun",
       G_CALLBACK (queue_overrun), play_base_bin);
   g_free (name);
@@ -228,9 +237,20 @@ remove_prerolls (GstPlayBaseBin * play_base_bin)
   play_base_bin->nstreams = 0;
 }
 
+static void
+unknown_type (GstElement * element, GstCaps * caps,
+    GstPlayBaseBin * play_base_bin)
+{
+  gchar *capsstr = gst_caps_to_string (caps);
+
+  g_warning ("don't know how to handle %s", capsstr);
+  g_free (capsstr);
+}
+
 static void
 no_more_pads (GstElement * element, GstPlayBaseBin * play_base_bin)
 {
+  GST_DEBUG ("no more pads\n");
   g_mutex_lock (play_base_bin->preroll_lock);
   g_cond_signal (play_base_bin->preroll_cond);
   g_mutex_unlock (play_base_bin->preroll_lock);
@@ -248,8 +268,15 @@ new_stream (GstElement * element, GstPad * pad, gboolean last,
   GstStreamType type;
   GstPad *srcpad;
 
+  GST_DEBUG ("play base: new stream\n");
+
   caps = gst_pad_get_caps (pad);
 
+  if (gst_caps_is_empty (caps)) {
+    g_warning ("no type on pad %s:%s\n", GST_DEBUG_PAD_NAME (pad));
+    return;
+  }
+
   structure = gst_caps_get_structure (caps, 0);
   mimetype = gst_structure_get_name (structure);
 
@@ -310,7 +337,7 @@ setup_source (GstPlayBaseBin * play_base_bin)
 
   {
     gboolean res;
-    gint sig1, sig2;
+    gint sig1, sig2, sig3;
     GstElement *old_dec;
 
     old_dec = play_base_bin->decoder;
@@ -342,16 +369,20 @@ setup_source (GstPlayBaseBin * play_base_bin)
         G_CALLBACK (new_stream), play_base_bin);
     sig2 = g_signal_connect (G_OBJECT (play_base_bin->decoder), "no-more-pads",
         G_CALLBACK (no_more_pads), play_base_bin);
+    sig3 = g_signal_connect (G_OBJECT (play_base_bin->decoder), "unknown-type",
+        G_CALLBACK (unknown_type), play_base_bin);
 
+    /* either when the queues are filled or when the decoder element has no more
+     * dynamic streams, the cond is unlocked. We can remove the signal handlers then
+     */
     g_mutex_lock (play_base_bin->preroll_lock);
     gst_element_set_state (play_base_bin->thread, GST_STATE_PLAYING);
     g_cond_wait (play_base_bin->preroll_cond, play_base_bin->preroll_lock);
     g_mutex_unlock (play_base_bin->preroll_lock);
 
-    g_signal_handlers_disconnect_matched (G_OBJECT (play_base_bin->decoder),
+    g_signal_handler_disconnect (G_OBJECT (play_base_bin->decoder), sig3);
-        G_SIGNAL_MATCH_ID, sig1, 0, NULL, NULL, NULL);
+    g_signal_handler_disconnect (G_OBJECT (play_base_bin->decoder), sig2);
-    g_signal_handlers_disconnect_matched (G_OBJECT (play_base_bin->decoder),
+    g_signal_handler_disconnect (G_OBJECT (play_base_bin->decoder), sig1);
-        G_SIGNAL_MATCH_ID, sig2, 0, NULL, NULL, NULL);
 
     play_base_bin->need_rebuild = FALSE;
   }
@@ -386,6 +417,9 @@ gst_play_base_bin_set_property (GObject * object, guint prop_id,
       }
       break;
     }
+    case ARG_QUEUE_SIZE:
+      play_base_bin->queue_size = g_value_get_uint64 (value);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -409,6 +443,9 @@ gst_play_base_bin_get_property (GObject * object, guint prop_id, GValue * value,
     case ARG_NSTREAMS:
       g_value_set_int (value, play_base_bin->nstreams);
       break;
+    case ARG_QUEUE_SIZE:
+      g_value_set_uint64 (value, play_base_bin->queue_size);
+      break;
     case ARG_STREAMINFO:
       g_value_set_pointer (value, play_base_bin->streaminfo);
       break;
@@ -421,7 +458,8 @@ gst_play_base_bin_get_property (GObject * object, guint prop_id, GValue * value,
 static void
 play_base_eos (GstBin * bin, GstPlayBaseBin * play_base_bin)
 {
-  g_print ("eos\n");
+  no_more_pads (GST_ELEMENT (bin), play_base_bin);
+
   gst_element_set_eos (GST_ELEMENT (play_base_bin));
 }
 
@@ -561,7 +599,7 @@ void
 gst_play_base_bin_mute_stream (GstPlayBaseBin * play_base_bin,
     GstStreamInfo * info, gboolean mute)
 {
-  g_print ("mute\n");
+  GST_DEBUG ("mute\n");
 }
 
 void
@@ -586,10 +624,11 @@ gst_play_base_bin_link_stream (GstPlayBaseBin * play_base_bin,
   }
   if (info) {
     if (!gst_pad_link (info->pad, pad)) {
-      g_print ("could not link\n");
+      GST_DEBUG ("could not link\n");
+      gst_play_base_bin_mute_stream (play_base_bin, info, TRUE);
     }
   } else {
-    g_print ("could not find pad to link\n");
+    GST_DEBUG ("could not find pad to link\n");
   }
 }
 
@@ -597,7 +636,7 @@ void
 gst_play_base_bin_unlink_stream (GstPlayBaseBin * play_base_bin,
     GstStreamInfo * info)
 {
-  g_print ("unlink\n");
+  GST_DEBUG ("unlink\n");
 }
 
 const GList *

gst/playback/gstplaybasebin.h

@@ -43,6 +43,7 @@ struct _GstPlayBaseBin {
   GMutex	*preroll_lock;
   GCond		*preroll_cond;
   GList		*preroll_elems;
+  guint64	 queue_size;
 
   /* internal thread */
   GstElement	*thread;