Code:

#include <gst/gst.h>
#include <gst/rtsp-server/rtsp-server.h>

#include <math.h>
#include <stdio.h>
#include <stdlib.h>
#include <mysql.h>
#include <string.h>
#include <time.h>

#include <signal.h>

int do_exit = FALSE;
void send_eos(int);

/* Structure to contain all our information, so we can pass it to callbacks */
typedef struct _CustomData {
  GstElement *pipeline;
  GstElement *source;
  GstElement *tee;
  GstElement *decodebin;
  GstElement *convert1;
  GstElement *deinterlacer;
  GstElement *queue;
  GstElement *facedetector;
  GstElement *faceblur;
  GstElement *motiondetect;
  GstElement *convert2;
  GstElement *scaler;
  GstElement *balancer;
  GstElement *videorater;
  GstElement *convert3;
  GstElement *convert4;
  GstElement *capsfilter;
  GstElement *videoencoder;
  GstElement *capsfilter2;
  GstElement *rtpencoder;
  GstElement *capsfilter3;
  GstElement *capsfilter4;
  GstElement *sink;
  GstElement *timeoverlay;

  //subpipe 2
  GstElement *origrtpdecoder;
  GstElement *origrtpencoder;
  GstElement *origsink;
} CustomData;

/* Handler for the pad-added signal */
static void pad_added_handler (GstElement *src, GstPad *pad, CustomData *data);

static void feature_detect_handler(GstElement *e, time_t *t, CustomData *data);
static void motion_start_handler(GstElement *e, time_t *t, CustomData *data);
static void motion_stop_handler(GstElement *e, time_t *t, CustomData *data);

/* rtsp server thread */
void *start_rtsp_stream(void *param);

CustomData data;
MYSQL *MySQLConRet =NULL, *MySQLConn = NULL;

gint orig_video_width=0, orig_video_height=0;

const char *hostname = "localhost",
		*user = "root",
		*pass = "root",
		*dbname = "lensbrick";

