gst/adder/gstadder.c: Fix adder seeking.

Original commit message from CVS:
* gst/adder/gstadder.c: (gst_adder_query_duration),
(forward_event_func), (forward_event), (gst_adder_src_event):
Fix adder seeking.
Make query/seeking code threadsafe.
* tests/check/Makefile.am:
* tests/check/elements/adder.c: (test_event_message_received),
(GST_START_TEST), (test_play_twice_message_received):
Fix adder test case.

ChangeLog

@@ -1,3 +1,15 @@
+2006-05-29  Wim Taymans  <wim@fluendo.com>
+
+	* gst/adder/gstadder.c: (gst_adder_query_duration),
+	(forward_event_func), (forward_event), (gst_adder_src_event):
+	Fix adder seeking.
+	Make query/seeking code threadsafe.
+
+	* tests/check/Makefile.am:
+	* tests/check/elements/adder.c: (test_event_message_received),
+	(GST_START_TEST), (test_play_twice_message_received):
+	Fix adder test case.
+
 2006-05-29  Tim-Philipp Müller  <tim at centricular dot net>
 
 	Patch by: Young-Ho Cha  <ganadist at chollian net>

gst/adder/gstadder.c

@@ -245,27 +245,38 @@ not_supported:
  * eachother which we can get from the first timestamps we see.
  * When we add a new stream (or remove a stream) the duration might
  * also become invalid again and we need to post a new DURATION
- * message to ntify this fact to the parent.
+ * message to notify this fact to the parent.
  * For now we take the max of all the upstream elements so the simple
  * cases work at least somewhat. */
 static gboolean
 gst_adder_query_duration (GstAdder * adder, GstQuery * query)
 {
-  GList *pads;
   gint64 max;
   gboolean res;
   GstFormat format;
-
+  GstIterator *it;
-  max = -1;
+  gboolean done;
-  res = TRUE;
 
   /* parse format */
   gst_query_parse_duration (query, &format, NULL);
 
-  GST_OBJECT_LOCK (adder);
+  max = -1;
-  pads = GST_ELEMENT_CAST (adder)->sinkpads;
+  res = TRUE;
-  for (; pads; pads = g_list_next (pads)) {
+  done = FALSE;
-    GstPad *pad = GST_PAD_CAST (pads->data);
+
+  it = gst_element_iterate_sink_pads (GST_ELEMENT_CAST (adder));
+  while (!done) {
+    GstIteratorResult ires;
+    gpointer item;
+
+    ires = gst_iterator_next (it, &item);
+    switch (ires) {
+      case GST_ITERATOR_DONE:
+        done = TRUE;
+        break;
+      case GST_ITERATOR_OK:
+      {
+        GstPad *pad = GST_PAD_CAST (item);
         gint64 duration;
 
         /* ask sink peer for duration */
@@ -275,17 +286,29 @@ gst_adder_query_duration (GstAdder * adder, GstQuery * query)
           /* valid unknown length, stop searching */
           if (duration == -1) {
             max = duration;
-        break;
+            done = TRUE;
           }
           /* else see if bigger than current max */
           else if (duration > max)
             max = duration;
         }
+        break;
+      }
+      case GST_ITERATOR_RESYNC:
+        max = -1;
+        res = TRUE;
+        break;
+      default:
+        res = FALSE;
+        done = TRUE;
+        break;
+    }
   }
-  GST_OBJECT_UNLOCK (adder);
 
+  if (res) {
     /* and store the max */
     gst_query_set_duration (query, format, max);
+  }
 
   return res;
 }
@@ -332,6 +355,22 @@ gst_adder_query (GstPad * pad, GstQuery * query)
   return res;
 }
 
+static gboolean
+forward_event_func (GstPad * pad, GValue * ret, GstEvent * event)
+{
+  gst_event_ref (event);
+  if (!gst_pad_push_event (pad, event)) {
+    g_value_set_boolean (ret, FALSE);
+    GST_WARNING_OBJECT (pad, "Sending event  %p (%s) failed.",
+        event, GST_EVENT_TYPE_NAME (event));
+  } else {
+    GST_LOG_OBJECT (pad, "Sent event  %p (%s).",
+        event, GST_EVENT_TYPE_NAME (event));
+  }
+  gst_object_unref (pad);
+  return TRUE;
+}
+
 /* forwards the event to all sinkpads, takes ownership of the 
  * event
  *
@@ -342,34 +381,24 @@ static gboolean
 forward_event (GstAdder * adder, GstEvent * event)
 {
   gboolean ret;
-  GList *pads;
+  GstIterator *it;
+  GValue vret = { 0 };
 
   GST_LOG_OBJECT (adder, "Forwarding event %p (%s)", event,
       GST_EVENT_TYPE_NAME (event));
 
   ret = TRUE;
 
-  GST_OBJECT_LOCK (adder);
+  g_value_init (&vret, G_TYPE_BOOLEAN);
-  pads = GST_ELEMENT_CAST (adder)->sinkpads;
+  g_value_set_boolean (&vret, TRUE);
-  for (; pads; pads = g_list_next (pads)) {
+  it = gst_element_iterate_sink_pads (GST_ELEMENT_CAST (adder));
-    GstPad *pad = GST_PAD_CAST (pads->data);
+  gst_iterator_fold (it, (GstIteratorFoldFunction) forward_event_func, &vret,
-
+      event);
-    gst_event_ref (event);
+  gst_iterator_free (it);
-    ret &= gst_pad_push_event (pad, event);
-
-    if (!ret) {
-      GST_WARNING_OBJECT (pad, "Sending event  %p (%s) failed.",
-          event, GST_EVENT_TYPE_NAME (event));
-      break;
-    } else {
-      GST_LOG_OBJECT (pad, "Sent event  %p (%s).",
-          event, GST_EVENT_TYPE_NAME (event));
-    }
-  }
-  GST_OBJECT_UNLOCK (adder);
-
   gst_event_unref (event);
 
+  ret = g_value_get_boolean (&vret);
+
   return ret;
 }
 
@@ -387,9 +416,33 @@ gst_adder_src_event (GstPad * pad, GstEvent * event)
       result = FALSE;
       break;
     case GST_EVENT_SEEK:
-      /* FIXME seek needs something smarter. */
+    {
+      GstSeekFlags flags;
+
+      /* parse the flushing flag */
+      gst_event_parse_seek (event, NULL, NULL, &flags, NULL, NULL, NULL, NULL);
+
+      /* if we are not flushing, just forward */
+      if (!flags & GST_SEEK_FLAG_FLUSH)
+        goto done;
+
+      /* make sure we accept nothing anymore and return WRONG_STATE */
+      gst_collect_pads_set_flushing (adder->collect, TRUE);
+
+      /* flushing seek, start flush downstream, the flush will be done
+       * when all pads received a FLUSH_STOP. */
+      gst_pad_push_event (adder->srcpad, gst_event_new_flush_start ());
+
+      /* now wait for the collected to be finished and mark a new 
+       * segment */
+      GST_OBJECT_LOCK (adder->collect);
+      adder->segment_pending = TRUE;
+      GST_OBJECT_UNLOCK (adder->collect);
+
+    done:
       result = forward_event (adder, event);
       break;
+    }
     case GST_EVENT_NAVIGATION:
       /* navigation is rather pointless. */
       result = FALSE;

