example.c

/**
 * gcc -o ex example.c -g `pkg-config --cflags --libs gstreamer-1.0 gio-2.0`
 *
 * Instruction
 *   1. Run 'ex' to listen srt stream
 *   2. Then, send stream to 'ex' by 'gst-launch-1.0' command
 *      $ gst-launch-1.0 videotestsrc ! x264enc tune=zerolatency ! queue ! h264parse ! mpegtsmux ! tsparse ! queue ! srtsink uri='srt://127.0.0.1:8888?mode=caller' 
 */
#include <gst/gst.h>
#include <gio/gio.h>

#define SRT_URI "srt://0.0.0.0:8888?mode=listener"

typedef struct _CustomData
{
  GstElement *pipeline;
  GMainLoop *loop;

  GFile *dump_file;
  GFileOutputStream *ostream;
} CustomData;

/* The appsink has received a buffer */
static GstFlowReturn
new_sample_cb (GstElement * sink, CustomData * data)
{
  GstSample *sample;

  /* Retrieve the buffer */
  g_signal_emit_by_name (sink, "pull-sample", &sample);
  if (sample) {
    GstBuffer *buffer;
    GstMemory *memory;
    GstMapInfo map_info;

    /* refer to API document to handle byte stream
     *   https://gstreamer.freedesktop.org/documentation/gstreamer/gstbuffer.html?gi-language=c
     */
    buffer = gst_sample_get_buffer (sample);
    memory = gst_buffer_get_all_memory (buffer);

    if (gst_memory_map (memory, &map_info, GST_MAP_READ)) {

      gssize wlen;

      /*
       * map_info.data(guint8 *):  a pointer to the mapped data
       * map_info.size(gsize)   :  the valid size in data
       */

      wlen = g_output_stream_write (G_OUTPUT_STREAM (data->ostream),
          map_info.data, map_info.size, NULL, NULL);

      g_print ("received %lu bytes, wrote %ld bytes\n", map_info.size, wlen);

      gst_memory_unmap (memory, &map_info);
    } else {
      // failed to get memory info
    }

    gst_memory_unref (memory);
    gst_sample_unref (sample);

    return GST_FLOW_OK;
  }

  return GST_FLOW_ERROR;
}


static void
bus_message_cb (GstBus * bus, GstMessage * msg, CustomData * data)
{

  switch (GST_MESSAGE_TYPE (msg)) {

    case GST_MESSAGE_ERROR:{
      GError *err;
      gchar *debug;

      gst_message_parse_error (msg, &err, &debug);
      g_print ("Error: %s\n", err->message);
      g_error_free (err);
      g_free (debug);

      gst_element_set_state (data->pipeline, GST_STATE_READY);
      g_main_loop_quit (data->loop);
      break;
    }

    case GST_MESSAGE_STATE_CHANGED:{
      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));
      }
      break;
    }

    default:
      break;
  }
}

int
main (int argc, char *argv[])
{
  CustomData data;
  GstElement *src;
  GstElement *sink;

  GstBus *bus;
  GError *error = NULL;

  GstFlowReturn ret;


  gst_init (&argc, &argv);

  /* Initialize our data structure */
  memset (&data, 0, sizeof (data));

  data.pipeline =
      gst_parse_launch ("srtsrc name=src ! queue ! tsparse ! appsink name=sink",
      &error);

  /* Create dump file */
  data.dump_file = g_file_new_for_path ("./dump_file.ts");
  data.ostream =
      g_file_create (data.dump_file, G_FILE_CREATE_NONE, NULL, &error);

  if (error) {
    g_print ("Error: %s\n", error->message);
    g_clear_error (&error);
  }

  src = gst_bin_get_by_name (GST_BIN (data.pipeline), "src");
  sink = gst_bin_get_by_name (GST_BIN (data.pipeline), "sink");

  /* set srt uri */
  g_object_set (src, "uri", SRT_URI, NULL);

  /* Configure appsink */
  g_object_set (sink, "emit-signals", TRUE, "async", FALSE, "sync", FALSE,
      NULL);
  g_signal_connect (sink, "new-sample", G_CALLBACK (new_sample_cb), &data);

  gst_object_unref (src);
  gst_object_unref (sink);

  bus = gst_element_get_bus (data.pipeline);
  gst_bus_add_signal_watch (bus);
  g_signal_connect (bus, "message", G_CALLBACK (bus_message_cb), &data);
  gst_object_unref (bus);

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

  data.loop = g_main_loop_new (NULL, FALSE);
  g_main_loop_run (data.loop);

  gst_element_set_state (data.pipeline, GST_STATE_NULL);

  g_main_loop_unref (data.loop);
  g_clear_object (&data.pipeline);

  g_object_unref (data.ostream);
  g_object_unref (data.dump_file);

  return 0;
}