#include <stdio.h>
#include <stdlib.h>
#include <gst/gst.h>
#include <glib.h>
#include <math.h>
#include <gmodule.h>
#include <string.h>
#include <sys/time.h>
#include <sys/wait.h>
#include <unistd.h>
#include "gstnvdsmeta.h"
#include "gst-nvmessage.h"

#include "deepstream_common.h"
#include "deepstream_config.h"
// #include "deepstream_sinks.h"
// #include "deepstream_app.h"
#include "nvdsmeta.h"
#include <cuda_runtime_api.h>
#include <gst/rtsp-server/rtsp-server.h>



#define MAX_DISPLAY_LEN 64

#define PGIE_CLASS_ID_VEHICLE 0
#define PGIE_CLASS_ID_PERSON 2
#define SET_GPU_ID(object, gpu_id) g_object_set (G_OBJECT (object), "gpu-id", gpu_id, NULL);
#define SET_MEMORY(object, mem_id) g_object_set (G_OBJECT (object), "nvbuf-memory-type", mem_id, NULL);
#define SINK_ELEMENT "nveglglessink"


/* The muxer output resolution must be set if the input streams will be of
 * different resolution. The muxer will scale all the input frames to this
 * resolution. */
#define MUXER_OUTPUT_WIDTH 1920
#define MUXER_OUTPUT_HEIGHT 1080
#define GST_CAPS_FEATURES_NVMM "memory:NVMM"
#define TILED_OUTPUT_WIDTH 1280
#define TILED_OUTPUT_HEIGHT 720
#define GPU_ID 0
#define MAX_NUM_SOURCES 3
#define RTSP_PORT_NUM 8554
#define BITRATE 4000000

#define CONFIG_GPU_ID "gpu-id"
#define PGIE_CONFIG_FILE "dstest_pgie_config.txt"


char codec[4]="H265";


gint g_num_sources = 0;
gint g_source_id_list[MAX_NUM_SOURCES];
gboolean g_eos_list[MAX_NUM_SOURCES];
gboolean g_source_enabled[MAX_NUM_SOURCES];
GstElement **g_source_bin_list = NULL;
GMutex eos_lock;

GMainLoop *loop=NULL;
GstElement *pipeline=NULL, *streammux=NULL, *pgie=NULL,
          *nvosd=NULL, *tiler=NULL, *nvvidconv=NULL, *streamdemux=NULL;
// gchar *uri[100];
gchar uri[1000];
int updsink_port = 5400;

static gboolean
bus_call (GstBus * bus, GstMessage * msg, gpointer data)
{
  GMainLoop *loop = (GMainLoop *) data;
  switch (GST_MESSAGE_TYPE (msg)) {
    case GST_MESSAGE_EOS:
      g_print ("End of stream\n");
      g_main_loop_quit (loop);
      break;
    case GST_MESSAGE_WARNING:
    {
      gchar *debug;
      GError *error;
      gst_message_parse_warning (msg, &error, &debug);
      g_printerr ("WARNING from element %s: %s\n",
          GST_OBJECT_NAME (msg->src), error->message);
      g_free (debug);
      g_printerr ("Warning: %s\n", error->message);
      g_error_free (error);
      break;
    }
    case GST_MESSAGE_ERROR:
    {
      gchar *debug;
      GError *error;
      gst_message_parse_error (msg, &error, &debug);
      g_printerr ("ERROR from element %s: %s\n",
          GST_OBJECT_NAME (msg->src), error->message);
      if (debug)
        g_printerr ("Error details: %s\n", debug);
      g_free (debug);
      g_error_free (error);
      g_main_loop_quit (loop);
      break;
    }
    case GST_MESSAGE_ELEMENT:
    {
      if (gst_nvmessage_is_stream_eos (msg)) {
        guint stream_id;
        if (gst_nvmessage_parse_stream_eos (msg, &stream_id)) {
          g_print ("Got EOS from stream %d\n", stream_id);
          g_mutex_lock (&eos_lock);
          g_eos_list[stream_id] = TRUE;
          g_mutex_unlock (&eos_lock);
        }
      }
      break;
    }
    default:
      break;
  }
  return TRUE;
}

