#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>

#include <pthread.h>

#include <gst/gst.h>


#define SERVER_PORT 30001

#define BROADCAST_START_CMD "startcam"
#define BROADCAST_STOP_CMD "stopbroadcast"

#define CAMERA_RTSP_URL "rtsp://asdasdasdasdasasdasdasdasdasd"
#define BROADCAST_RTMP_URL "rtmp://asdasdasdasd live=1"
#define AUDIO_RTSP_URL "rtsp://asdasdasdasdasdasdasdasdasdasdadasdasdadasdasdas"

#define NUM_OF_CAMERA 5

static pthread_t pthread_server_thread;
static pthread_t pthread_gst_pipeline_thread;
static pthread_mutex_t lock_cam_switch;

static GstElement *pipeline = NULL;

struct video_pipeline_type {
	GstElement *rtspsrc;
	GstElement *rtspsrc_queue;
	GstElement *rtph264depay;
	GstElement *shift;
	/*GstElement *sink_sync;*/
};

struct video_pipeline_type video_pipeline[NUM_OF_CAMERA];

struct rtspsrc_pad_added_camindex_type
{
    int camindex;
} rtspsrc_pad_added_camindex[NUM_OF_CAMERA];

static GstElement *rtsp_video_switch = NULL;
static GstPad *rtsp_video_switch_srcpad;
static GMainLoop *loop = NULL;

static char rtspurl[NUM_OF_CAMERA][255];

static sockfd = -1;

static int old_cam_index = -1;
static int new_cam_index = -1;


static void on_pad_added_audio (GstElement *element, GstPad *pad, void *data)
{
    GstPad *sinkpad;
    GstElement * element1 = (GstElement *) data;
    sinkpad = gst_element_get_static_pad (element1, "sink");
    gst_pad_link (pad, sinkpad);
    gst_object_unref (sinkpad);
}

static GstPadProbeReturn
pad_probe_cb (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
	GstBuffer *buffer;
	GstBufferFlags flags;
	buffer = gst_pad_probe_info_get_buffer(info);
	flags = GST_BUFFER_FLAGS(buffer);
	if (flags & GST_BUFFER_FLAG_DELTA_UNIT) {
		return GST_PAD_PROBE_DROP;
	} else
		return GST_PAD_PROBE_REMOVE;
}

static void switch_cam(int dontdrop) {
	gint nb_sources;
	GstPad *active_pad;
	g_object_get (G_OBJECT (rtsp_video_switch), "n-pads", &nb_sources, NULL);
	g_object_get (G_OBJECT (rtsp_video_switch), "active-pad", &active_pad, NULL);
	fprintf(stderr, "callback thread: current number of sources : %d, active source %s\n", nb_sources, gst_pad_get_name (active_pad));
    fprintf(stderr, "callback thread: Line %d: camera is switched from cam%d to cam%d\n", __LINE__, old_cam_index, new_cam_index);
    GstPad *sinkpad;
    char sinkpad_name[255];
    snprintf(sinkpad_name, sizeof(sinkpad_name), "sink_%u", new_cam_index-1);
    sinkpad = gst_element_get_static_pad (rtsp_video_switch, sinkpad_name);
    g_object_set (G_OBJECT (rtsp_video_switch), "active-pad", sinkpad, NULL);
    while(GST_STATE (GST_ELEMENT (rtsp_video_switch)) != GST_STATE_PLAYING)
    {
		usleep(200);
        fprintf(stderr, "callback thread: waiting for rtsp_video_switch to change to playing state\n");
    }
    if (!dontdrop)
		gst_pad_add_probe (rtsp_video_switch_srcpad, GST_PAD_PROBE_TYPE_BUFFER, pad_probe_cb, rtsp_video_switch, NULL);

    gst_object_unref (sinkpad);
}

