/* ex: set tabstop=2 shiftwidth=2 expandtab: */
/*
 * GStreamer
 * Copyright (C) 2010 Marc Leeman <marc.leeman@gmail.com>
 * 
 */

/**
 * SECTION:element-plugin
 *
 * Recombine GStreamer buffers into larger ones.
 *
 * <refsect2>
 * <title>Example launch line</title>
 * |[
 * gst-launch -v -m udpsrc ! mpegtsparse ! bufresize size=1450 ! udpsink
 * ]|
 * </refsect2>
 */

#ifdef HAVE_CONFIG_H
#  include <config.h>
#endif

#include <gst/gst.h>

#include "gstbufresize.h"

GST_DEBUG_CATEGORY_STATIC (gst_bufresize_debug);
#define GST_CAT_DEFAULT gst_bufresize_debug

/* 0 size means just pass the buffer along */
#define GST_RECOMBINE_DEFAULT_BUFFER_SIZE (0)
#define GST_RECOMBINE_MIN_BUFFER_SIZE (0)
#define GST_RECOMBINE_MAX_BUFFER_SIZE (16*1024*1024)

enum
{
  PROP_0,
  PROP_SIZE,
  PROP_LAST,
};

static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE ("src",
    GST_PAD_SRC,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS ("ANY")
    );

static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE ("sink",
    GST_PAD_SINK,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS ("ANY")
    );

GST_BOILERPLATE (GstRecombine, gst_bufresize, GstElement,
    GST_TYPE_ELEMENT);

/* set/get params */
static void gst_bufresize_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec);
static void gst_bufresize_get_property (GObject * object, guint prop_id,
    GValue * value, GParamSpec * pspec);

static gboolean gst_bufresize_set_caps (GstPad * pad, GstCaps * caps);
static GstFlowReturn gst_bufresize_chain (GstPad * pad, GstBuffer * buf);

/* GObject vmethod implementations */

static void
gst_bufresize_base_init (gpointer gclass)
{
  GstElementClass *element_class = GST_ELEMENT_CLASS (gclass);

  gst_element_class_set_details_simple(element_class,
    "Generic",
    "Recombine GST Buffers",
    "Recombine smaller GST buffers into larger ones",
    "Marc Leeman <marc.leeman@gmail.com>");

  gst_element_class_add_pad_template (element_class,
      gst_static_pad_template_get (&src_factory));
  gst_element_class_add_pad_template (element_class,
      gst_static_pad_template_get (&sink_factory));
}

/* initialize the plugin's class */
static void
gst_bufresize_class_init (GstRecombineClass * klass)
{
  GObjectClass *gobject_class;
  GstElementClass *gstelement_class;

  gobject_class = (GObjectClass *) klass;
  gstelement_class = (GstElementClass *) klass;

  gobject_class->set_property = gst_bufresize_set_property;
  gobject_class->get_property = gst_bufresize_get_property;

  g_object_class_install_property (gobject_class, 
      PROP_SIZE,
      g_param_spec_uint(
	      "size", 
	      "Size", 
	      "Maximum buffer size",
	      GST_RECOMBINE_MIN_BUFFER_SIZE,
	      GST_RECOMBINE_MAX_BUFFER_SIZE,
	      GST_RECOMBINE_DEFAULT_BUFFER_SIZE, 
	      G_PARAM_READWRITE));
}

/* initialize the new element
 * instantiate pads and add them to element
 * set pad calback functions
 * initialize instance structure
 */
static void
gst_bufresize_init (GstRecombine * filter,
    GstRecombineClass * gclass)
{
  filter->sinkpad = gst_pad_new_from_static_template (&sink_factory, "sink");
  gst_pad_set_setcaps_function (filter->sinkpad,
                                GST_DEBUG_FUNCPTR(gst_bufresize_set_caps));
  gst_pad_set_getcaps_function (filter->sinkpad,
                                GST_DEBUG_FUNCPTR(gst_pad_proxy_getcaps));
  gst_pad_set_chain_function (filter->sinkpad,
                              GST_DEBUG_FUNCPTR(gst_bufresize_chain));

  filter->srcpad = gst_pad_new_from_static_template (&src_factory, "src");
  gst_pad_set_getcaps_function (filter->srcpad,
                                GST_DEBUG_FUNCPTR(gst_pad_proxy_getcaps));

  gst_element_add_pad (GST_ELEMENT (filter), filter->sinkpad);
  gst_element_add_pad (GST_ELEMENT (filter), filter->srcpad);

  /* defaults */
  filter->size= GST_RECOMBINE_DEFAULT_BUFFER_SIZE;
  filter->combinebuf= NULL;
}

