Restart source on EOS

src/main.c

@@ -239,6 +239,80 @@ webrtc_client_disconnected_cb (MssHttpServer *server, MssClientId client_id,
   }
 }
 
+struct RestartData {
+  GstElement *src;
+  GstElement *pipeline;
+};
+
+static void
+free_restart_data (gpointer user_data)
+{
+  struct RestartData *rd = user_data;
+
+  gst_object_unref (rd->src);
+  g_free (rd);
+}
+
+static gboolean
+restart_source (gpointer user_data)
+{
+  struct RestartData *rd = user_data;
+  GstElement *tsparse;
+  GstStateChangeReturn ret;
+
+  gst_element_set_state (rd->src, GST_STATE_NULL);
+  gst_element_set_locked_state (rd->src, TRUE);
+  tsparse = gst_bin_get_by_name (GST_BIN (rd->pipeline), "tsparse");
+  gst_bin_add (GST_BIN (rd->pipeline), rd->src);
+  gst_element_link (rd->src, tsparse);
+  gst_element_set_locked_state (rd->src, FALSE);
+  ret = gst_element_set_state (rd->src, GST_STATE_PLAYING);
+  g_assert (ret != GST_STATE_CHANGE_FAILURE);
+  gst_object_unref (tsparse);
+
+  g_debug ("Restarted source after EOS");
+
+  return G_SOURCE_REMOVE;
+}
+
+static GstPadProbeReturn
+src_event_cb (GstPad *pad, GstPadProbeInfo *info, gpointer user_data)
+{
+  GstElement *pipeline = user_data;
+  GstElement *src;
+  struct RestartData *rd;
+
+  if (GST_EVENT_TYPE (GST_PAD_PROBE_INFO_EVENT (info)) != GST_EVENT_EOS)
+    return GST_PAD_PROBE_PASS;
+
+  src = gst_pad_get_parent_element (pad);
+
+  gst_bin_remove (GST_BIN (pipeline), src);
+
+  rd = g_new (struct RestartData, 1);
+  rd->src = src;
+  rd->pipeline = pipeline;
+  g_idle_add_full (G_PRIORITY_HIGH_IDLE, restart_source, rd, free_restart_data);
+
+  return GST_PAD_PROBE_DROP;
+}
+
+static gboolean
+print_stats (gpointer user_data)
+{
+  GstElement *src = user_data;
+  GstStructure *s;
+  char *str;
+
+  g_object_get (src, "stats", &s, NULL);
+  str = gst_structure_to_string (s);
+  g_debug ("%s", str);
+  g_free (str);
+  gst_structure_free (s);
+
+  return G_SOURCE_CONTINUE;
+}
+
 int main (int argc, char *argv[])
 {
   GOptionContext *option_context;
@@ -247,6 +321,8 @@ int main (int argc, char *argv[])
   GstElement *pipeline;
   GError *error = NULL;
   GstBus *bus;
+  GstElement *src;
+  GstPad *srcpad;
   GstStateChangeReturn ret;
 
   option_context = g_option_context_new (NULL);
@@ -266,7 +342,7 @@ int main (int argc, char *argv[])
 
   http_server = mss_http_server_new ();
 
-  pipeline_str = g_strdup_printf ("srtsrc uri=%s?mode=listener ! tsparse ! tee name=t ! "
+  pipeline_str = g_strdup_printf ("srtsrc uri=%s?mode=listener name=src ! tsparse name=tsparse ! tee name=t ! "
       "queue ! tsdemux latency=50 ! decodebin ! videoconvert ! autovideosink "
       "t. ! queue leaky=downstream max-size-buffers=400 ! tsdemux ! h264parse ! video/x-h264, stream-format=avc ! h264parse ! mpegtsmux ! hlssink location=%s/segment%%05d.ts playlist-location=%s/playlist.m3u8 target-duration=1 playlist-length=3 "
       "t. ! queue leaky=downstream max-size-buffers=10 ! rtpmp2tpay ! ristsink bonding-addresses=%s "
@@ -304,6 +380,14 @@ int main (int argc, char *argv[])
       srt_uri,
       rist_addresses);
 
+  src = gst_bin_get_by_name (GST_BIN (pipeline), "src");
+  srcpad = gst_element_get_static_pad (src, "src");
+  gst_pad_add_probe (srcpad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM, src_event_cb,
+      pipeline, NULL);
+  g_timeout_add (1000, print_stats, src);
+  gst_object_unref (srcpad);
+  gst_object_unref (src);
+
   ret = gst_element_set_state (pipeline, GST_STATE_PLAYING);
   g_assert (ret != GST_STATE_CHANGE_FAILURE);