libs/gst/base/gstcollectpads.c: Make sure the EOS flag is cleared from pads after a flush or stop. Fixes #343538.

Original commit message from CVS:
Patch by: Alessandro Decina <alessandro at nnva dot org>
* libs/gst/base/gstcollectpads.c: (gst_collect_pads_stop):
Make sure the EOS flag is cleared from pads after a flush
or stop. Fixes #343538.
* tests/check/libs/collectpads.c: (GST_START_TEST),
(gst_collect_pads_suite):
Added test for collectpads reusage after EOS.

ChangeLog

@@ -1,3 +1,15 @@
+2006-06-01  Wim Taymans  <wim@fluendo.com>
+
+	Patch by: Alessandro Decina <alessandro at nnva dot org>
+
+	* libs/gst/base/gstcollectpads.c: (gst_collect_pads_stop):
+	Make sure the EOS flag is cleared from pads after a flush
+	or stop. Fixes #343538.
+
+	* tests/check/libs/collectpads.c: (GST_START_TEST),
+	(gst_collect_pads_suite):
+	Added test for collectpads reusage after EOS.
+
 2006-05-30  Sebastien Moutte  <sebastien@moutte.net>
 
 	* gst/gst.c:

common

@@ -1 +1 @@
-Subproject commit b0fd90b1cfb51107e8a511a1f3983a06bdd18638
+Subproject commit 80b4e2c08717159760ebaa40715f8919b3f77ba9

libs/gst/base/gstcollectpads.c

@@ -515,6 +515,8 @@ gst_collect_pads_stop (GstCollectPads * pads)
 
   /* Stop collect pads */
   pads->started = FALSE;
+  pads->eospads = 0;
+  pads->queuedpads = 0;
 
   /* loop over the master pad list and flush buffers */
   collected = pads->abidata.ABI.pad_list;
@@ -527,9 +529,8 @@ gst_collect_pads_stop (GstCollectPads * pads)
       buffer_p = &data->buffer;
       gst_buffer_replace (buffer_p, NULL);
       data->pos = 0;
-      /* one less pad with queued data now */
-      pads->queuedpads--;
     }
+    data->abidata.ABI.eos = FALSE;
   }
 
   GST_COLLECT_PADS_PAD_UNLOCK (pads);

tests/check/libs/collectpads.c

@@ -24,6 +24,15 @@
 #include <gst/check/gstcheck.h>
 #include <gst/base/gstcollectpads.h>
 