int main(int argc, char *argv[]) {
  GstBus *bus = NULL;
  GstMessage *msg;
  GstStateChangeReturn ret;
  gboolean terminate = FALSE;
  gboolean add_timestamp = FALSE;

  time_t time_diff;
  time_t pipeline_retries[5] = {0,0,0,0,0};
  int next_slot = 0, most_recent_slot = -1;

  /* Initialize GStreamer */
  gst_init (&argc, &argv);

  if(argc>1 && ( strncmp("-h",argv[1],3) == 0 || strncmp("--help",argv[1],7) == 0 ) ) {
	g_print("Usage: rtsp-server <rtsp-feed-url> [add_time] [roi of motion detect]");
	exit(0);
  }

  if(argc > 2) {
    add_timestamp = strncmp("add_time", argv[2], 9);
  }

  MySQLConn = mysql_init(NULL);
  MySQLConRet = mysql_real_connect(MySQLConn,
				hostname,
				user,
				pass,
				dbname,
				0,
				NULL,
				0);

  if(MySQLConRet == NULL) {
	g_print("Mysql Error: %s\n",mysql_error(MySQLConn));
  }

  //todo: Create multiple threads - one for multiple pipeline maintenance, and
	//1 for rtsp streaming server and 1 for reload cmd handling

  /* Create the elements */
  data.source = gst_element_factory_make ("rtspsrc", "source");
  data.tee = gst_element_factory_make ("tee", "tee");
  data.decodebin = gst_element_factory_make ("decodebin", "decodebin");
  data.convert1 = gst_element_factory_make ("videoconvert", "convert1");
  data.deinterlacer = gst_element_factory_make ("deinterlace", "deinterlacer");
  data.queue = gst_element_factory_make ("queue2", "queue");
  data.facedetector = gst_element_factory_make ("facedetect", "facedetector");
  data.faceblur = gst_element_factory_make ("faceblur", "faceblur");
  data.motiondetect = gst_element_factory_make ("motiondetect", "motiondetect");
  data.convert2 = gst_element_factory_make ("videoconvert", "convert2");
  data.scaler = gst_element_factory_make ("videoscale", "scaler");
  data.balancer = gst_element_factory_make ("videobalance", "balancer");
  data.videorater = gst_element_factory_make ("videorate", "videorater");
  data.convert3 = gst_element_factory_make ("videoconvert", "convert3");
  data.capsfilter = gst_element_factory_make ("capsfilter", "capsfilter");
  data.videoencoder = gst_element_factory_make ("x264enc", "videoencoder");
  data.rtpencoder = gst_element_factory_make ("rtph264pay", "rtpencoder");
  data.capsfilter2 = gst_element_factory_make ("capsfilter", "capsfilter2");
  data.capsfilter3 = gst_element_factory_make ("capsfilter", "capsfilter3");
  data.capsfilter4 = gst_element_factory_make ("capsfilter", "capsfilter4");
  data.convert4 = gst_element_factory_make ("videoconvert", "convert4");
  data.sink = gst_element_factory_make ("udpsink", "processed_sink");
  data.origsink = gst_element_factory_make ("udpsink", "original_sink");
  data.origrtpdecoder = gst_element_factory_make ("rtph264depay", "orig_rtpdecoder");
  data.origrtpencoder = gst_element_factory_make ("rtph264pay", "orig_rtpencoder");
  data.timeoverlay = gst_element_factory_make ("clockoverlay", "timer");
  //data.sink = gst_element_factory_make ("xvimagesink", "sink");

  /* Create the empty pipeline */
  data.pipeline = gst_pipeline_new ("process-pipeline");

  if (!data.pipeline || !data.source || !data.capsfilter ||
	!data.decodebin || !data.convert1 || !data.deinterlacer
	|| !data.queue || !data.facedetector
	|| !data.convert2  || !data.capsfilter3
	|| !data.scaler || !data.balancer || !data.timeoverlay
	|| !data.videorater || !data.convert3  || !data.tee || !data.origsink
	|| !data.faceblur || !data.convert4 || !data.capsfilter4 || !data.motiondetect
	|| !data.videoencoder || !data.rtpencoder || !data.capsfilter2
	|| !data.sink || !data.origrtpdecoder || !data.origrtpencoder) {
    g_printerr ("Not all elements could be created.\n");
    return -1;
  }

  /*Set time overlay text props*/
  g_object_set (G_OBJECT(data.timeoverlay), "font-desc", "Arial bold 35",
	 "time-format", "%d-%m-%Y %H:%M:%S", NULL);

  /*Set queue props*/
  g_object_set (G_OBJECT(data.queue), "max-size-buffers", 3, NULL);

  /*Set face detector profile*/
  g_object_set (G_OBJECT(data.facedetector), "profile", "/usr/local/share/OpenCV/haarcascades/haarcascade_frontalface_default.xml", "updates", 1, NULL);

  /*Set face detector profile*/
  g_object_set (G_OBJECT(data.faceblur), "profile", "/usr/local/share/OpenCV/haarcascades/haarcascade_frontalface_default.xml", NULL);

  /*Set motion detector ROI*/
  g_object_set (G_OBJECT(data.motiondetect), "roi", (argc>3?argv[3]:""), "silent", TRUE, NULL);

  /*Set udp port sink for original video*/
  g_object_set (G_OBJECT(data.origsink), "port", 5001, NULL);

  /*Set udp port sink*/
  g_object_set (G_OBJECT(data.sink), "port", 5000, NULL);

  /*Set rtpencoder config-interval*/
  g_object_set (G_OBJECT(data.origrtpencoder), "config-interval", 1, NULL);

  /* Set the capsfilter property */

  g_object_set (G_OBJECT(data.capsfilter2), "caps", gst_caps_new_simple("video/x-raw",
									"framerate",GST_TYPE_FRACTION,8,1,
									//"clock-rate",G_TYPE_INT, 90000,
									NULL), NULL );
  g_object_set (G_OBJECT(data.capsfilter3), "caps", gst_caps_new_simple("video/x-raw",
									"format",G_TYPE_STRING,"RGB",
									NULL), NULL );
  g_object_set (G_OBJECT(data.capsfilter4), "caps", gst_caps_new_simple("video/x-raw",
									"format",G_TYPE_STRING,"I420",
									NULL), NULL );

  /* Build the pipeline. Note that we are NOT linking the source at this
   * point. We will do it later. */
  gst_bin_add_many (GST_BIN (data.pipeline), data.source, data.capsfilter ,
	data.decodebin , data.convert1 , data.deinterlacer, data.timeoverlay,
	data.queue, data.facedetector, data.convert2, data.capsfilter2,
	data.capsfilter3, data.faceblur, data.convert4, data.capsfilter4,
	data.motiondetect, data.tee, data.scaler, data.balancer,
	data.videorater, data.convert3, data.videoencoder, data.rtpencoder,
	data.sink, NULL);
  if (!gst_element_link_many (
				data.convert1,
				data.deinterlacer,
				data.queue,
				data.convert2,
				data.scaler,
				data.capsfilter,
				data.balancer,
				data.videorater,
				data.capsfilter2,
				data.convert3,
				data.capsfilter3,
				//data.facedetector,
				//data.faceblur,
				data.motiondetect,
				data.timeoverlay,
				data.convert4,
				data.capsfilter4,
				data.videoencoder,
				data.rtpencoder,
				data.sink,
				NULL )) {
    g_printerr ("Elements could not be linked.\n");
    gst_object_unref (data.pipeline);
    return -1;
  }

  /* Set the URI to play */
  g_object_set (data.source, "location", argv[1], NULL);

  /* Connect to the pad-added signal */
  g_signal_connect (data.source, "pad-added", G_CALLBACK (pad_added_handler), &data);
  g_signal_connect (data.decodebin, "pad-added", G_CALLBACK (pad_added_handler), &data);
  g_signal_connect (data.motiondetect, "motion-start", G_CALLBACK (motion_start_handler), &data);
  g_signal_connect (data.motiondetect, "motion-stop", G_CALLBACK (motion_stop_handler), &data);

restart_pipeline:

  pipeline_retries[next_slot] = time(NULL);
  g_print("Attempt/Retry %d at %ld\n", (next_slot+1), pipeline_retries[next_slot]);
  next_slot = (next_slot+1) % 5;
  terminate = FALSE;

  /* Start playing */
  ret = gst_element_set_state (data.pipeline, GST_STATE_PLAYING);
  if (ret == GST_STATE_CHANGE_FAILURE) {
    g_printerr ("Unable to set the pipeline to the playing state.\n");
    gst_object_unref (data.pipeline);
    return -1;
  }

  g_print("Return value of setting pipeline to playing state = %d\n", ret);

  signal(SIGINT,send_eos);

  /* Listen to the bus */
  if(!bus) {
	  bus = gst_element_get_bus (data.pipeline);
  }

  do {
    msg = gst_bus_timed_pop_filtered (bus, GST_CLOCK_TIME_NONE,
        GST_MESSAGE_STATE_CHANGED | GST_MESSAGE_ERROR | GST_MESSAGE_EOS);
    //g_print("Msg received, %p\n", msg);

    /* Parse message */
    if (msg != NULL) {
      GError *err;
      gchar *debug_info;

      switch (GST_MESSAGE_TYPE (msg)) {
        case GST_MESSAGE_ERROR:
          gst_message_parse_error (msg, &err, &debug_info);
          g_printerr ("Error received from element %s: %s\n", GST_OBJECT_NAME (msg->src), err->message);
          g_printerr ("Debugging information: %s\n", debug_info ? debug_info : "none");
          g_clear_error (&err);
          g_free (debug_info);
          terminate = TRUE;
          break;
        case GST_MESSAGE_EOS:
          g_print ("End-Of-Stream reached.\n");
          terminate = TRUE;
          break;
        case GST_MESSAGE_STATE_CHANGED:
          /* We are only interested in state-changed messages from the pipeline */
          if (GST_MESSAGE_SRC (msg) == GST_OBJECT (data.pipeline)) {
            GstState old_state, new_state, pending_state;
            gst_message_parse_state_changed (msg, &old_state, &new_state, &pending_state);
            g_print ("Pipeline state changed from %s to %s:\n",
                gst_element_state_get_name (old_state), gst_element_state_get_name (new_state));

	    g_print("\nSink Caps is %s\n",
		gst_caps_to_string(gst_pad_get_current_caps(gst_element_get_static_pad(data.sink,"sink"))));
	    g_print("\nOriginal video Sink Caps is %s\n",
		gst_caps_to_string(gst_pad_get_current_caps(gst_element_get_static_pad(data.origsink,"sink"))));
          }
          break;
        default:
          /* We should not reach here */
          g_printerr ("Unexpected message received.\n");
          break;
      }
      gst_message_unref (msg);
    }
  } while (!terminate);

  //todo: check if it is a transient error - if not, then should exit

  //if transient error
  gst_element_set_state (data.pipeline, GST_STATE_NULL);
  if(do_exit) {
	goto terminate_process;
  }

  //check if number of retries exceeded 5 within 90 seconds
  //todo: make retry limit of 5, sleep of 10 seconds for next retry
	// and time limit of 90 seconds configurable
  if(pipeline_retries[next_slot] != 0 ) {
        //non-zero next slot means we have done >=5 retries
	most_recent_slot = next_slot - 1;
	if(most_recent_slot == -1) {
		most_recent_slot = 4;
	}

	//oldest slot = next_slot, so do most_recent_slot - next_slot
	g_print("most recent_attempt %d = %ld\n",most_recent_slot,pipeline_retries[most_recent_slot]);
	g_print("old attempt %d = %ld\n",next_slot,pipeline_retries[next_slot]);
	time_diff = time(NULL) - pipeline_retries[next_slot];
	if(time_diff < 60) {
		//within a span of 60 seconds, 5 retries have happened
		g_print("Terminating process\n");
		goto terminate_process;
	}
  }

  g_print("Pipeline set to NULL state and sleeping for 10 seconds\n");
  sleep(10); //todo: to be made configurable
  g_print("Pipeline restarting\n");
  goto restart_pipeline;

terminate_process:

  /* Free resources */
  gst_object_unref (bus);
  //gst_element_set_state (data.pipeline, GST_STATE_NULL);
  gst_object_unref (data.pipeline);
  mysql_close(MySQLConn);
  return 0;
}

