identity: Return FLUSHING instead of EOS and don't start waiting for anything if currently flushing

Otherwise we might try unscheduling a clock id (that does not exist
yet), then the streaming thread waits for id and the state change never
continues because the streaming thread is blocked.

Also shutting down and flushing and similar should return FLUSHING, not
EOS. The stream is not over, we're just not accepting any buffers
anymore.

2 files changed, 19 insertions, 2 deletions

diff --git a/plugins/elements/gstidentity.c b/plugins/elements/gstidentity.c
index 119fb5886..6867c3eba 100644
--- a/plugins/elements/gstidentity.c
+++ b/plugins/elements/gstidentity.c
@@ -305,9 +305,18 @@ gst_identity_do_sync (GstIdentity * identity, GstClockTime running_time)
 
     GST_OBJECT_LOCK (identity);
 
+    if (identity->flushing) {
+      GST_OBJECT_UNLOCK (identity);
+      return GST_FLOW_FLUSHING;
+    }
+
     while (identity->blocked)
       g_cond_wait (&identity->blocked_cond, GST_OBJECT_GET_LOCK (identity));
 
+    if (identity->flushing) {
+      GST_OBJECT_UNLOCK (identity);
+      return GST_FLOW_FLUSHING;
+    }
 
     if ((clock = GST_ELEMENT (identity)->clock)) {
       GstClockReturn cret;
@@ -336,8 +345,8 @@ gst_identity_do_sync (GstIdentity * identity, GstClockTime running_time)
         gst_clock_id_unref (identity->clock_id);
         identity->clock_id = NULL;
       }
-      if (cret == GST_CLOCK_UNSCHEDULED)
-        ret = GST_FLOW_EOS;
+      if (cret == GST_CLOCK_UNSCHEDULED || identity->flushing)
+        ret = GST_FLOW_FLUSHING;
     }
     GST_OBJECT_UNLOCK (identity);
   }
@@ -430,11 +439,16 @@ gst_identity_sink_event (GstBaseTransform * trans, GstEvent * event)
   } else {
     if (GST_EVENT_TYPE (event) == GST_EVENT_FLUSH_START) {
       GST_OBJECT_LOCK (identity);
+      identity->flushing = TRUE;
       if (identity->clock_id) {
         GST_DEBUG_OBJECT (identity, "unlock clock wait");
         gst_clock_id_unschedule (identity->clock_id);
       }
       GST_OBJECT_UNLOCK (identity);
+    } else if (GST_EVENT_TYPE (event) == GST_EVENT_FLUSH_STOP) {
+      GST_OBJECT_LOCK (identity);
+      identity->flushing = FALSE;
+      GST_OBJECT_UNLOCK (identity);
     }
 
     ret = GST_BASE_TRANSFORM_CLASS (parent_class)->sink_event (trans, event);
@@ -917,6 +931,7 @@ gst_identity_change_state (GstElement * element, GstStateChange transition)
       break;
     case GST_STATE_CHANGE_READY_TO_PAUSED:
       GST_OBJECT_LOCK (identity);
+      identity->flushing = FALSE;
       identity->blocked = TRUE;
       GST_OBJECT_UNLOCK (identity);
       if (identity->sync)
@@ -930,6 +945,7 @@ gst_identity_change_state (GstElement * element, GstStateChange transition)
       break;
     case GST_STATE_CHANGE_PAUSED_TO_READY:
       GST_OBJECT_LOCK (identity);
+      identity->flushing = TRUE;
       if (identity->clock_id) {
         GST_DEBUG_OBJECT (identity, "unlock clock wait");
         gst_clock_id_unschedule (identity->clock_id);
diff --git a/plugins/elements/gstidentity.h b/plugins/elements/gstidentity.h
index 5dc229a30..e622b491a 100644
--- a/plugins/elements/gstidentity.h
+++ b/plugins/elements/gstidentity.h
@@ -55,6 +55,7 @@ struct _GstIdentity {
 
   /*< private >*/
   GstClockID     clock_id;
+  gboolean       flushing;
   gint 	 	 error_after;
   gfloat 	 drop_probability;
   gint		 datarate;