Merge pull request #102 from thaytan/issue-84

Issue 84 - send EOS when closing recording.

Fixes #84.

tools/gstrecorder.c

@@ -72,6 +72,9 @@ gst_recorder_init (GstRecorder * rec)
   rec->width = 0;
   rec->height = 0;
 
+  // Recording pipeline needs clean shut-down
+  // via EOS to close out each recording
+  GST_WORKER (rec)->send_eos_on_stop = TRUE;
   //INFO ("init %p", rec);
 }
 
@@ -287,55 +290,33 @@ gst_recorder_get_pipeline_string (GstRecorder * rec)
   desc = g_string_new ("");
 
   // Encode the video with lossless jpeg
-  g_string_append_printf (
-      desc,
-      "intervideosrc name=source_video channel=composite_video ");
-  g_string_append_printf (
-      desc,
-      "! video/x-raw,width=%d,height=%d ", rec->width, rec->height);
-  g_string_append_printf (desc, "! queue2 ");
-  g_string_append_printf (desc, "! jpegenc quality=100 ");
-  g_string_append_printf (desc, "! mux. ");
-  g_string_append_printf (desc, "\n");
+  g_string_append_printf (desc,
+      "intervideosrc name=source_video channel=composite_video "
+      "! video/x-raw,width=%d,height=%d "
+      "! queue ! jpegenc quality=100 ! mux. \n", rec->width, rec->height);
 
   // Don't encode the audio
-  g_string_append_printf (
-      desc,
-      "interaudiosrc name=source_audio channel=composite_audio ");
-  g_string_append_printf (desc, "! queue2 ");
-  g_string_append_printf (desc, "! mux. ");
-  g_string_append_printf (desc, "\n");
+  g_string_append_printf (desc,
+      "interaudiosrc name=source_audio channel=composite_audio ! queue ! mux. \n");
 
   // Output in streamable mkv format
-  g_string_append_printf (
-      desc,
-      "matroskamux name=mux"
-      " streamable=true "
-      " writing-app='gst-switch' "
-      " min-index-interval=1000000 "
-      );
+  g_string_append_printf (desc,
+      "matroskamux name=mux streamable=true "
+      " writing-app='gst-switch' min-index-interval=1000000 ");
+
   g_string_append_printf (desc, "! tee name=result ");
   g_string_append_printf (desc, "\n");
 
   if (filename) {
-    g_string_append_printf (desc, "result. ");
-    g_string_append_printf (desc, "! queue2 ");
-    g_string_append_printf (
-        desc,
+    g_string_append_printf (desc, "result. ! queue max-size-buffers=1 "
         "! filesink name=disk_sink sync=false location=\"%s\" ", filename);
     g_free ((gpointer) filename);
-    g_string_append_printf (desc, "\n");
   }
 
-  g_string_append_printf (desc, "result. ");
-  g_string_append_printf (desc, "! queue2 ");
-  g_string_append_printf (desc, "! gdppay ");
-  g_string_append_printf (
-      desc,
-      "! tcpserversink name=tcp_sink sync=false port=%d ",
-      rec->sink_port);
+  g_string_append_printf (desc, "result. ! queue max-size-buffers=1 ! gdppay "
+      "! tcpserversink name=tcp_sink sync=false port=%d ", rec->sink_port);
 
-  INFO("Recording pipeline\n----\n%s\n---", desc->str);
+  INFO ("Recording pipeline\n----\n%s\n---", desc->str);
 
   return desc;
 }

tools/gstworker.c

@@ -84,6 +84,7 @@ gst_worker_init (GstWorker * worker)
   worker->watch = 0;
 
   g_mutex_init (&worker->pipeline_lock);
+  g_cond_init (&worker->shutdown_cond);
 
   //INFO ("gst_worker init %p", worker);
 }
@@ -145,6 +146,7 @@ gst_worker_finalize (GstWorker * worker)
 
   INFO ("gst_worker finalize %p", worker);
   g_mutex_clear (&worker->pipeline_lock);