/* This function will be called by the pad-added signal */
static void pad_added_handler (GstElement *src, GstPad *new_pad, CustomData *data) {

  GstPad *sink_pad = NULL;
  GstPadLinkReturn ret;
  GstCaps *new_pad_caps = NULL;
  GstStructure *new_pad_struct = NULL;
  const gchar *new_pad_type = NULL,
		*new_pad_media = NULL,
		*new_pad_encoding = NULL;
  pthread_t thread_id;
  int thread_status;

  if(src == data->source) {

      sink_pad = gst_element_get_static_pad (data->tee, "sink");
      g_print ("Received new pad '%s' from '%s':\n",
	GST_PAD_NAME (new_pad), GST_ELEMENT_NAME (src));

      /* If our converter is already linked, we have nothing to do here */
      if (gst_pad_is_linked (sink_pad)) {
        g_print ("  We are already linked. Ignoring.\n");
        goto exit;
      }

      /* Check the new pad's type */
      new_pad_caps = gst_pad_query_caps (new_pad,NULL);
      new_pad_struct = gst_caps_get_structure (new_pad_caps, 0);
      new_pad_type = gst_structure_get_name (new_pad_struct);
      new_pad_media = gst_structure_get_string(new_pad_struct,"media");
      new_pad_encoding = gst_structure_get_string(new_pad_struct,"encoding-name");

      g_print(" Caps is = '%s'",gst_caps_to_string(new_pad_caps));

      if (!g_str_has_prefix (new_pad_type, "application/x-rtp")
    	|| !g_str_has_prefix (new_pad_media, "video")) {
        g_print ("  It has type '%s' which is not  video RTP. Ignoring.\n", new_pad_type);
        goto exit;
      }

      /* Attempt the link */
      ret = gst_pad_link (new_pad, sink_pad);
      if (GST_PAD_LINK_FAILED (ret)) {
        g_print ("  Type is '%s' but link failed.\n", new_pad_type);
	//todo: properly exit instead of calling exit fn abruptly
	exit(-1);
      } else {
        g_print ("  Link succeeded (type '%s').\n", new_pad_type);
      }

      if(!gst_element_link(data->tee, data->decodebin)) {
	g_print("Linking processing branch pipe to tee failed\n");
	//todo: properly exit instead of calling exit fn abruptly
	exit(-1);
      }

      gst_element_set_state( data->origrtpdecoder, GST_STATE_PAUSED );
      gst_element_set_state( data->origrtpencoder, GST_STATE_PAUSED );
      gst_element_set_state( data->origsink, GST_STATE_PAUSED );

      gst_bin_add_many(GST_BIN(data->pipeline), data->origrtpencoder, data->origrtpdecoder, data->origsink, NULL);
      if(!gst_element_link_many(data->tee, data->origrtpdecoder, data->origrtpencoder, data->origsink, NULL)) {
	g_print("Setting subpipe to stream original video failed. Couldnt link all elements in pipe\n");
	//todo: properly exit instead of calling exit fn abruptly
	exit(-1);
      }

      gst_element_set_state( data->origrtpdecoder, GST_STATE_PLAYING );
      gst_element_set_state( data->origrtpencoder, GST_STATE_PLAYING );
      gst_element_set_state( data->origsink, GST_STATE_PLAYING );

  } else if( src == data->decodebin) {

      sink_pad = gst_element_get_static_pad (data->convert1, "sink");

      g_print ("Received new pad '%s' from '%s':\n", GST_PAD_NAME (new_pad), GST_ELEMENT_NAME (src));

      /* If our converter is already linked, we have nothing to do here */
      if (gst_pad_is_linked (sink_pad)) {
        g_print ("  We are already linked. Ignoring.\n");
        goto exit;
      }

      new_pad_caps = gst_pad_query_caps (new_pad,NULL);
      new_pad_struct = gst_caps_get_structure (new_pad_caps, 0);
      gst_structure_get_int(new_pad_struct,"width", &orig_video_width);
      gst_structure_get_int(new_pad_struct,"height", &orig_video_height);
      g_print("%d = width, %d = height\n", orig_video_width, orig_video_height);

      g_print(" Caps is = '%s'",gst_caps_to_string(new_pad_caps));

      /* Set the capsfilter property */
      if(orig_video_height > 400 && orig_video_width > 400) {

    	  g_object_set (G_OBJECT(data->capsfilter), "caps", gst_caps_new_simple("video/x-raw",
    									"width",G_TYPE_INT,orig_video_width/4,
    									"height",G_TYPE_INT,orig_video_height/4,
    									NULL), NULL );
    	  g_print("scaled: %d = width, %d = height\n", orig_video_width/4, orig_video_height/4);
      }

      /* Attempt the link */
      ret = gst_pad_link (new_pad, sink_pad);
      if (GST_PAD_LINK_FAILED (ret)) {
        g_print ("  link failed.\n");
	//todo: properly exit instead of abrupt exit
	exit(-1);
      } else {
        g_print ("  Link succeeded .\n");
        thread_status = pthread_create( &thread_id, NULL, start_rtsp_stream, NULL);

      }
   }

exit:
  /* Unreference the new pad's caps, if we got them */
  if (new_pad_caps != NULL)
    gst_caps_unref (new_pad_caps);

  /* Unreference the sink pad */
  gst_object_unref (sink_pad);
}