static void
decodebin_child_added (GstChildProxy * child_proxy, GObject * object,
    gchar * name, gpointer user_data)
{
  g_print ("decodebin child added %s\n", name);
  if (g_strrstr (name, "decodebin") == name) {
    g_signal_connect (G_OBJECT (object), "child-added",
        G_CALLBACK (decodebin_child_added), user_data);
  }
  if (g_strstr_len (name, -1, "nvv4l2decoder") == name) {
    g_print ("Seting bufapi_version\n");
    g_object_set (object, "bufapi-version", TRUE, NULL);
  }
}


static void
cb_newpad (GstElement * decodebin, GstPad * pad, gpointer data)
{
  GstCaps *caps = gst_pad_query_caps (pad, NULL);
  const GstStructure *str = gst_caps_get_structure (caps, 0);
  const gchar *name = gst_structure_get_name (str);

  g_print ("decodebin new pad %s\n", name);
  if (!strncmp (name, "video", 5)) {
    gint source_id = (*(gint *) data);
    gchar pad_name[16] = { 0 };
    GstPad *sinkpad = NULL;
    g_snprintf (pad_name, 15, "sink_%u", source_id);
    sinkpad = gst_element_get_request_pad (streammux, pad_name);
    if (gst_pad_link (pad, sinkpad) != GST_PAD_LINK_OK) {
      g_print ("Failed to link decodebin to pipeline\n");
    } else {
      g_print ("Decodebin linked to pipeline\n");
    }
    gst_object_unref (sinkpad);
  }
}

static GstElement *
create_uridecode_bin (guint index, gchar * filename)
{
  GstElement *bin = NULL;
  gchar bin_name[16] = { };

  g_print ("creating uridecodebin for [%s]\n", filename);
  g_source_id_list[index] = index;
  g_snprintf (bin_name, 15, "source-bin-%02d", index);
  bin = gst_element_factory_make ("uridecodebin", bin_name);
  g_object_set (G_OBJECT (bin), "uri", filename, NULL);
  g_signal_connect (G_OBJECT (bin), "pad-added",
      G_CALLBACK (cb_newpad), &g_source_id_list[index]);
  g_signal_connect (G_OBJECT (bin), "child-added",
      G_CALLBACK (decodebin_child_added), &g_source_id_list[index]);
  g_source_enabled[index] = TRUE;

  return bin;
}

static void
stop_release_source (gint source_id)
{
  GstStateChangeReturn state_return;
  gchar pad_name[16];
  GstPad *sinkpad = NULL;
  state_return =
      gst_element_set_state (g_source_bin_list[source_id], GST_STATE_NULL);
  switch (state_return) {
    case GST_STATE_CHANGE_SUCCESS:
      g_print ("STATE CHANGE SUCCESS\n\n");
      g_snprintf (pad_name, 15, "sink_%u", source_id);
      sinkpad = gst_element_get_static_pad (streammux, pad_name);
      gst_pad_send_event (sinkpad, gst_event_new_flush_stop (FALSE));
      gst_element_release_request_pad (streammux, sinkpad);
      g_print ("STATE CHANGE SUCCESS %p\n\n", sinkpad);
      gst_object_unref (sinkpad);
      gst_bin_remove (GST_BIN (pipeline), g_source_bin_list[source_id]);
      source_id--;
      g_num_sources--;
      break;
    case GST_STATE_CHANGE_FAILURE:
      g_print ("STATE CHANGE FAILURE\n\n");
      break;
    case GST_STATE_CHANGE_ASYNC:
      g_print ("STATE CHANGE ASYNC\n\n");
      state_return =
          gst_element_get_state (g_source_bin_list[source_id], NULL, NULL,
          GST_CLOCK_TIME_NONE);
      g_snprintf (pad_name, 15, "sink_%u", source_id);
      sinkpad = gst_element_get_static_pad (streammux, pad_name);
      gst_pad_send_event (sinkpad, gst_event_new_flush_stop (FALSE));
      gst_element_release_request_pad (streammux, sinkpad);
      g_print ("STATE CHANGE ASYNC %p\n\n", sinkpad);
      gst_object_unref (sinkpad);
      gst_bin_remove (GST_BIN (pipeline), g_source_bin_list[source_id]);
      source_id--;
      g_num_sources--;
      break;
    case GST_STATE_CHANGE_NO_PREROLL:
      g_print ("STATE CHANGE NO PREROLL\n\n");
      break;
    default:
      break;
  }


}

