gstreamer-rtsp-server test-appsrc

/* GStreamer
 * Copyright (C) 2008 Wim Taymans <wim.taymans at gmail.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
 * Boston, MA 02110-1301, USA.
 */

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <getopt.h>
#include <gst/gst.h>
#include <gst/app/gstappsrc.h>
#include <gst/app/gstappsink.h>
#include <gst/rtsp-server/rtsp-server.h>
#include <macqc-core.h>

typedef struct _App App;
struct _App
{
	GstElement *videosink;
};
App s_app;

typedef struct
{
	App *globalApp;
	GstClockTime timestamp;
} MyContext;

GMainLoop *loop;
GstClockTime base_time, minlat, maxlat;
gint fps_n, fps_d;

void usage()
{
	printf("Usage: macq-gst-rtsp-server -f ini_file [-p port]\n");
	printf("Any appsrc launch line with property \"name=mysrc\" works as long as it contains elements named pay%%d.\n");
	printf("Each element with pay%%d names will be a stream.\n");
	exit(0);
}

const char* get_in_section(miniconfiguration* conf, const char* section, const char* key)
{
  const char* value;
  if (!miniconfig_get_in_section(conf, section, key, &value, ""))
	{
	g_print ("[rtsp-server] failed to load parameter %s. exiting.\n", key);
	exit(1);
	}
  return value;
}

void check_port(const char* port)
{
	int port_nb = atoi(port);
	if ((port_nb < 1024 || port_nb > 65535)  && port_nb != 554)
		{
		g_print("Port number must be 554 or between 1024 and 65535\n");
		exit(1);
		}
}

/* called when we need to give data to appsrc */
static void
need_data (GstElement * appsrc, guint unused, MyContext * ctx)
{
  GstFlowReturn ret;
  GstSample *sample = gst_app_sink_pull_sample (GST_APP_SINK(ctx->globalApp->videosink));
  if (sample != NULL)
  {
        GstBuffer *buffer = gst_sample_get_buffer(sample);
        gst_sample_unref (sample);
        GST_BUFFER_PTS (buffer) = ctx->timestamp;
        GST_BUFFER_DURATION (buffer) = gst_util_uint64_scale_int (GST_SECOND, fps_d, fps_n);
        ctx->timestamp += GST_BUFFER_DURATION (buffer);
        g_signal_emit_by_name (appsrc, "push-buffer", buffer, &ret);
  }
}

/* called when a new media src_pipeline is constructed. We can query the
 * src_pipeline and configure our appsrc */
static void
media_configure (GstRTSPMediaFactory * factory, GstRTSPMedia * media,
    App *app)
{
  GstElement *element, *appsrc;
  GstClock *clock;
  MyContext *ctx;

  /* get the element used for providing the streams of the media */
  element = gst_rtsp_media_get_element (media);

  /* set appsrc pipeline clock to the same clock as appsink */
  clock = gst_system_clock_obtain();
  gst_pipeline_use_clock(GST_PIPELINE(element), clock);
  gst_object_unref(clock);
  gst_element_set_base_time(element, base_time);
  gst_element_set_start_time(element, GST_CLOCK_TIME_NONE);

  /* get our appsrc, we named it 'mysrc' with the name property */
  appsrc = gst_bin_get_by_name_recurse_up (GST_BIN (element), "mysrc");

  gst_rtsp_media_set_reusable(media, TRUE);

  /* this instructs appsrc that we will be dealing with timed buffer */
  gst_util_set_object_arg (G_OBJECT (appsrc), "format", "time");

  /* configure the caps of the video */
  g_object_set (G_OBJECT (appsrc), "max-bytes", gst_app_src_get_max_bytes(GST_APP_SRC(appsrc)), NULL);

  /* configure the min and max latencies */
  g_object_set (G_OBJECT (appsrc), "min-latency", minlat, NULL);
  g_object_set (G_OBJECT (appsrc), "max-latency", maxlat, NULL);

  ctx = g_new0 (MyContext, 1);
  ctx->globalApp = app;
  ctx->timestamp = 0;

  /* make sure the data is freed when the media is gone */
  g_object_set_data_full (G_OBJECT (media), "my-extra-data", ctx,
      (GDestroyNotify) g_free);

  /* install the callback that will be called when a buffer is needed */
  g_signal_connect (appsrc, "need-data", (GCallback) need_data, ctx);
  gst_object_unref(appsrc);
  gst_object_unref(element);
}