void *start_rtsp_stream(void *param) {

  GMainLoop *loop;
  GstRTSPServer *server;
  GstRTSPMountPoints *mounts;
  GstRTSPMediaFactory *factory, *origfactory;

  loop = g_main_loop_new (NULL, FALSE);

  /* create a server instance */
  server = gst_rtsp_server_new ();
  gst_rtsp_server_set_address(server,"0.0.0.0");

  /* get the mount points for this server, every server has a default object
   * that be used to map uri mount points to media factories */
  mounts = gst_rtsp_server_get_mount_points (server);

  /* make a media factory for a test stream. The default media factory can use
   * gst-launch syntax to create pipelines.
   * any launch line works as long as it contains elements named pay%d. Each
   * element with pay%d names will be a stream */
  factory = gst_rtsp_media_factory_new ();
  gst_rtsp_media_factory_set_launch (factory, " ( "
	" udpsrc port=5000 ! "
	" capsfilter caps=\"application/x-rtp,clock-rate=90000,payload=96,encoding-name=H264,media=video\" name=pay0 "
	" ) " );

  /* attach the test factory to the /test url */
  gst_rtsp_mount_points_add_factory (mounts, "/processed", factory);

  origfactory = gst_rtsp_media_factory_new ();
  gst_rtsp_media_factory_set_launch (origfactory, " ( "
	" udpsrc port=5001 ! "
	" capsfilter caps=\"application/x-rtp,clock-rate=90000,payload=96,encoding-name=H264,media=video\" name=pay0 "
	" ) " );
  /* attach the original video factory to the /original url */
  gst_rtsp_mount_points_add_factory (mounts, "/original", origfactory);

  /* don't need the ref to the mapper anymore */
  g_object_unref (mounts);

  /* attach the server to the default maincontext */
  gst_rtsp_server_attach (server, NULL);

  /* start serving */
  g_print ("processed/debug stream ready at rtsp://:8554/processed\n");
  g_print ("original stream ready at rtsp://:8554/original\n");
  g_main_loop_run (loop);

  return 0;

}

