building pipeline manually now

1 files changed, 213 insertions, 40 deletions

diff --git a/src/sydra.c b/src/sydra.c
index 9152f9e..2c03b92 100644
--- a/src/sydra.c
+++ b/src/sydra.c
@@ -153,61 +153,234 @@ static gboolean bus_call(GstBus *bus, GstMessage *msg, gpointer data)
   return TRUE;
 }
 
-static char* build_sender_pipeline_desc(options_t* opt)
+static GstElement* sydra_create_bin_from_desc(const char* type, const char* desc)
 {
-  struct timespec now;
-  clock_gettime(CLOCK_REALTIME, &now);
-  struct tm bd_time;
-  localtime_r(&(now.tv_sec), &bd_time);
-  char recfile[1024]; //TODO: fix this hardcoded length
-  strftime(recfile, sizeof(recfile), opt->rec_name_format_, &bd_time);
+  GError *error = NULL;
+  GstElement *bin =  gst_parse_bin_from_description(desc, TRUE, &error);
+  if(!bin) {
+    log_printf(ERROR, "Bin description for %s parser error: %s", type, error ? error->message : "unknown");
+    g_error_free(error);
+    return NULL;
+  }
+  if(error) {
+    log_printf(WARNING, "Bin description for %s parser warning: %s", type, error ? error->message : "unknown");
+    g_error_free(error);
+  }
+  return bin;
+}
+
+static GstElement* sydra_create_element(const char* type, const char* name)
+{
+  GstElement *e = gst_element_factory_make (type, name);
+  if(!e) {
+    log_printf(ERROR, "Error creating element %s%sof type %s", name ? name : "", name ? " " : "", type);
+    return NULL;
+  }
+  return e;
+}
+
+static gboolean sydra_link_pads(GstElement* src, GstPad* src_pad, const char* src_pad_name, GstElement* sink, GstPad* sink_pad, const char* sink_pad_name)
+{
+  if(!src_pad || !sink_pad)
+    return FALSE;
+
+  GstPadLinkReturn ret = gst_pad_link(src_pad, sink_pad);
+  gst_object_unref(GST_OBJECT(src_pad));
+  gst_object_unref(GST_OBJECT(sink_pad));
+
+  if(GST_PAD_LINK_FAILED(ret)) {
+    gchar* src_name = gst_element_get_name(src);
+    gchar* sink_name = gst_element_get_name(sink);
+    log_printf(ERROR, "Error linking request pad '%s' of '%s' with static pad '%s' of '%s'",
+               src_pad_name, src_name, sink_pad_name, sink_name);
+    g_free(src_name);
+    g_free(sink_name);
+    return FALSE;
+  }
+  return TRUE;
+}
+
+static gboolean sydra_link_request_static(GstElement* src, const char* src_pad_name, GstElement* sink, const char* sink_pad_name)
+{
+  GstPad *src_pad = gst_element_get_request_pad(src, src_pad_name);
+  GstPad *sink_pad = gst_element_get_static_pad(sink, sink_pad_name);
+  return sydra_link_pads(src, src_pad, src_pad_name, sink, sink_pad, sink_pad_name);
+}
+
+static gboolean sydra_link_static_request(GstElement* src, const char* src_pad_name, GstElement* sink, const char* sink_pad_name)
+{
+  GstPad *src_pad = gst_element_get_static_pad(src, src_pad_name);
+  GstPad *sink_pad = gst_element_get_request_pad(sink, sink_pad_name);
+  return sydra_link_pads(src, src_pad, src_pad_name, sink, sink_pad, sink_pad_name);
+}
+
+static gboolean sydra_link_static_static(GstElement* src, const char* src_pad_name, GstElement* sink, const char* sink_pad_name)
+{
+  GstPad *src_pad = gst_element_get_static_pad(src, src_pad_name);
+  GstPad *sink_pad = gst_element_get_static_pad(sink, sink_pad_name);
+  return sydra_link_pads(src, src_pad, src_pad_name, sink, sink_pad, sink_pad_name);
+}
+
+struct media_elements {
+  const char* name_;
+
+  const char* src_str_;
+  GstElement* src_;
+
+  GstElement* tee_raw_;
+
+  const char* enc_str_;
+  GstElement* enc_;
+
+  GstElement* tee_enc_;
+
+  const char* payloader_str_;
+  GstElement* payloader_;
+};
+
+static gboolean create_media_elements(struct media_elements *me, GstElement* pipeline, GstElement *rtp, uint32_t idx)
+{
+  char bin_name[32];
+
+  snprintf(bin_name, sizeof(bin_name), "%s source", me->name_);
+  me->src_ = sydra_create_bin_from_desc(bin_name, me->src_str_);
+  me->tee_raw_ = sydra_create_element("tee", NULL);
+  GstElement *qr = sydra_create_element("queue", NULL);
+
+  snprintf(bin_name, sizeof(bin_name), "%s encoder", me->name_);
+  me->enc_ = sydra_create_bin_from_desc(bin_name, me->enc_str_);
+  me->tee_enc_ = sydra_create_element("tee", NULL);
+  GstElement *qe = sydra_create_element("queue", NULL);
+
+  me->payloader_ = sydra_create_element(me->payloader_str_, NULL);
+
+  if(!me->src_ || !me->tee_raw_ || !qr || !me->enc_ || !me->tee_enc_ || !qe || !me->payloader_) {
+    return FALSE;
+  }
+
+  log_printf(DEBUG, "%s path created successfully!", me->name_);
+
+  gst_bin_add_many (GST_BIN(pipeline), me->src_, me->tee_raw_, qr, me->enc_, me->tee_enc_, qe, me->payloader_, NULL);
+  gst_element_link(me->src_, me->tee_raw_);
+  gst_element_link_many(qr, me->enc_, me->tee_enc_, NULL);
+  gst_element_link(qe, me->payloader_);
+
+  char pad_name[32];
+  snprintf(pad_name, sizeof(bin_name), "send_rtp_sink_%u", idx);
+  if(!sydra_link_request_static(me->tee_raw_, "src_%u", qr, "sink") ||
+     !sydra_link_request_static(me->tee_enc_, "src_%u", qe, "sink") ||
+     !sydra_link_static_request(me->payloader_, "src", rtp, pad_name)) {
+    return FALSE;
+  }
+
+  log_printf(DEBUG, "%s path linked successfully!", me->name_);
+  return TRUE;
+}
+
+static gboolean create_udp_elements(options_t* opt, GstElement* pipeline, GstElement* rtp)
+{
+  GstElement *udp_rtp_video = sydra_create_element("udpsink", "udprtpv");
+  GstElement *udp_rtcp_video = sydra_create_element("udpsink", "udprtcpv");
+  GstElement *udp_rtp_audio = sydra_create_element("udpsink", "udprtpa");
+  GstElement *udp_rtcp_audio = sydra_create_element("udpsink", "udprtcpa");
+
+  if(!udp_rtp_video || !udp_rtcp_video || !udp_rtp_audio || !udp_rtcp_audio)
+    return FALSE;
+
+  log_printf(DEBUG, "udp elements created successfully!");
 
   int rtp_port_v = opt->rtp_port_base_;
   int rtcp_port_v = opt->rtp_port_base_ + 100;
   int rtp_port_a = opt->rtp_port_base_ + 200;
   int rtcp_port_a = opt->rtp_port_base_ + 300;
 
-  char* sender_desc;
-  int slen = asprintf(&sender_desc, "rtpbin name=rtpbin                              \
-    %s ! tee name=vt                                                                 \
-        vt. ! queue silent=true ! %s ! tee name=cvt                                  \
-        cvt. ! %s ! rtpbin.send_rtp_sink_0                                           \
-        rtpbin.send_rtp_src_0 ! udpsink port=%d host=%s                              \
-        rtpbin.send_rtcp_src_0 ! udpsink port=%d host=%s sync=false async=false      \
-    %s ! tee name=at                                                                 \
-        at. ! queue silent=true ! %s ! %s ! rtpbin.send_rtp_sink_1                   \
-        rtpbin.send_rtp_src_1 ! udpsink port=%d host=%s                              \
-        rtpbin.send_rtcp_src_1 ! udpsink port=%d host=%s sync=false async=false      \
-    at. ! queue silent=true ! %s name=recmux ! filesink location=\"%s\"              \
-    vt. ! queue silent=true ! textoverlay text=\" local \" shaded-background=true halignment=center valignment=baseline font-desc=\"Sans 18\" ! %s",
-     opt->video_src_, opt->video_enc_, opt->video_payloader_, rtp_port_v, opt->rtp_host_, rtcp_port_v, opt->rtp_host_,
-     opt->audio_src_, opt->audio_enc_stream_, opt->audio_payloader_, rtp_port_a, opt->rtp_host_, rtcp_port_a, opt->rtp_host_,
-     opt->rec_mux_, recfile, opt->videosink_);
-
-  return (slen < 0) ? NULL : sender_desc;
+  g_object_set(G_OBJECT(udp_rtp_video), "host", opt->rtp_host_, "port", rtp_port_v, NULL);
+  g_object_set(G_OBJECT(udp_rtcp_video), "host", opt->rtp_host_, "port", rtcp_port_v, "sync", FALSE, "async", FALSE, NULL);
+  g_object_set(G_OBJECT(udp_rtp_audio), "host", opt->rtp_host_, "port", rtp_port_a, NULL);
+  g_object_set(G_OBJECT(udp_rtcp_audio), "host", opt->rtp_host_, "port", rtcp_port_a, "sync", FALSE, "async", FALSE, NULL);
+
+  gst_bin_add_many(GST_BIN (pipeline), udp_rtp_video, udp_rtcp_video, udp_rtp_audio, udp_rtcp_audio, NULL);
+
+  if(!sydra_link_static_static(rtp, "send_rtp_src_0", udp_rtp_video, "sink") ||
+     !sydra_link_request_static(rtp, "send_rtcp_src_0", udp_rtcp_video, "sink") ||
+     !sydra_link_static_static(rtp, "send_rtp_src_1", udp_rtp_audio, "sink") ||
+     !sydra_link_request_static(rtp, "send_rtcp_src_1", udp_rtcp_audio, "sink"))
+    return FALSE;
+
+  log_printf(DEBUG, "udp elements linked successfully!");
+
+  return TRUE;
 }
 