static void on_pad_added_video (GstElement *element, GstPad *pad, void *data)
{
    GstPad *sinkpad;
    int camindex = ((struct rtspsrc_pad_added_camindex_type *) data)->camindex;
    char sinkpad_name[255];
    fprintf(stderr, "rtspsrc on_pad_added_video signal received for camindex %d\n", camindex);
    
    sinkpad = gst_element_get_static_pad(video_pipeline[camindex].rtspsrc_queue, "sink");
    if(sinkpad == NULL) {
		fprintf(stderr, "unable to get sinkpad from queue element for camindex %d\n", camindex);
		exit(1);
	}
	if(gst_pad_is_linked(pad)) {
		fprintf(stderr, "Warning! received on_pad_added signal for rtspsrc which is already linked for camindex %d\n", camindex);
		gst_object_unref (sinkpad);
		return;
	} else if(gst_pad_is_linked(sinkpad)) {
		fprintf(stderr, "Warning! received on_pad_added signal for rtspsrc_queue which is already linked for camindex %d\n", camindex);
		gst_object_unref (sinkpad);
		return;
	} else {
		if(gst_pad_link (pad, sinkpad) != GST_PAD_LINK_OK)
		{
			fprintf(stderr, "unable to link rtspsrc with rtspsrc_queue for camindex %d\n", camindex);
			exit(1);
		}
    }
    gst_object_unref (sinkpad);
    if (new_cam_index == camindex+1)
		switch_cam(1);
	fprintf(stderr, "rtspsrc reconnected successfully for camindex %d\n", camindex);
}

static gboolean camswitch_timeout_cb (gpointer user_data)
{
    pthread_mutex_lock(&lock_cam_switch);
    if (old_cam_index != new_cam_index) {
		switch_cam(0);
        old_cam_index = new_cam_index;
    }
    pthread_mutex_unlock(&lock_cam_switch);
    return TRUE;
}

static gboolean bus_call (GstBus *bus,GstMessage *msg, gpointer data)
{
    GMainLoop *loop = (GMainLoop *) data;

    switch (GST_MESSAGE_TYPE (msg))
    {
    case GST_MESSAGE_LATENCY:
    {
        if(gst_bin_recalculate_latency(GST_BIN(pipeline)) == TRUE)
            printf("succesfully redistributed latency on pipeline\n");
        else
            printf("failed to redistribute latency\n");
        break;
    }
    case GST_MESSAGE_NEW_CLOCK:
    {
        printf("a new clock is selected for pipeline\n");
        break;
    }
    case GST_MESSAGE_CLOCK_PROVIDE:
    {
        printf("clock provider changes  for a element for pipeline\n");
        break;
    }
    case GST_MESSAGE_CLOCK_LOST:
    {
        printf("current clock is unusable for pipeline and it will select new clock on next state change to playing \n");
        break;
    }
    case GST_MESSAGE_EOS:
    {
        g_print ("GST_MESSAGE_EOS: Stream Ends\n");
        g_main_loop_quit (loop);
        break;
    }
    case GST_MESSAGE_ERROR:
    {
        gchar  *debug;
        GError *error;

        gst_message_parse_error (msg, &error, &debug);
        g_print("debug from errorlog: %s\n", debug);
        g_free (debug);

        g_printerr ("Error from errorlog: %s\n", error->message);
        g_error_free (error);
        exit(1);
        break;
    }
    case GST_MESSAGE_INFO:
    {
        gchar  *debug;
        GError *error;

        gst_message_parse_info (msg, &error, &debug);

        g_print("debug from infolog: %s\n", debug);
        g_printerr ("Error from infolog: %s\n", error->message);
        g_free (debug);
        g_error_free (error);

        break;
    }
    default:
        break;
    }

    return TRUE;
}