static gboolean
delete_sources (gpointer data)
{
  // gint source_id;
  g_mutex_lock (&eos_lock);
  for (gint source_id = 0; source_id < MAX_NUM_SOURCES; source_id++) {
    if (g_eos_list[source_id] == TRUE && g_source_enabled[source_id] == TRUE) {
      g_source_enabled[source_id] = FALSE;
      stop_release_source (source_id);
    }
  }
  g_mutex_unlock (&eos_lock);

  if (g_num_sources == 0) {
    g_main_loop_quit (loop);
    g_print ("All sources Stopped quitting\n");
    return FALSE;
  }

  g_print("\nEnter the ID-Deleted: ");
  gchar temp[2];
  scanf("%[^\n]%*c",temp);
  gint source_id = (gint)*temp -48;

  g_source_enabled[source_id]=FALSE;

  /*Release the source*/
  g_print("Calling stop %d \n",source_id);
  stop_release_source(source_id);

  if (g_num_sources == 0) {
    g_main_loop_quit (loop);
    g_print ("All sources Stopped quitting\n");
    return FALSE;
  }

  return TRUE;
}


static gboolean
add_sources (gpointer data)
{
  gint source_id = g_num_sources; 
  GstElement *source_bin;
  GstStateChangeReturn state_return;

  g_print("\nSelect Initially: Add source < a > | Delete source < d >: \n\n");
  gchar input[2];
  scanf("%[^\n]%*c",input);
  if (input[0] == 'a'){
    g_print("\nEnter the Source_ID: ");
    gchar temp[2];
    scanf("%[^\n]%*c",temp);
    // fflush(stdin);
    source_id = (gint)*temp -48;

    g_source_enabled[source_id]=TRUE;

    g_print("\nEnter the Source_Uri: ");
    // fgets(uri,1000,stdin);
    scanf("%[^\n]%*c",uri);
    g_print("\nSource_ID = %d --- Source_URI = %s \n\n",source_id,uri);

    g_print ("Calling Start %d \n", source_id);
    source_bin = create_uridecode_bin (source_id, uri);
    if (!source_bin) {
      g_printerr ("Failed to create source bin. Exiting.\n");
      return -1;
    }
    g_source_bin_list[source_id] = source_bin;
    gst_bin_add (GST_BIN (pipeline), source_bin);
    state_return =
        gst_element_set_state (g_source_bin_list[source_id], GST_STATE_PLAYING);
    switch (state_return) {
      case GST_STATE_CHANGE_SUCCESS:
        g_print ("STATE CHANGE SUCCESS\n\n");
        source_id++;
        break;
      case GST_STATE_CHANGE_FAILURE:
        g_print ("STATE CHANGE FAILURE\n\n");
        break;
      case GST_STATE_CHANGE_ASYNC:
        g_print ("STATE CHANGE ASYNC\n\n");
        state_return =
            gst_element_get_state (g_source_bin_list[source_id], NULL, NULL,
            GST_CLOCK_TIME_NONE);
        source_id++;
        break;
      case GST_STATE_CHANGE_NO_PREROLL:
        g_print ("STATE CHANGE NO PREROLL\n\n");
        break;
      default:
        break;
    }
    g_num_sources++;
    return TRUE;
  }
  else {
    g_timeout_add_seconds (0, delete_sources, (gpointer) g_source_bin_list);
    return TRUE;
  }
  return TRUE;
}

// static GstRTSPServer *server;
static guint server_count=0;
static GMutex server_cnt_lock;

static gboolean
start_rtsp_streaming (int index, guint rtsp_port_num, guint updsink_port_num,
                      guint64 udp_buffer_size)
{
    GstRTSPMountPoints *mounts;
    GstRTSPMediaFactory *factory;
    GstRTSPServer *server;
    gchar rtsp_out[16]={ };
    char udpsrc_pipeline[512];

    char port_num_Str[64] = { 0 };
    char *encoder_name;


    if (udp_buffer_size == 0)
        udp_buffer_size = 512 * 1024;

    sprintf (udpsrc_pipeline,
             "( udpsrc name=pay0 port=%d buffer-size=524288 caps=\"application/x-rtp, media=video, "
             "clock-rate=90000, encoding-name=H265, payload=96 \" )",
             updsink_port_num);

    sprintf (port_num_Str, "%d", rtsp_port_num);
    
    // g_mutex_lock (&server_cnt_lock);

    server = gst_rtsp_server_new ();
    g_object_set (server, "service", port_num_Str, NULL);

    mounts = gst_rtsp_server_get_mount_points (server);

    factory = gst_rtsp_media_factory_new ();
    gst_rtsp_media_factory_set_launch (factory, udpsrc_pipeline);

    g_snprintf(rtsp_out ,15 ,"/ds-test-%d", index);

    gst_rtsp_mount_points_add_factory (mounts, rtsp_out, factory);

    g_object_unref (mounts);

    gst_rtsp_server_attach (server, NULL);

    // g_mutex_lock (&server_cnt_lock);

    g_print ("\n *** DeepStream: Launched RTSP Streaming at rtsp://192.168.1.23:%d%s ***\n\n",
             rtsp_port_num, rtsp_out);

    return TRUE;
}