static void event_handler(GstElement *e, time_t *t, CustomData *data, gchar *event_name) {
	struct tm *tmp;
	int status = 0;
	char sql[200] = "insert into event_notifications values('";
	gchar time_str[100];

	g_print("Hello, captured the event. Timestamp = %ld, ", *t);
	tmp = localtime(t);
	strftime(time_str, 100, "%Y-%m-%d %H:%M:%S", tmp);
	g_print("===%s\n",time_str);

		//todo: call mysql-real-escape-string()
		strncat(sql,event_name,200);
		//todo: replace local4 with actual stream name
		strncat(sql,"','local4','",200);
		strncat(sql,time_str,200);
		strncat(sql,"','{}');",200);

	g_print("SQL: %s\n",sql);
	status = mysql_query( MySQLConn, sql);
	if(status) {
		g_print("SQL Error: %s\n", mysql_error(MySQLConn));
	}
}

static void motion_start_handler(GstElement *e, time_t *t, CustomData *data) {

	event_handler(e, t, data, "motion-start");
}

static void motion_stop_handler(GstElement *e, time_t *t, CustomData *data) {

	event_handler(e, t, data, "motion-stop");
}

void send_eos(int sig) {

	do_exit = TRUE;
	gst_element_send_event(data.pipeline,gst_event_new_eos());
}