void * gst_pipeline_start (void *data)
{
    GstElement *audio_rtspsrc, *audio_queue1, *audio_rtpmpadepay, *audio_mpegaudioparse, *audio_mad, *audio_audioconvert, /* *audio_queue2, */ *audio_voaacenc, *audio_aacparse /*,  *audio_queue3 */;

	GstElement *segment, *video_queue;
    GstElement *h264parse, *flvmux, /* *mixing_queue, */ *rtmpsink;

    GError *error = NULL;
    GstBus *bus;
    int i = 0;
    int camindex = *((int *) data);
    free(data);
    gst_init(NULL, NULL);

    pipeline = gst_pipeline_new ("broadcast pipeline");
    g_assert (pipeline);

    //construct audio pipeline
    audio_rtspsrc = gst_element_factory_make ("rtspsrc", "audio_rtspsrc");
    audio_queue1 = gst_element_factory_make ("queue", "audio_queue1");
    audio_rtpmpadepay = gst_element_factory_make ("rtpmpadepay", "audio_rtpmpadepay");
    audio_mpegaudioparse = gst_element_factory_make ("mpegaudioparse", "audio_mpegaudioparse");
    audio_mad = gst_element_factory_make ("mad", "audio_mad");
    audio_audioconvert = gst_element_factory_make ("audioconvert", "audio_audioconvert");
    //audio_queue2 = gst_element_factory_make ("queue", "audio_queue2");
    audio_voaacenc = gst_element_factory_make ("voaacenc", "audio_voaacenc");
    audio_aacparse = gst_element_factory_make ("aacparse", "audio_aacparse");
    //audio_queue3 = gst_element_factory_make ("queue", "audio_queue");

    if (!audio_rtspsrc || !audio_queue1 || !audio_rtpmpadepay || !audio_mpegaudioparse || !audio_mad || !audio_audioconvert || /* !audio_queue2 || */ !audio_voaacenc || !audio_aacparse /* ||  !audio_queue3 */)
    {
        g_printerr ("One of the elements in audio pipeline wasn't created... Exiting\n");
        exit(1);
    }
    g_object_set (G_OBJECT (audio_rtspsrc), "location", AUDIO_RTSP_URL, "latency", 2000, "do-retransmission", TRUE, NULL);
    //g_object_set (G_OBJECT (audio_queue1), "leaky", TRUE, "flush-on-eos", TRUE, NULL);
    g_object_set (G_OBJECT (audio_voaacenc), "bitrate", 64000, NULL);

    //construct video pipeline
    rtsp_video_switch = gst_element_factory_make ("input-selector", "video_switch");
    segment = gst_element_factory_make ("identity", "identity-segment");
    g_object_set (G_OBJECT (segment), "silent", TRUE, "single-segment", TRUE, NULL);
    video_queue = gst_element_factory_make ("queue", "video_queue");
    //g_object_set (G_OBJECT (video_queue), "leaky", TRUE, "flush-on-eos", TRUE, NULL);
    g_object_set (G_OBJECT (rtsp_video_switch), "sync-streams", TRUE, "sync-mode", 0, "cache-buffers", TRUE, NULL);
    for (i=0; i < NUM_OF_CAMERA; ++i)
    {
        char rtspsrc_name[255];
        snprintf(rtspsrc_name, sizeof(rtspsrc_name), "rtspsrc_%d", i);
        video_pipeline[i].rtspsrc = gst_element_factory_make ("rtspsrc", rtspsrc_name);
        snprintf(rtspurl[i], 255, CAMERA_RTSP_URL, i+1);
        fprintf(stderr, "setting rtsp url for name %s as %s\n", rtspsrc_name, rtspurl[i]);
        g_object_set (G_OBJECT (video_pipeline[i].rtspsrc), "location", rtspurl[i], "latency", 2000, "do-retransmission", TRUE, NULL);
        
        char queue_name[255];
        snprintf(queue_name, sizeof(queue_name), "queue_%d", i);
        video_pipeline[i].rtspsrc_queue = gst_element_factory_make ("queue2", queue_name);
        g_object_set (G_OBJECT (video_pipeline[i].rtspsrc_queue), "max-size-time", 200000000000, "max-size-bytes", 2097152000, "max-size-buffers", 40000, NULL);
        
        char rtph264depay_name[255];
        snprintf(rtph264depay_name, sizeof(rtph264depay_name), "rtph264depay_%d", i);
        video_pipeline[i].rtph264depay = gst_element_factory_make ("rtph264depay", rtph264depay_name);
        
        char shift_name[255];
        snprintf(shift_name, sizeof(shift_name), "shift_%d", i);
        video_pipeline[i].shift = gst_element_factory_make ("shift", shift_name);
        g_object_set (G_OBJECT (video_pipeline[i].shift), "delay", 550000, "running-time", TRUE, NULL);
        
        /*char sink_sync_name[255];
        snprintf(sink_sync_name, sizeof(sink_sync_name), "sink_sync_%d", i);
        video_pipeline[i].sink_sync = gst_element_factory_make ("identity", sink_sync_name);
        g_object_set (G_OBJECT (video_pipeline[i].sink_sync), "sync", TRUE, NULL);*/
        
        if(!video_pipeline[i].rtspsrc || !video_pipeline[i].rtspsrc_queue || !video_pipeline[i].rtph264depay || !video_pipeline[i].shift /*|| !video_pipeline[i].sink_sync*/)
        {
            fprintf (stderr, "failed to create video pipeline for camindex %d ... Exiting\n", i);
            exit(1);
        }
    }
    
	//mixing and broadcast pipeline
	h264parse = gst_element_factory_make("h264parse", "h264parse");
    flvmux = gst_element_factory_make("flvmux", "flvmux");
    //mixing_queue = gst_element_factory_make("queue", "mixing_queue");
    rtmpsink = gst_element_factory_make("rtmpsink", "rtmpsink");

    if(!flvmux || !h264parse /* || !mixing_queue || */ || !rtmpsink)
    {
        fprintf (stderr, "One of the elements in mixing pipeline wasn't created... Exiting\n");
        exit(1);
    }

    g_object_set (G_OBJECT (flvmux), "streamable", TRUE, NULL);
    g_object_set (G_OBJECT (rtmpsink), "sync", TRUE, "async", TRUE, "location", BROADCAST_RTMP_URL, NULL);

    /* Add Elements to the Bin */
    gst_bin_add_many (GST_BIN (pipeline), audio_rtspsrc, audio_queue1, audio_rtpmpadepay, audio_mpegaudioparse, audio_mad, audio_audioconvert, /*audio_queue2,*/ audio_voaacenc, audio_aacparse, /*audio_queue3,*/
    
    video_pipeline[0].rtspsrc, video_pipeline[1].rtspsrc, video_pipeline[2].rtspsrc, video_pipeline[3].rtspsrc, video_pipeline[4].rtspsrc,
    video_pipeline[0].rtspsrc_queue, video_pipeline[1].rtspsrc_queue, video_pipeline[2].rtspsrc_queue, video_pipeline[3].rtspsrc_queue, video_pipeline[4].rtspsrc_queue,
    video_pipeline[0].rtph264depay, video_pipeline[1].rtph264depay, video_pipeline[2].rtph264depay, video_pipeline[3].rtph264depay, video_pipeline[4].rtph264depay,
    video_pipeline[0].shift, video_pipeline[1].shift, video_pipeline[2].shift, video_pipeline[3].shift, video_pipeline[4].shift,
    /*video_pipeline[0].sink_sync, video_pipeline[1].sink_sync, video_pipeline[2].sink_sync, video_pipeline[3].sink_sync, video_pipeline[4].sink_sync,*/
    
    
    rtsp_video_switch, segment, video_queue,
    h264parse, flvmux, /* mixing_queue, */ rtmpsink, NULL);

	// partial linking
    if (!gst_element_link_many (audio_queue1, audio_rtpmpadepay, audio_mpegaudioparse, audio_mad, audio_audioconvert, /* audio_queue2, */ audio_voaacenc, audio_aacparse, /* audio_queue3, */ NULL))
    {
        fprintf(stderr, "Linking Audio pipeline has Failed...\n");
        exit(1);
    }
    if (!gst_element_link_many(rtsp_video_switch, segment, video_queue, h264parse, NULL)) {
		fprintf(stderr, "Linking rtsp_video_switch, segment and video_queue has Failed...\n");
		exit(1);
	}
    if (!gst_element_link_many (flvmux, /* mixing_queue,*/ rtmpsink, NULL))
    {
        fprintf(stderr, "Linking mixer with broadcast pipeline has Failed...\n");
        exit(1);
    }

    rtsp_video_switch_srcpad = gst_element_get_static_pad (rtsp_video_switch, "src");

    // Linking audio pipeline with flvmux
    GstPadTemplate *flvmux_sink_pad_template_audio;
    if (!(flvmux_sink_pad_template_audio = gst_element_class_get_pad_template(GST_ELEMENT_GET_CLASS(flvmux), "audio")))
    {
        fprintf(stderr, "Unable to get pad template for audio for flvmux element");
        exit(1);
    }
    GstPad * audio_aacparse_src_pad = gst_element_get_static_pad (audio_aacparse, "src");
    if(audio_aacparse_src_pad == NULL)
    {
        fprintf(stderr, "unable to get src pad for aacparse element\n");
        exit(1);
    }
    GstPad * flvmux_sink_audio_pad = gst_element_request_pad(flvmux, flvmux_sink_pad_template_audio, NULL, NULL);
    if(flvmux_sink_audio_pad == NULL)
    {
        fprintf(stderr, "unable to get audio sink pad for flvmux\n");
        exit(1);
    }
    if (gst_pad_link (audio_aacparse_src_pad, flvmux_sink_audio_pad) != GST_PAD_LINK_OK )
    {
        fprintf(stderr, "unable to link aacparse with flvmixer\n");
        exit(1);
    }

	// Linking video_queue with flvmux
    GstPadTemplate *flvmux_sink_pad_template_video;
    if (!(flvmux_sink_pad_template_video = gst_element_class_get_pad_template(GST_ELEMENT_GET_CLASS(flvmux), "video")))
    {
        printf ("Unable to get pad template for video for flvmux element");
        exit(1);
    }
    GstPad * h264parse_src_pad = gst_element_get_static_pad (h264parse, "src");
    if(h264parse_src_pad == NULL)
    {
        fprintf(stderr, "unable to get src pad for h264parse element\n");
        exit(1);
    }
    GstPad * flvmux_sink_video_pad = gst_element_request_pad(flvmux, flvmux_sink_pad_template_video, NULL, gst_caps_new_simple("video/x-h264", "stream-format", G_TYPE_STRING, "avc", NULL));
    if(flvmux_sink_video_pad == NULL)
    {
        fprintf(stderr, "unable to get video sink pad for flvmux\n");
        exit(1);
    }
    if (gst_pad_link (h264parse_src_pad, flvmux_sink_video_pad) != GST_PAD_LINK_OK )
    {
        fprintf(stderr, "unable to link h264parse with flvmixer\n");
        exit(1);
    }

	// Linking audio rtsp src with rest of audio pipeline
    g_signal_connect(audio_rtspsrc , "pad-added", G_CALLBACK (on_pad_added_audio), audio_queue1);
    
    // Linking all video pipeline for all cameras
    for(i=0; i < NUM_OF_CAMERA; ++i)
    {
        rtspsrc_pad_added_camindex[i].camindex = i;
        fprintf(stderr, "trying to connect pad for cam %d\n", i);
        if (!gst_element_link_many (video_pipeline[i].rtspsrc_queue, video_pipeline[i].rtph264depay, video_pipeline[i].shift, /*video_pipeline[i].sink_sync,*/ rtsp_video_switch, NULL))
		{
			printf("Linking failed for video pipeline ... Exiting \n");
			exit(1);
		}
		g_signal_connect(video_pipeline[i].rtspsrc , "pad-added", G_CALLBACK (on_pad_added_video), &rtspsrc_pad_added_camindex[i]);
	}

    ////////////////////////////////////////////////////////////
    bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
    loop = g_main_loop_new (NULL, FALSE);
    g_timeout_add_seconds (1, camswitch_timeout_cb, loop); // limit camera swithing gap to minimum 5 seconds
    gst_bus_add_watch (bus, bus_call, loop);
    gst_object_unref (bus);
    gst_element_set_state (pipeline, GST_STATE_PLAYING);
    g_main_loop_run (loop);
}