static GstRTSPFilterResult
client_filter (GstRTSPServer * server, GstRTSPClient * client,
               gpointer user_data)
{
    return GST_RTSP_FILTER_REMOVE;
}

static void
destroy_sink_bin ()
{
    GstRTSPMountPoints *mounts;
    GstRTSPSessionPool *pool;
    GstRTSPServer *server;
    mounts = gst_rtsp_server_get_mount_points (server);
    for(int i=0; i< MAX_NUM_SOURCES; i++){
      gchar rtsp_out[16]={ };
      g_snprintf(rtsp_out ,15 ,"/ds-test-%d", i);
      gst_rtsp_mount_points_remove_factory (mounts, rtsp_out);
    }
    g_object_unref (mounts);
    gst_rtsp_server_client_filter (server, client_filter, NULL);
    pool = gst_rtsp_server_get_session_pool (server);
    gst_rtsp_session_pool_cleanup (pool);
    g_object_unref (pool);
}


static GstElement 
*create_udpsink_bin (guint index, guint updsink_port_num){

  GstElement *bin = NULL;
  gchar elem_name[50];
  gchar rtppay_name[50];
  GstPad *pad, *ghost_pad;
  gchar bin_name[16] = { };

  g_snprintf (bin_name, 15, "sink-bin-%02d", index);
  bin = gst_bin_new (bin_name);
  
  GstElement *queue = NULL, *cap_filter = NULL, *transform = NULL,*nvosd=NULL, *transform_1=NULL,
  *encoder = NULL,*rtppay = NULL, *codecparse = NULL, *sink = NULL;
  GstCaps *caps = NULL;
  
  queue = gst_element_factory_make (NVDS_ELEM_QUEUE, "queue");
  
  transform = gst_element_factory_make (NVDS_ELEM_VIDEO_CONV, "nvvidconv");

  transform_1 = gst_element_factory_make (NVDS_ELEM_VIDEO_CONV, "nvvidconv_post");

  nvosd = gst_element_factory_make ("nvdsosd", "nv-onscreendisplay");
  
  // g_snprintf (elem_name, sizeof (elem_name), "sink-bin-%02d-capfilter", index);
  cap_filter = gst_element_factory_make (NVDS_ELEM_CAPS_FILTER, "filter");
  caps = gst_caps_from_string("video/x-raw(memory:NVMM), format=I420");
  g_object_set(G_OBJECT(cap_filter), "caps", caps, NULL);

  codecparse = gst_element_factory_make("h265parse", "h265-parser");
  
  encoder = gst_element_factory_make("nvv4l2h265enc", "rtsp-encoder");

  rtppay = gst_element_factory_make("rtph265pay","rtsp-pay");
  
  // g_object_set(G_OBJECT(encoder),"bitrate", 400000, NULL);
  // g_object_set(G_OBJECT(encoder),"iframeinterval", 0, NULL);
  // g_object_set(G_OBJECT(encoder),"gpu-id",0, NULL);
  g_object_set (G_OBJECT (encoder), "preset-level", 1, NULL);
  g_object_set (G_OBJECT (encoder), "insert-sps-pps", 1, NULL);
  g_object_set (G_OBJECT (encoder), "bufapi-version", 1, NULL);

  g_object_set(G_OBJECT(transform),"gpu-id",0, NULL);
  
  sink = gst_element_factory_make ("udpsink", "updsink");
  
  g_object_set(G_OBJECT(sink), "host", "127.0.0.1", "port", updsink_port_num, "async", FALSE, "sync", 1, NULL);
  
  gst_bin_add_many(GST_BIN(bin),queue, transform, nvosd,transform_1, 
                                cap_filter, encoder,rtppay,sink,NULL);

  gst_element_link_many (queue, transform, nvosd,transform_1, 
                                cap_filter, encoder, rtppay,sink, NULL);
  
  pad=gst_element_get_static_pad (queue,"sink");
  ghost_pad=gst_ghost_pad_new ("sink", pad);
  gst_pad_set_active (ghost_pad, TRUE);

  if (!gst_element_add_pad (bin, ghost_pad)){
        g_printerr("Failed to add ghost padd in sink bin \n");
        return NULL;
  }
  start_rtsp_streaming (index, 8554, updsink_port_num,0);
  
  return bin;
}