+#define fail_unless_collected(expected)           \
+G_STMT_START {                                    \
+  g_mutex_lock (lock);                            \
+  while (expected == TRUE && collected == FALSE)  \
+    g_cond_wait (cond, lock);                     \
+  fail_unless_equals_int (collected, expected);   \
+  g_mutex_unlock (lock);                          \
+} G_STMT_END;
+
 typedef struct
 {
   char foo;
@@ -157,18 +166,14 @@ GST_START_TEST (test_collect)
   data1->buffer = buf1;
   thread1 = g_thread_create (push_buffer, data1, TRUE, NULL);
   /* here thread1 is blocked and srcpad1 has a queued buffer */
-  fail_unless (collected == FALSE);
+  fail_unless_collected (FALSE);
 
   data2->pad = srcpad2;
   data2->buffer = buf2;
   thread2 = g_thread_create (push_buffer, data2, TRUE, NULL);
 
   /* now both pads have a buffer */
-  g_mutex_lock (lock);
+  fail_unless_collected (TRUE);
-  while (collected == FALSE)
-    g_cond_wait (cond, lock);
-  fail_unless (collected == TRUE);
-  g_mutex_unlock (lock);
 
   tmp = gst_collect_pads_pop (collect, (GstCollectData *) data1);
   fail_unless (tmp == buf1);
@@ -211,17 +216,13 @@ GST_START_TEST (test_collect_eos)
   data1->buffer = buf1;
   thread1 = g_thread_create (push_buffer, data1, TRUE, NULL);
   /* here thread1 is blocked and srcpad1 has a queued buffer */
-  fail_unless (collected == FALSE);
+  fail_unless_collected (FALSE);
 
   data2->pad = srcpad2;
   data2->event = gst_event_new_eos ();
   thread2 = g_thread_create (push_event, data2, TRUE, NULL);
   /* now sinkpad1 has a buffer and sinkpad2 has EOS */
-  g_mutex_lock (lock);
+  fail_unless_collected (TRUE);
-  while (collected == FALSE)
-    g_cond_wait (cond, lock);
-  fail_unless (collected == TRUE);
-  g_mutex_unlock (lock);
 
   tmp = gst_collect_pads_pop (collect, (GstCollectData *) data1);
   fail_unless (tmp == buf1);
@@ -241,6 +242,91 @@ GST_START_TEST (test_collect_eos)
 
 GST_END_TEST;
 
+GST_START_TEST (test_collect_twice)
+{
+  GstBuffer *buf1, *buf2, *tmp;
+  GThread *thread1, *thread2;
+
+  data1 = (TestData *) gst_collect_pads_add_pad (collect,
+      sinkpad1, sizeof (TestData));
+  fail_unless (data1 != NULL);
+
+  data2 = (TestData *) gst_collect_pads_add_pad (collect,
+      sinkpad2, sizeof (TestData));
+  fail_unless (data2 != NULL);
+
+  buf1 = gst_buffer_new ();
+
+  /* start collect pads */
+  gst_collect_pads_start (collect);
+
+  /* queue a buffer */
+  data1->pad = srcpad1;
+  data1->buffer = buf1;
+  thread1 = g_thread_create (push_buffer, data1, TRUE, NULL);
+  /* here thread1 is blocked and srcpad1 has a queued buffer */
+  fail_unless_collected (FALSE);
+
+  /* push EOS on the other pad */
+  data2->pad = srcpad2;
+  data2->event = gst_event_new_eos ();
+  thread2 = g_thread_create (push_event, data2, TRUE, NULL);
+
+  /* one of the pads has a buffer, the other has EOS */
+  fail_unless_collected (TRUE);
+
+  tmp = gst_collect_pads_pop (collect, (GstCollectData *) data1);
+  fail_unless (tmp == buf1);
+  /* there's nothing to pop from the one which received EOS */
+  tmp = gst_collect_pads_pop (collect, (GstCollectData *) data2);
+  fail_unless (tmp == NULL);
+
+  /* these will return immediately as at this point the threads have been
+   * unlocked and are finished */
+  g_thread_join (thread1);
+  g_thread_join (thread2);
+
+  gst_collect_pads_stop (collect);
+  collected = FALSE;
+
+  buf2 = gst_buffer_new ();
+
+  /* start collect pads */
+  gst_collect_pads_start (collect);
+
+  /* push buffers on the pads */
+  data1->pad = srcpad1;
+  data1->buffer = buf1;
+  thread1 = g_thread_create (push_buffer, data1, TRUE, NULL);
+  /* here thread1 is blocked and srcpad1 has a queued buffer */
+  fail_unless_collected (FALSE);
+
+  data2->pad = srcpad2;
+  data2->buffer = buf2;
+  thread2 = g_thread_create (push_buffer, data2, TRUE, NULL);
+
+  /* now both pads have a buffer */
+  fail_unless_collected (TRUE);
+
+  tmp = gst_collect_pads_pop (collect, (GstCollectData *) data1);
+  fail_unless (tmp == buf1);
+  tmp = gst_collect_pads_pop (collect, (GstCollectData *) data2);
+  fail_unless (tmp == buf2);
+
+  /* these will return immediately as at this point the threads have been
+   * unlocked and are finished */
+  g_thread_join (thread1);
+  g_thread_join (thread2);
+
+  gst_collect_pads_stop (collect);
+
+  gst_buffer_unref (buf1);
+  gst_buffer_unref (buf2);
+
+}
+
+GST_END_TEST;
+
 static Suite *
 gst_collect_pads_suite ()
 {
@@ -254,6 +340,7 @@ gst_collect_pads_suite ()
   tcase_add_test (general, test_pad_add_remove);
   tcase_add_test (general, test_collect);
   tcase_add_test (general, test_collect_eos);
+  tcase_add_test (general, test_collect_twice);
 
   return suite;
 }