GST RTP receiver

gistfile1.txt

#include <glib.h>

#include <gst.h>
#include <gst/app/gstappsink.h>

#define DEFAULT_PORT 5000

typedef void (*received_frame) (RtpReceiver* r, unsigned char* data, gsize size, gboolean eos) RtpReceiverCallback;

typedef struct _RtpReceiver {

	GstPipeline pipeline;
	GstElement *udpsrc, *rtph264depay, *decodebin, *videoconvert, *appsink;
	RtpReceiverCallback cb;

} RtpReceiver;

typedef struct _RtpDataInfo {
	unsigned char* data;
	gsize size;
} RtpDataInfo;

static RtpDataInfo buffer_to_data_info (GstBuffer * buf)
{
	RtpDataInfo info;
	GstMapInfo buf_info;
	unsigned char* data;

	if (!gst_buffer_map (buf, &buf_info, GST_MAP_READ)) {
		info.data = NULL;
		info.size = 0;
		return info;
	}

	data = (unsigned char*) malloc (buf_info.size);
	memcpy (data, buf_info.data, buf_info.size);

	gst_buffer_unmap (buf, &buf_info);

	info.data = data;
	info.size = buf_info.size;

	return info;
}

static void eos (GstAppSink *appsink, gpointer user_data)
{
	RtpReceiver* r = (RtpReceiver*) user_data;

	r.cb (r, NULL, 0, TRUE);
}

static GstFlowReturn new_preroll (GstAppSink *appsink, gpointer user_data)
{
	GstSample* sample = gst_app_sink_pull_preroll (appsink);

	g_return_val_if_fail (sample, GST_FLOW_ERROR);

	GstBuffer* buf = gst_sample_get_buffer (sample);
	RtpDataInfo info = buffer_to_data_info (buf);

	r.cb (r, info.data, info.size, FALSE);

	gst_sample_unref (sample);

	return GST_FLOW_OK;
}

static GstFlowReturn new_sample (GstAppSink *appsink, gpointer user_data)
{
	GstSample* sample = gst_app_sink_pull_sample (appsink);

	g_return_val_if_fail (sample, GST_FLOW_ERROR);

	GstBuffer* buf = gst_sample_get_buffer (sample);
	RtpDataInfo info = buffer_to_data_info (buf);

	r.cb (r, info.data, info.size, FALSE);

	gst_sample_unref (sample);
}

RtpReceiver* rtp_receiver_new(RtpReceiverCallback cb)
{
	GstCaps *caps;
	GstAppSinkCallbacks callbacks;
	RtpReceiver* state = g_slice_new0(RtpReceiver);

	state.cb = cb;

	state.pipeline = gst_pipeline_new ();

	state.udpsrc = gst_element_factory_make ("udpsrc", NULL);
	state.rtph264depay = gst_element_factory_make ("rtph264depay", NULL);
	state.decodebin = gst_element_factory_make ("decodebin", NULL);
	state.videoconvert = gst_element_factory_make ("videoconvert", NULL);
	state.appsink = gst_element_factory_make ("appsink", NULL);

	gst_bin_add_many (pipeline,
		state.udpsrc, state.rtph264depay, state.decodebin, state.videoconvert, state.appsink, NULL);

	if (!gst_element_link_many (state.udpsrc, state.rtph264depay, state.decodebin, state.videoconvert, state.appsink, NULL)) {
		g_warning ("Failed to link pipeline");
		return NULL;
	}

	g_object_set (state.udpsrc, "port", DEFAULT_PORT, NULL);

	caps = gst_caps_new_simple ("video/x-raw",
	   "format", G_TYPE_STRING, "RGBA",
	   //"framerate", GST_TYPE_FRACTION, 25, 1,
	   //"width", G_TYPE_INT, 320,
	   //"height", G_TYPE_INT, 240,
	   NULL);

	gst_app_sink_set_caps (GST_APP_SINK (state.appsink), caps);

	callbacks.eos = eos;
	callbacks.new_preroll = new_preroll;
	callbacks.new_sample = new_sample;

	gst_app_set_callbacks (state.appsink, callbacks, state, NULL);

	gst_element_set_state (state.pipeline, GST_STATE_PLAYING);

	gst_caps_unref (caps);

	return state;
}

void rtp_receiver_free(RtpReceiver *r)
{
	gst_element_set_state (r.pipeline, GST_STATE_NULL);
	gst_object_unref (r.pipeline);
	g_slice_free(r);
}