int main (int argc, char *argv[])
{
  GstBus *bus = NULL;
  guint bus_watch_id;
  guint i, num_sources;
  guint tiler_rows, tiler_columns;
  guint pgie_batch_size;

#ifdef PLATFORM_TEGRA
  GstElement *nvtransform;
#endif

  num_sources = 1;

  /* Standard GStreamer initialization */
  gst_init (&argc, &argv);
  loop = g_main_loop_new (NULL, FALSE);

  g_mutex_init (&eos_lock);

  /* Create gstreamer elements */
  /* Create Pipeline element that will form a connection of other elements */
  pipeline = gst_pipeline_new ("dstest-pipeline");

  /* Use nvinfer to run inferencing on decoder's output,
   * behaviour of inferencing is set through config file */
  streammux = gst_element_factory_make ("nvstreammux", "stream-muxer");
  // nvvidconv = gst_element_factory_make ("nvvideoconvert", "nvvidconv");
  streamdemux = gst_element_factory_make("nvstreamdemux", "demuxer");
  // pgie = gst_element_factory_make ("nvinfer", "primary-nvinference-engine");
  // nvosd = gst_element_factory_make ("nvdsosd", "nv-onscreendisplay");
  /* Use nvtiler to stitch o/p from upstream components */
  // tiler = gst_element_factory_make ("nvmultistreamtiler", "nvtiler");

  if (!pipeline || !streammux || !streamdemux) {
    g_printerr ("One element could not be created. Exiting.\n");
    return -1;
  }
  // g_object_get (G_OBJECT (pgie), "batch-size", &pgie_batch_size, NULL);

  g_object_set (G_OBJECT (pgie), "batch-size", MAX_NUM_SOURCES, NULL);
  
  SET_GPU_ID (streammux, GPU_ID);
  // SET_GPU_ID (pgie, GPU_ID);

  // tiler_rows = (guint) sqrt (num_sources);
  // tiler_columns = (guint) ceil (1.0 * num_sources / tiler_rows);
  // /* we set the osd properties here */
  // g_object_set (G_OBJECT (tiler), "rows", tiler_rows, "columns", tiler_columns,
  //     "width", TILED_OUTPUT_WIDTH, "height", TILED_OUTPUT_HEIGHT, NULL);
  // SET_GPU_ID (tiler, GPU_ID);
  // SET_GPU_ID (nvvidconv, GPU_ID);
  // SET_GPU_ID (nvosd, GPU_ID);
// #ifndef PLATFORM_TEGRA
//   SET_GPU_ID (sink, GPU_ID);
// #endif
  gst_bin_add (GST_BIN (pipeline), streammux);
  g_object_set (G_OBJECT (streammux), "live-source", 1, NULL);
  g_object_set (G_OBJECT (streammux), "batched-push-timeout", 2500000, NULL);
  g_object_set (G_OBJECT (streammux), "batch-size", 10, NULL);
  g_object_set (G_OBJECT (streammux), "nvbuf-memory-type", 3, NULL);
  g_object_set (G_OBJECT (streammux), "width", MUXER_OUTPUT_WIDTH, "height",MUXER_OUTPUT_HEIGHT, NULL);
  SET_GPU_ID (streammux, GPU_ID);

  g_source_bin_list = g_malloc0 (sizeof (GstElement *) * MAX_NUM_SOURCES);
  
  /* Inital create the input source  */
  g_print("\nEnter the Source_ID: ");
  gchar temp[2];
  scanf("%[^\n]%*c",temp);
  gint source_id = (gint)*temp -48;
  g_print("\nEnter the Source_Uri: ");
  scanf("%[^\n]%*c",uri);
 
  g_print("\nSource_ID = %d --- Source_URI = %s \n\n",source_id,uri);

  GstElement *source_bin = create_uridecode_bin (source_id,uri);
  if (!source_bin) {
    g_printerr ("Failed to create source bin. Exiting.\n");
    return -1;
  }
  g_source_bin_list[source_id] = source_bin;
  gst_bin_add (GST_BIN (pipeline), source_bin);
  
  g_num_sources = num_sources;

  gst_bin_add (GST_BIN(pipeline), streamdemux);
  /*Inital sink_bin for demuxer*/
  for (int i=0; i<MAX_NUM_SOURCES;i++){
    GstPad *sinkpad, *srcpad;
    gchar pad_nam[16]={};
    // guint udpsink_last = updsink_port + source_id;
    // start_rtsp_streaming (i, 8554, updsink_port+i, 0);
    
    GstElement *sink_bin=create_udpsink_bin(i, updsink_port+i);
    
    if (!sink_bin){
      g_printerr("Failed to create sink bin. Exiting \n");
      return -1;
    }
    gst_bin_add(GST_BIN(pipeline), sink_bin);

    g_snprintf(pad_nam, 15, "src_%u", i); 

    srcpad = gst_element_get_request_pad (streamdemux, pad_nam);
    if (!srcpad) {
      g_printerr ("Streamdemux request source pad failed. Exiting.\n");
      return -1;
    }

    sinkpad = gst_element_get_static_pad (sink_bin, "sink");
    if (!sinkpad) {
      g_printerr ("Failed to get sink pad of sink bin. Exiting.\n");
      return -1;
    }

    if (gst_pad_link (srcpad, sinkpad) != GST_PAD_LINK_OK) {
      g_printerr ("Failed to link source bin to stream demuxer. Exiting.\n");
      return -1;
    }
    
    gst_object_unref (srcpad);
    gst_object_unref (sinkpad);
  }

  // gst_bin_add_many (GST_BIN (pipeline), nvvidconv, nvosd, NULL);

#ifdef PLATFORM_TEGRA
  gst_bin_add (GST_BIN (pipeline), nvtransform);
#endif

  /* we link the elements together */
 
#ifdef PLATFORM_TEGRA
  if (!gst_element_link_many (streammux,  
            nvtransform, streamdemux, NULL)) {
    g_printerr ("Elements could not be linked 1. Exiting.\n");
    return -1;
  }
#else
  if (!gst_element_link_many (streammux, 
            streamdemux, NULL)) {
    g_printerr ("Elements could not be linked 2. Exiting.\n");
    return -1;
  }
#endif


  /* we add a message handler */
  bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
  bus_watch_id = gst_bus_add_watch (bus, bus_call, loop);
  gst_object_unref (bus);

  /* Set the pipeline to "playing" state */
  gst_element_set_state (pipeline, GST_STATE_PAUSED);

  if (gst_element_set_state (pipeline,
          GST_STATE_PLAYING) == GST_STATE_CHANGE_FAILURE) {
    g_printerr ("Failed to set pipeline to playing. Exiting.\n");
    return -1;
  }

  /* Wait till pipeline encounters an error or EOS */
  g_print ("Running...\n");
  g_timeout_add_seconds (20, add_sources, (gpointer) g_source_bin_list);

  // while(1){
  //   g_print("\nSelect Initially: Add source < a > | Delete source < d >: \n\n");
  //   gchar input[2];
  //   scanf("%[^\n]%*c",input);
  //   if (input[0] == 'a'){
  //     bool add=add_sources();
  //     if (!add){
  //       break;
  //     }
  //   }
  //   else if(input[0]=='d'){
  //     bool del=delete_sources();
  //     if (!del){
  //       break;
  //     }
  //   }
  // }
  
  g_main_loop_run (loop);

  /* Out of the main loop, clean up nicely */
  g_print ("Returned, stopping playback\n");
  gst_element_set_state (pipeline, GST_STATE_NULL);
  g_print ("Deleting pipeline\n");
  destroy_sink_bin();
  gst_object_unref (GST_OBJECT (pipeline));
  g_source_remove (bus_watch_id);
  g_main_loop_unref (loop);
  g_free (g_source_bin_list);
  g_free (uri);
  g_mutex_clear (&eos_lock);
  return 0;
}