-int main_loop(options_t* opt)
+static GstElement* create_pipeline(options_t* opt)
 {
-  log_printf(INFO, "entering main loop");
+  log_printf(DEBUG, "creating pipeline");
 
-  char* sender_desc = build_sender_pipeline_desc(opt);
-  if(!sender_desc) {
-    log_printf(ERROR, "memory error while building sender pipeline description");
-    return -1;
+  GstElement *pipeline = gst_pipeline_new ("sydra");
+  if(!pipeline) {
+    log_printf(ERROR, "Creating pipeline failed");
+    return NULL;
+  }
+  GstElement *rtp = sydra_create_element("rtpbin", "rtpbin");
+  if(!rtp || !gst_bin_add(GST_BIN(pipeline), rtp)) {
+    return NULL;
   }
 
-  GError *error = NULL;
-  GstElement *pipeline = gst_parse_launch(sender_desc, &error);
-  if(!pipeline || error) {
-    log_printf(ERROR, "Pipeline parser Error: %s", error ? error->message : "unknown");
-    g_error_free(error);
-    free(sender_desc);
-    if(pipeline)
-      gst_object_unref(GST_OBJECT(pipeline));
+  log_printf(DEBUG, "rtpbin created");
+
+  struct media_elements video = { "video", opt->video_src_, NULL, NULL,
+                                  opt->video_enc_, NULL, NULL,
+                                  opt->video_payloader_, NULL };
+  struct media_elements audio = { "audio", opt->audio_src_, NULL, NULL,
+                                  opt->audio_enc_stream_, NULL, NULL,
+                                  opt->audio_payloader_, NULL };
+  if(!create_media_elements(&video, pipeline, rtp, 0) ||
+     !create_media_elements(&audio, pipeline, rtp, 1) ||
+     !create_udp_elements(opt, pipeline, rtp)) {
+    return NULL;
+  }
+
+  return pipeline;
+}
+
+/* static char* build_sender_pipeline_desc(options_t* opt) */
+/* { */
+  /* struct timespec now; */
+  /* clock_gettime(CLOCK_REALTIME, &now); */
+  /* struct tm bd_time; */
+  /* localtime_r(&(now.tv_sec), &bd_time); */
+  /* char recfile[1024]; //TODO: fix this hardcoded length */
+  /* strftime(recfile, sizeof(recfile), opt->rec_name_format_, &bd_time); */
+
+/*   char* sender_desc; */
+/*   int slen = asprintf(&sender_desc, "rtpbin name=rtpbin                              \ */
+/*     %s ! tee name=vt                                                                 \ */
+/*         vt. ! queue silent=true ! %s ! tee name=cvt                                  \ */
+/*         cvt. ! %s ! rtpbin.send_rtp_sink_0                                           \ */
+/*         rtpbin.send_rtp_src_0 ! udpsink port=%d host=%s                              \ */
+/*         rtpbin.send_rtcp_src_0 ! udpsink port=%d host=%s sync=false async=false      \ */
+/*     %s ! tee name=at                                                                 \ */
+/*         at. ! queue silent=true ! %s ! %s ! rtpbin.send_rtp_sink_1                   \ */
+/*         rtpbin.send_rtp_src_1 ! udpsink port=%d host=%s                              \ */
+/*         rtpbin.send_rtcp_src_1 ! udpsink port=%d host=%s sync=false async=false      \ */
+/*     at. ! queue silent=true ! %s name=recmux ! filesink location=\"%s\"              \ */
+/*     vt. ! queue silent=true ! textoverlay text=\" local \" shaded-background=true halignment=center valignment=baseline font-desc=\"Sans 18\" ! %s", */
+/*      opt->video_src_, opt->video_enc_, opt->video_payloader_, rtp_port_v, opt->rtp_host_, rtcp_port_v, opt->rtp_host_, */
+/*      opt->audio_src_, opt->audio_enc_stream_, opt->audio_payloader_, rtp_port_a, opt->rtp_host_, rtcp_port_a, opt->rtp_host_, */
+/*      opt->rec_mux_, recfile, opt->videosink_); */
+
+/*   return (slen < 0) ? NULL : sender_desc; */
+/* } */
+
+int main_loop(options_t* opt)
+{
+  log_printf(INFO, "entering main loop");
+
+  GstElement *pipeline = create_pipeline(opt);
+  if(!pipeline) {
+    log_printf(ERROR, "creating pipeline failed");
     return -1;
   }
-  free(sender_desc);
 
   GMainLoop *loop = g_main_loop_new(NULL, FALSE);