+  g_cond_clear (&worker->shutdown_cond);
 
   g_free (worker->name);
   worker->name = NULL;
@@ -378,29 +380,34 @@ gboolean
 gst_worker_stop_force (GstWorker * worker, gboolean force)
 {
   GstStateChangeReturn ret = GST_STATE_CHANGE_FAILURE;
+  gboolean no_eos;
 
   g_return_val_if_fail (GST_IS_WORKER (worker), FALSE);
 
   GST_WORKER_LOCK_PIPELINE (worker);
 
   if (worker->pipeline) {
-#if 1
     GstState state;
 
     ret = gst_element_get_state (worker->pipeline, &state, NULL,
         GST_CLOCK_TIME_NONE);
+    no_eos = (state == GST_STATE_PLAYING) && !worker->send_eos_on_stop;
 
-    if (state == GST_STATE_PLAYING || force) {
+    if (force || no_eos) {
       ret = gst_element_set_state (worker->pipeline, GST_STATE_NULL);
 
       gst_bus_set_flushing (worker->bus, TRUE);
 
       g_timeout_add (5,
           (GSourceFunc) gst_worker_state_ready_to_null_proxy, worker);
     }
-#else
-    ret = gst_element_set_state (worker->pipeline, GST_STATE_NULL);
-#endif
+    else if (state == GST_STATE_PLAYING) {
+      /* Send an EOS to cleanly shutdown */
+      gst_element_send_event (worker->pipeline, gst_event_new_eos());
+
+      /* Go to sleep until the EOS handler calls stop_force (worker, TRUE); */
+      g_cond_wait (&worker->shutdown_cond, &worker->pipeline_lock);
+    }
   }
 
   GST_WORKER_UNLOCK_PIPELINE (worker);
@@ -449,7 +456,7 @@ gst_worker_missing_plugin (GstWorker *worker, GstStructure *structure)
 static void
 gst_worker_handle_eos (GstWorker * worker)
 {
-  gst_worker_stop (worker);
+  gst_worker_stop_force (worker, TRUE);
 }
 
 static void
@@ -603,6 +610,21 @@ gst_worker_pipeline_state_changed (GstWorker * worker,
   return TRUE;
 }
 
+static GstBusSyncReply
+gst_worker_message_sync (GstBus * bus, GstMessage * message, GstWorker *worker)
+{
+  switch (GST_MESSAGE_TYPE (message)) {
+    case GST_MESSAGE_EOS:
+      /* When we see EOS, wake up and gst_worker_stop that might be waiting */
+      g_cond_signal (&worker->shutdown_cond);
+      break;
+    default:
+      break;
+  }
+
+  return GST_BUS_PASS;
+}
+
 static gboolean
 gst_worker_message (GstBus * bus, GstMessage * message, GstWorker * worker)
 {
@@ -762,6 +784,8 @@ gst_worker_prepare_unsafe (GstWorker * worker)
   if (!worker->watch)
     goto error_add_watch;
 
+  gst_bus_set_sync_handler (worker->bus, (GstBusSyncHandler) (gst_worker_message_sync), worker, NULL);
+
   if (workerclass->prepare && !workerclass->prepare (worker))
     goto error_prepare;
 

tools/gstworker.h

@@ -105,6 +105,7 @@ struct _GstWorker
     //GstSwitchServer *server; /*!<  */
 
   GMutex pipeline_lock; /*!< Mutex for %pipeline */
+  GCond shutdown_cond; /*!< Cond for shutting down pipelines cleanly */
   GstElement *pipeline; /*!< The pipeline. */
   GstBus *bus; /*!< The pipeline bus. */
 
@@ -115,6 +116,11 @@ struct _GstWorker
   gboolean auto_replay; /*!< The worker should replay if it's TRUE */
   gboolean paused_for_buffering; /*!< Mark for buffering pause. */
   guint watch; /*!< The watch number of the pipeline bus. */
+
+  /*!< TRUE if the recording pipeline needs clean shut-down
+   * via an EOS event to finish up before stopping
+   */
+  gboolean send_eos_on_stop;
 };
 
 /**