static void
gst_bufresize_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec)
{
  GstRecombine *filter = GST_RECOMBINE (object);

  switch (prop_id) {
    case PROP_SIZE:
      filter->size= g_value_get_uint (value);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}

static GstFlowReturn
gst_bufresize_chain (GstPad * pad, GstBuffer * buffer)
{
  GstRecombine *filter;
  GstBufferListIterator *it;
  guint size = 0u;
  GstFlowReturn ret = GST_FLOW_OK;

  filter = GST_RECOMBINE (GST_OBJECT_PARENT (pad));

  size = GST_BUFFER_SIZE (buffer);

  if(size > filter->size){
    GST_WARNING("Recombine size %u is smaller than %u buffer size, "
        "pushing buffer unmodified", filter->size, size);
    ret = gst_pad_push (filter->srcpad, buffer);
  }
  else{
    if(filter->combinebuf == NULL){

      GST_LOG("Init buffer.");

      /* no buffer yet */
      filter->combinebuf = gst_buffer_list_new();
      filter->it = gst_buffer_list_iterate (filter->combinebuf);
      gst_buffer_list_iterator_add_group (filter->it);
      gst_buffer_list_iterator_add (filter->it, buffer);

      filter->listsize = GST_BUFFER_SIZE(buffer);
    }
    else if((filter->listsize + size) <= filter->size){
      GST_LOG("Add buffer to list.");
      gst_buffer_list_iterator_add (filter->it, buffer);

      filter->listsize += GST_BUFFER_SIZE(buffer);
      GST_LOG("Done.");
    }
    else{
      /* no room to add current buffer, push stored one and start over
       * again */
      GST_LOG("Push buf with size %u.",
          filter->listsize);

      it = gst_buffer_list_iterate (filter->combinebuf);

      /* free old iterator */
      gst_buffer_list_iterator_free (filter->it);

      while (gst_buffer_list_iterator_next_group (it)) {
        ret = gst_pad_push (filter->srcpad, gst_buffer_list_iterator_merge_group(it));
      }

      gst_buffer_list_iterator_free (it);
      gst_buffer_list_unref (filter->combinebuf);

      filter->combinebuf = gst_buffer_list_new();
      filter->it = gst_buffer_list_iterate (filter->combinebuf);
      gst_buffer_list_iterator_add_group (filter->it);
      gst_buffer_list_iterator_add (filter->it, buffer);

      filter->listsize = GST_BUFFER_SIZE(buffer);
    }
  }

  return ret;
}

static void
gst_bufresize_get_property (GObject * object, guint prop_id,
    GValue * value, GParamSpec * pspec)
{
  GstRecombine *filter = GST_RECOMBINE (object);

  switch (prop_id) {
    case PROP_SIZE:
      g_value_set_uint(value, filter->size);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}

static gboolean
gst_bufresize_set_caps (GstPad * pad, GstCaps * caps)
{
  GstRecombine *filter;
  GstPad *otherpad;

  filter = GST_RECOMBINE (gst_pad_get_parent (pad));
  otherpad = (pad == filter->srcpad) ? filter->sinkpad : filter->srcpad;
  gst_object_unref (filter);

  return gst_pad_set_caps (otherpad, caps);
}

/* entry point to initialize the plug-in
 * initialize the plug-in itself
 * register the element factories and other features
 */
static gboolean
bufresize(GstPlugin * plugin)
{
  /* debug category for fltering log messages
   *
   * exchange the string 'Template plugin' with your description
   */
  GST_DEBUG_CATEGORY_INIT (
      gst_bufresize_debug, 
      "bufresize",
      0, 
      "GStreamer Buffer bufresize"
      );

  return gst_element_register (
      plugin, 
      "bufresize", 
      GST_RANK_NONE,
      GST_TYPE_RECOMBINE
      );
}

/* just in case */
#ifndef PACKAGE
#define PACKAGE "bufresize"
#endif

GST_PLUGIN_DEFINE (
    GST_VERSION_MAJOR,
    GST_VERSION_MINOR,
    "bufresize",
    "GStreamer Buffer Recombine",
    bufresize,
    VERSION,
    "GPL",
    "Barco",
    "http://www.barco.com/"
)