void * server_thread (void *data)
{
    char msg[256];

    for(;;)
    {
        int newsockfd = -1;
        struct sockaddr_in cli_addr;
        int clilen = 0;
        int n = 0;

        clilen = sizeof(cli_addr);
        newsockfd = accept(sockfd, (struct sockaddr *) &cli_addr, &clilen);
        if (newsockfd < 0)
        {
            printf("Line %d, error on accept", __LINE__);
        }
        bzero(msg, 256);
        n = read(newsockfd, msg, 255);
        if(n < 0)
        {
            printf("Line %d, error reading from socket\n", __LINE__);
        }

        // process msg command
        printf("Line %d, msg is \n%s \n", __LINE__, msg);
        if(strncmp(msg, BROADCAST_START_CMD, strlen(BROADCAST_START_CMD)) == 0)
        {
            int camindex;
            sscanf(msg, BROADCAST_START_CMD " %d", &camindex);
            if(pipeline == NULL)
            {
                pthread_mutex_lock(&lock_cam_switch);
                old_cam_index = new_cam_index = camindex;
                fprintf(stderr, "Line %d: MAIN THREAD: Switching to camera %d\n", __LINE__, camindex);
                int *arg = malloc(sizeof(*arg));
                if (arg == NULL)
                {
                    fprintf(stderr, "Couldn't allocate memory for gstreamer thread arg.\n");
                    pthread_mutex_destroy(&lock_cam_switch);
                    exit(1);
                }
                *arg = camindex;
                pthread_create(&pthread_gst_pipeline_thread, NULL, gst_pipeline_start, arg);
                sleep(15);// dirty hack to wait for pipeline to start
                pthread_mutex_unlock(&lock_cam_switch);
            }
            else
            {
				pthread_mutex_lock(&lock_cam_switch);
                fprintf(stderr, "Line %d: MAIN THREAD: Switching to camera %d\n", __LINE__, camindex);
                new_cam_index = camindex;
                pthread_mutex_unlock(&lock_cam_switch);
            }
        }
        else if(strncmp(msg, BROADCAST_STOP_CMD, strlen(BROADCAST_STOP_CMD)) == 0)
        {
            printf("Line %d: received cmd for stopping broadcast\n", __LINE__);
            if(pipeline != NULL)
            {
                gst_element_set_state (pipeline, GST_STATE_NULL);
                g_print ("Eliminating Pipeline\n");
                gst_object_unref (GST_OBJECT (pipeline));
                printf("broadcast stopped\n");
                g_main_loop_quit (loop);
                pthread_join(pthread_gst_pipeline_thread, NULL);
                printf("Exiting gstreamer main loop\n");
                pipeline = NULL;
                exit(1);
            }
        }

        n = write(newsockfd, "I got your message", 18);
        if (n < 0)
        {
            printf("Line %d, error writing to socket\n", __LINE__);
        }
        close(newsockfd);
        printf("waiting for next command\n");
    }
}


int
main (int argc, char **argv)
{
    struct sockaddr_in serv_addr;
    if (pthread_mutex_init(&lock_cam_switch, NULL) != 0)
    {
        printf("\n mutex init failed for camera switching\n");
        exit(1);
    }
    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    if (sockfd < 0)
        printf("Line %d, error opening socket",__LINE__);
    bzero((char *) &serv_addr, sizeof(serv_addr));

    serv_addr.sin_family = AF_INET;
    serv_addr.sin_addr.s_addr = INADDR_ANY;
    serv_addr.sin_port = htons(SERVER_PORT);

    if (bind(sockfd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0)
        printf("Line %d, error on binding to port %d", __LINE__, SERVER_PORT);

    listen(sockfd, SOMAXCONN);

    setenv("GST_DEBUG", "2", 1);
    pthread_create(&pthread_server_thread, NULL, server_thread, NULL);
    pthread_join(pthread_server_thread, NULL);
    pthread_mutex_destroy(&lock_cam_switch);
    return 0;
}
