gst_experiment.c

#include <gst/gst.h>
#include <glib.h>
#include <czmq.h>
#include <pthread.h>
#include <stdio.h>

/* Gstreamer "main" pipeline */
GstElement *pipeline;

void* zmq_thread(void *data)
{
	void *ctx = zctx_new();
	void *sub = zsocket_new(ctx, ZMQ_SUB);
	char msg[512];
	int i;

	zsocket_connect(sub, "tcp://127.0.0.1:4060");
	zsocket_set_subscribe(sub,"");

	printf("ZMQ: Subscriber initialized\n");

	while (1) {
		snprintf(msg, sizeof(msg), "%s", zstr_recv(sub));
		//printf("Message: %s\n", msg);
		if (!strncmp(msg, "START", 5)) {
			gst_element_set_state(pipeline, GST_STATE_PLAYING);
		} else {
			gst_element_set_state(pipeline, GST_STATE_PAUSED);
		}
		printf("ZMQ: %s\n", msg);
	}

	return NULL;
}

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_ERROR:
	{
		gchar *debug;
		GError *error;

		gst_message_parse_error(msg, &error, &debug);
		g_free(debug);

		g_printerr("Error: %s\n", error->message);
		g_error_free(error);

		g_main_loop_quit(loop);
		break;
	}
	default:
		break;
	}

	return TRUE;
}

void gst_controller_zmq_thr_creat(void)
{
	pthread_t zmq_thr;
	int ret;

	ret = pthread_create(&zmq_thr, NULL, zmq_thread, NULL);

	/* FIXME be aware of the fact that the thread runs "forever" and
	   we do not call pthread_join() anywhere */
}

#define V4L_DEFAULT_DEV			"/dev/video0"
int main(int argc, char *argv[])
{
	GMainLoop *loop;
	GstBus *bus;
	guint bus_watch_id;
	char* v4l_dev;

	GstElement *source;
	GstElement *imx_videotransform;
	GstElement *imx_h264;
	GstElement *h264parse;
	GstElement *flvmux;
	GstElement *rtmpsink;
	GstCaps *caps_v4l;
	GstCaps *caps_h264;
	GstCaps *caps_flv;

	gst_controller_zmq_thr_creat();

	if (argc > 1)
		v4l_dev = argv[1];
	else
		v4l_dev = V4L_DEFAULT_DEV;
	g_print("Using v4l device %s\n", v4l_dev);


	/* Initialisation */
	gst_init(&argc, &argv);
	loop = g_main_loop_new(NULL, FALSE);

	/* Create gstreamer elements */
	pipeline = gst_pipeline_new("meo-rtmp-stream");
	source = gst_element_factory_make("v4l2src", NULL);
	imx_videotransform = gst_element_factory_make("imxipuvideotransform", NULL);
	imx_h264 = gst_element_factory_make("imxvpuenc_h264", NULL);
	h264parse = gst_element_factory_make("h264parse", NULL);
	flvmux = gst_element_factory_make("flvmux", NULL);
	rtmpsink = gst_element_factory_make("rtmpsink", NULL);

	if (!pipeline || !source || !imx_videotransform ||
	    !imx_h264 || !h264parse || !flvmux || !rtmpsink) {
		g_printerr("One element could not be created. Exiting.\n");
		return -1;
	}

	caps_v4l = gst_caps_from_string("video/x-raw,width=640,height=480,framerate=20/1");
	caps_h264 = gst_caps_from_string("video/x-h264");
	caps_flv = gst_caps_from_string("video/x-flv");

	g_object_set(G_OBJECT(source), "device", v4l_dev, NULL);
	g_object_set(G_OBJECT(imx_h264), "gop-size", 40,
		     "idr-interval", 16,
		     "bitrate", 512, NULL);
	g_object_set(G_OBJECT(flvmux), "streamable", TRUE, NULL);
	g_object_set(G_OBJECT(rtmpsink), "location",
		     "rtmp://46.28.108.237:1935/meo?access_token=5f6f3c10d3f3448caf3fbb89d63f36a1/132",
		     "sync", FALSE, NULL);

	/* 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);

	gst_bin_add_many(GST_BIN(pipeline), source, imx_videotransform,
			 imx_h264, h264parse, flvmux, rtmpsink, NULL);

	gst_element_link_filtered(source, imx_videotransform, caps_v4l);
	gst_caps_unref(caps_v4l);
	gst_element_link_many(imx_videotransform, imx_h264, h264parse, NULL);
	gst_element_link_filtered(h264parse, flvmux, caps_h264);
	gst_caps_unref(caps_h264);
	gst_element_link_filtered(flvmux, rtmpsink, caps_flv);
	gst_caps_unref(caps_flv);

	gst_element_set_state(pipeline, GST_STATE_PAUSED);
	g_print("Running...\n");
	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");
	gst_object_unref(GST_OBJECT(pipeline));
	g_source_remove(bus_watch_id);
	g_main_loop_unref(loop);

	return 0;
}