// Bus message handler
static gboolean
bus_callback(GstBus *bus, GstMessage *msg, gpointer data)
{
  GstElement *sink_pipeline = GST_ELEMENT(data);
  switch (GST_MESSAGE_TYPE(msg)) {
  case GST_MESSAGE_EOS:
        if (!gst_element_seek(sink_pipeline, 1.0, GST_FORMAT_TIME, GST_SEEK_FLAG_FLUSH,
          GST_SEEK_TYPE_SET, 1000000000,
          GST_SEEK_TYPE_NONE, GST_CLOCK_TIME_NONE)) {
                g_message("Seek failed!");
        }
	g_printerr("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 in bus: %s\n", error->message);
        g_error_free(error);
        g_main_loop_quit(loop);
        break;
  }
  default:
        break;
  }
  return TRUE;
}

int
main (int argc, char *argv[])
{
  int c;
  const char *file = NULL;
  char *port = NULL;

  static struct option long_options[] =
   {
    {"file",	required_argument, 0, 'f'},
    {"port",	optional_argument, 0, 'p'},
    {"help",	no_argument, 0, 'h'},
    {0, 0, 0, 0}
   };

  while (1)
    {

	c = getopt_long (argc, argv, "hf:p:", long_options, NULL);
	if (c == -1)
	  break;

	switch (c)
	  {
	  case 'f':
	    file = optarg;
	    break;

	  case 'p':
	    port = optarg;
	    break;

	  default:
	    usage();
	  }
    }

  if (!file)
    {
    usage();
    }

  GstRTSPServer *server;
  GstRTSPMountPoints *mounts;
  GstRTSPMediaFactory *factory;
  GstElement *sink_pipeline;
  GstBus *bus;
  GstClock *clock;
  App *app = &s_app;

  /* read config file */
  const char *appsink_chain = NULL;
  const char *appsrc_chain = NULL;
  const char *rtsp_stream_name = NULL;
  const char *port_from_file = NULL;

  miniconfiguration* conf = miniconfig_create();
  if (mfile_exists(file))
	miniconfig_load_file(conf, file, 0);
  else
  	{
	g_print ("Can't find configuration file: \"%s\"\n", file);
	exit(1);
	}

  appsrc_chain = get_in_section(conf, "general", "appsrc-chain");
  appsink_chain = get_in_section(conf, "general", "appsink-chain");
  rtsp_stream_name = get_in_section(conf, "general", "rtsp-stream-name");
  miniconfig_get_in_section(conf, "general", "port", &port_from_file, "");

  gst_init (&argc, &argv);

  loop = g_main_loop_new (NULL, FALSE);


  /* create sink_pipeline */
  GError *error = NULL;
  sink_pipeline = gst_parse_launch(appsink_chain, &error);
  if(error != NULL)
  	g_printerr("Error in appsink_chain: %s\n", error->message);

  clock = gst_system_clock_obtain();
  gst_pipeline_use_clock(GST_PIPELINE(sink_pipeline), clock);
  gst_object_unref(clock);
  base_time = gst_clock_get_time(gst_system_clock_obtain());
  gst_element_set_base_time(sink_pipeline, base_time);
  gst_element_set_start_time(sink_pipeline, GST_CLOCK_TIME_NONE);

  app->videosink = gst_bin_get_by_name(GST_BIN(sink_pipeline), "mysink");
  if(!app->videosink) {
        g_printerr("Failed to get sink element by name.\n");
        return -1;
  }

  /* parse pipeline string for any framerate property */
  GstIterator *it = gst_bin_iterate_recurse(GST_BIN(sink_pipeline));
  GValue item = G_VALUE_INIT;
  GstElement *element;
  GstCaps *caps;
  const GstStructure *structure;
  const GValue *framerate;
  gboolean done = FALSE;
  fps_n = 1, fps_d = 1;

  while (!done)
  {
	switch(gst_iterator_next (it, &item))
	{
		case GST_ITERATOR_OK:
			element = g_value_get_object(&item);
			if (strcmp(G_OBJECT_TYPE_NAME(element),"GstCapsFilter") == 0)
			{
				g_object_get(G_OBJECT(element), "caps", &caps, NULL);
				structure = gst_caps_get_structure(caps, 0);
				framerate = gst_structure_get_value(structure, "framerate");
				if (framerate != NULL)
				{
					fps_n = gst_value_get_fraction_numerator(framerate);
					fps_d = gst_value_get_fraction_denominator(framerate);
				}
			}
			g_value_reset(&item);
			break;
		case GST_ITERATOR_RESYNC:
			gst_iterator_resync(it);
			break;
		default:
			done = TRUE;
			break;
	}
  }
  g_value_unset(&item);
  gst_iterator_free(it);

  bus = gst_pipeline_get_bus (GST_PIPELINE (sink_pipeline));
  gst_bus_add_watch (bus, bus_callback, sink_pipeline);
  gst_object_unref(bus);

  /* start playing sink_pipeline */
  gst_element_set_state (sink_pipeline, GST_STATE_PLAYING);
  
  /* wait until it's up and running or failed */
  if (gst_element_get_state(sink_pipeline, NULL, NULL, -1) == GST_STATE_CHANGE_FAILURE)
  {
	g_error("Failed to go into PLAYING state");
  }

  GstQuery *query = gst_query_new_latency();
  if (gst_element_query(sink_pipeline, query))
  {
	gboolean live;
	gst_query_parse_latency(query, &live, &minlat, &maxlat);
  	g_print("LIVE: %d\n", live);
	printf("Minimum latency: %" GST_TIME_FORMAT "\n", GST_TIME_ARGS(minlat));
	printf("Maximum latency: %" GST_TIME_FORMAT  "\n", GST_TIME_ARGS(maxlat));
  }
  gst_query_unref(query);

  /* create a server instance */
  server = gst_rtsp_server_new ();

  if (!port)
	{
	if (port_from_file[0])
		port = strdup(port_from_file);
	else
		port = "8554";
	}
  check_port(port);
  gst_rtsp_server_set_service(server, port);

  /* 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 src_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_shared (factory, TRUE);
  gst_rtsp_media_factory_set_eos_shutdown(factory, TRUE);

  gst_rtsp_media_factory_set_launch (factory, appsrc_chain);

  /* notify when our media is ready, This is called whenever someone asks for
   * the media and a new src_pipeline with our appsrc is created */
  g_signal_connect (factory, "media-configure", (GCallback) media_configure,
      app);

  char *url = malloc(strlen(rtsp_stream_name)+2);
  sprintf(url, "/%s", rtsp_stream_name);

  /* destroy config file */
  miniconfig_destroy(conf);

  /* attach the appsrc_chain factory to the /rtsp_stream_name url */
  gst_rtsp_mount_points_add_factory (mounts, url, factory);

  /* 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 ("stream ready at rtsp://127.0.0.1:%s%s\n", port, url);

  g_main_loop_run (loop);

  g_print("Returned, stopping playback\n");
  gst_element_set_state (sink_pipeline, GST_STATE_NULL);

  g_print("Deleting sink_pipeline\n");
  gst_object_unref (GST_OBJECT(sink_pipeline));
  g_main_loop_unref(loop);

  return 0;
}

Hi Mezzano,

I'm trying to use this sample code. However, it depends on macqc-core.h. I couldn't figure out where can I find this file.

Can you give me a hint on this please?

Thanks