tests/check/Makefile.am

@@ -43,6 +43,7 @@ check_PROGRAMS = \
 	$(check_alsa) \
 	$(check_vorbis) \
 	$(check_theora) \
+        elements/adder	\
 	elements/audioconvert \
 	elements/audioresample \
 	elements/audiotestsrc \

tests/check/elements/adder.c

@@ -76,6 +76,7 @@ test_event_message_received (GstBus * bus, GstMessage * message,
       g_main_loop_quit (main_loop);
       break;
     default:
+      g_assert_not_reached ();
       break;
   }
 }
@@ -114,7 +115,7 @@ GST_START_TEST (test_event)
   fail_unless (res == TRUE, NULL);
 
   seek_event = gst_event_new_seek (1.0, GST_FORMAT_TIME,
-      GST_SEEK_FLAG_SEGMENT,
+      GST_SEEK_FLAG_SEGMENT | GST_SEEK_FLAG_FLUSH,
       GST_SEEK_TYPE_SET, (GstClockTime) 0,
       GST_SEEK_TYPE_SET, (GstClockTime) 2 * GST_SECOND);
 
@@ -134,9 +135,10 @@ GST_START_TEST (test_event)
   res = gst_element_set_state (bin, GST_STATE_PAUSED);
   fail_unless (res != GST_STATE_CHANGE_FAILURE, NULL);
 
-  /* FIXME, PAUSED is async and seek might not work before being prerolled.
+  /* wait for completion */
-   * though it should work in this case, as audiotestsrc is a live source
+  res = gst_element_get_state (bin, NULL, NULL, GST_CLOCK_TIME_NONE);
-   */
+  fail_unless (res != GST_STATE_CHANGE_FAILURE, NULL);
+
   res = gst_element_send_event (bin, seek_event);
   fail_unless (res == TRUE, NULL);
 
@@ -184,9 +186,18 @@ test_play_twice_message_received (GstBus * bus, GstMessage * message,
         res = gst_element_set_state (GST_ELEMENT (bin), GST_STATE_PAUSED);
         fail_unless (res != GST_STATE_CHANGE_FAILURE, NULL);
 
+        /* wait for completion */
+        res =
+            gst_element_get_state (GST_ELEMENT (bin), NULL, NULL,
+            GST_CLOCK_TIME_NONE);
+        fail_unless (res != GST_STATE_CHANGE_FAILURE, NULL);
+
         res = gst_element_send_event (GST_ELEMENT (bin), play_seek_event);
         fail_unless (res == TRUE, NULL);
 
+        /* event is now gone */
+        play_seek_event = NULL;
+
         res = gst_element_set_state (GST_ELEMENT (bin), GST_STATE_PLAYING);
         fail_unless (res != GST_STATE_CHANGE_FAILURE, NULL);
       } else {
@@ -194,6 +205,7 @@ test_play_twice_message_received (GstBus * bus, GstMessage * message,
       }
       break;
     default:
+      g_assert_not_reached ();
       break;
   }
 }
@@ -228,7 +240,7 @@ GST_START_TEST (test_play_twice)
   fail_unless (res == TRUE, NULL);
 
   play_seek_event = gst_event_new_seek (1.0, GST_FORMAT_TIME,
-      GST_SEEK_FLAG_SEGMENT,
+      GST_SEEK_FLAG_SEGMENT | GST_SEEK_FLAG_FLUSH,
       GST_SEEK_TYPE_SET, (GstClockTime) 0,
       GST_SEEK_TYPE_SET, (GstClockTime) 2 * GST_SECOND);
 
@@ -247,6 +259,12 @@ GST_START_TEST (test_play_twice)
   res = gst_element_set_state (bin, GST_STATE_PAUSED);
   fail_unless (res != GST_STATE_CHANGE_FAILURE, NULL);
 
+  /* wait for completion */
+  res =
+      gst_element_get_state (GST_ELEMENT (bin), NULL, NULL,
+      GST_CLOCK_TIME_NONE);
+  fail_unless (res != GST_STATE_CHANGE_FAILURE, NULL);
+
   res = gst_element_send_event (bin, gst_event_ref (play_seek_event));
   fail_unless (res == TRUE, NULL);