/* GStreamer

 * Copyright (C) 2006 Thomas Vander Stichele <thomas at apestaart dot org>

 *

 * 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., 59 Temple Place - Suite 330,

 * Boston, MA 02111-1307, USA.

 */

/**

 * SECTION:element-gdppay

 * @see_also: gdpdepay

 *

 * This element payloads GStreamer buffers and events using the

 * GStreamer Data Protocol.

 * gst-launch -v -m videotestsrc num-buffers=50 ! gdppay ! filesink location=test.gdp

 * with the example shown in gdpdepay.

 * </refsect2>

 */

#ifdef HAVE_CONFIG_H

#include "config.h"

#endif

#include <gst/dataprotocol/dataprotocol.h>

#include "gstgdppay.h"

/* elementfactory information */

static const GstElementDetails gdp_pay_details =

GST_ELEMENT_DETAILS ("GDP Payloader",

    "GDP/Payloader",

    "Payloads GStreamer Data Protocol buffers",

    "Thomas Vander Stichele <thomas at apestaart dot org>");

static GstStaticPadTemplate gdp_pay_sink_template =

GST_STATIC_PAD_TEMPLATE ("sink",

    GST_PAD_SINK,

    GST_PAD_ALWAYS,

    GST_STATIC_CAPS_ANY);

static GstStaticPadTemplate gdp_pay_src_template =

GST_STATIC_PAD_TEMPLATE ("src",

    GST_PAD_SRC,

    GST_PAD_ALWAYS,

    GST_STATIC_CAPS ("application/x-gdp"));

GST_DEBUG_CATEGORY_STATIC (gst_gdp_pay_debug);

#define GST_CAT_DEFAULT gst_gdp_pay_debug

#define DEFAULT_CRC_HEADER TRUE

#define DEFAULT_CRC_PAYLOAD FALSE

#define DEFAULT_VERSION GST_DP_VERSION_1_0

enum

{

  PROP_0,

  PROP_CRC_HEADER,

  PROP_CRC_PAYLOAD,

  PROP_VERSION,

};

#define _do_init(x) \

    GST_DEBUG_CATEGORY_INIT (gst_gdp_pay_debug, "gdppay", 0, \

    "GDP payloader");

GST_BOILERPLATE_FULL (GstGDPPay, gst_gdp_pay, GstElement,

    GST_TYPE_ELEMENT, _do_init);

static void gst_gdp_pay_reset (GstGDPPay * this);

static GstFlowReturn gst_gdp_pay_chain (GstPad * pad, GstBuffer * buffer);

static gboolean gst_gdp_pay_src_event (GstPad * pad, GstEvent * event);

static gboolean gst_gdp_pay_sink_event (GstPad * pad, GstEvent * event);

static GstStateChangeReturn gst_gdp_pay_change_state (GstElement *

    element, GstStateChange transition);

static void gst_gdp_pay_set_property (GObject * object, guint prop_id,

    const GValue * value, GParamSpec * pspec);

static void gst_gdp_pay_get_property (GObject * object, guint prop_id,

    GValue * value, GParamSpec * pspec);

static void gst_gdp_pay_finalize (GObject * gobject);

static void

gst_gdp_pay_base_init (gpointer g_class)

{

  GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);

  gst_element_class_set_details (element_class, &gdp_pay_details);

  gst_element_class_add_pad_template (element_class,

      gst_static_pad_template_get (&gdp_pay_sink_template));

  gst_element_class_add_pad_template (element_class,

      gst_static_pad_template_get (&gdp_pay_src_template));

}

static void

gst_gdp_pay_class_init (GstGDPPayClass * klass)

{

  GObjectClass *gobject_class;

  GstElementClass *gstelement_class;

  gobject_class = (GObjectClass *) klass;

  gstelement_class = (GstElementClass *) klass;

  gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_gdp_pay_set_property);

  gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_gdp_pay_get_property);

  gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_gdp_pay_finalize);

  g_object_class_install_property (gobject_class, PROP_CRC_HEADER,

      g_param_spec_boolean ("crc-header", "CRC Header",

          "Calculate and store a CRC checksum on the header",

  g_object_class_install_property (gobject_class, PROP_CRC_PAYLOAD,

      g_param_spec_boolean ("crc-payload", "CRC Payload",

          "Calculate and store a CRC checksum on the payload",

  g_object_class_install_property (gobject_class, PROP_VERSION,

      g_param_spec_enum ("version", "Version",

          "Version of the GStreamer Data Protocol",

  gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_gdp_pay_change_state);

}

static void

gst_gdp_pay_init (GstGDPPay * gdppay, GstGDPPayClass * g_class)

{

  gdppay->sinkpad =

      gst_pad_new_from_static_template (&gdp_pay_sink_template, "sink");

  gst_pad_set_chain_function (gdppay->sinkpad,

      GST_DEBUG_FUNCPTR (gst_gdp_pay_chain));

  gst_pad_set_event_function (gdppay->sinkpad,

      GST_DEBUG_FUNCPTR (gst_gdp_pay_sink_event));

  gst_element_add_pad (GST_ELEMENT (gdppay), gdppay->sinkpad);

  gdppay->srcpad =

      gst_pad_new_from_static_template (&gdp_pay_src_template, "src");

  gst_pad_set_event_function (gdppay->srcpad,

      GST_DEBUG_FUNCPTR (gst_gdp_pay_src_event));

  gst_element_add_pad (GST_ELEMENT (gdppay), gdppay->srcpad);

  gdppay->crc_header = DEFAULT_CRC_HEADER;

  gdppay->crc_payload = DEFAULT_CRC_PAYLOAD;

  gdppay->header_flag = gdppay->crc_header | gdppay->crc_payload;

  gdppay->version = DEFAULT_VERSION;

  gdppay->offset = 0;

  gdppay->packetizer = gst_dp_packetizer_new (gdppay->version);

}

static void

gst_gdp_pay_finalize (GObject * gobject)

{

  GstGDPPay *this = GST_GDP_PAY (gobject);

  gst_gdp_pay_reset (this);

  gst_dp_packetizer_free (this->packetizer);

  GST_CALL_PARENT (G_OBJECT_CLASS, finalize, (gobject));

}

static void

gst_gdp_pay_reset (GstGDPPay * this)

{

  GST_DEBUG_OBJECT (this, "Resetting GDP object");

  /* clear the queued buffers */

  while (this->queue) {

    GstBuffer *buffer;

    buffer = GST_BUFFER_CAST (this->queue->data);

    /* delete buffer from queue now */

    this->queue = g_list_delete_link (this->queue, this->queue);

  }

  if (this->caps) {

    gst_caps_unref (this->caps);

    this->caps = NULL;

  }

  if (this->caps_buf) {

    gst_buffer_unref (this->caps_buf);

    this->caps_buf = NULL;

  }

  if (this->tag_buf) {

    gst_buffer_unref (this->tag_buf);

    this->tag_buf = NULL;

  }

  if (this->new_segment_buf) {

    gst_buffer_unref (this->new_segment_buf);

    this->new_segment_buf = NULL;

  }

  this->sent_streamheader = FALSE;

  this->offset = 0;

}

/* set OFFSET and OFFSET_END with running count */

static void

gst_gdp_stamp_buffer (GstGDPPay * this, GstBuffer * buffer)

{

  GST_BUFFER_OFFSET (buffer) = this->offset;

  GST_BUFFER_OFFSET_END (buffer) = this->offset + GST_BUFFER_SIZE (buffer);

  this->offset = GST_BUFFER_OFFSET_END (buffer);

}

static GstBuffer *

gst_gdp_buffer_from_caps (GstGDPPay * this, GstCaps * caps)

{

  GstBuffer *headerbuf;

  GstBuffer *payloadbuf;

  guint8 *header, *payload;

  guint len;

  if (!this->packetizer->packet_from_caps (caps, this->header_flag, &len,

          &header, &payload))

    goto packet_failed;

  GST_LOG_OBJECT (this, "creating GDP header and payload buffer from caps");

  headerbuf = gst_buffer_new ();

  gst_buffer_set_data (headerbuf, header, len);

  GST_BUFFER_MALLOCDATA (headerbuf) = header;

  payloadbuf = gst_buffer_new ();

  gst_buffer_set_data (payloadbuf, payload,

      gst_dp_header_payload_length (header));

  GST_BUFFER_MALLOCDATA (payloadbuf) = payload;

  return gst_buffer_join (headerbuf, payloadbuf);

  /* ERRORS */

packet_failed:

  {

    GST_WARNING_OBJECT (this, "could not create GDP header from caps");

    return NULL;

  }

}

static GstBuffer *

gst_gdp_pay_buffer_from_buffer (GstGDPPay * this, GstBuffer * buffer)

{

  GstBuffer *headerbuf;

  guint8 *header;

  guint len;

  if (!this->packetizer->header_from_buffer (buffer, this->header_flag, &len,

          &header))

    goto no_buffer;

  GST_LOG_OBJECT (this, "creating GDP header and payload buffer from buffer");

  headerbuf = gst_buffer_new ();

  gst_buffer_set_data (headerbuf, header, len);

  GST_BUFFER_MALLOCDATA (headerbuf) = header;

  /* we do not want to lose the ref on the incoming buffer */

  gst_buffer_ref (buffer);

  return gst_buffer_join (headerbuf, buffer);

  /* ERRORS */

no_buffer:

  {

    GST_WARNING_OBJECT (this, "could not create GDP header from buffer");

    return NULL;

  }

}

static GstBuffer *

gst_gdp_buffer_from_event (GstGDPPay * this, GstEvent * event)

{

  GstBuffer *headerbuf;

  GstBuffer *payloadbuf;

  guint8 *header, *payload;

  guint len;

  gboolean ret;

  ret =

      this->packetizer->packet_from_event (event, this->header_flag, &len,

      &header, &payload);

  if (!ret)

    goto no_event;

  GST_LOG_OBJECT (this, "creating GDP header and payload buffer from event");

  headerbuf = gst_buffer_new ();

  gst_buffer_set_data (headerbuf, header, len);

  GST_BUFFER_MALLOCDATA (headerbuf) = header;

  payloadbuf = gst_buffer_new ();

  gst_buffer_set_data (payloadbuf, payload,

      gst_dp_header_payload_length (header));

  GST_BUFFER_MALLOCDATA (payloadbuf) = payload;

  return gst_buffer_join (headerbuf, payloadbuf);

  /* ERRORS */

no_event:

  {

    GST_WARNING_OBJECT (this, "could not create GDP header from event %s (%d)",

        gst_event_type_get_name (event->type), event->type);

    return NULL;

  }

}

/* set our caps with streamheader, based on the latest newsegment and caps,

 * and (possibly) GDP-serialized buffers of the streamheaders on the src pad */

static GstFlowReturn

gst_gdp_pay_reset_streamheader (GstGDPPay * this)

{

  GstCaps *caps;

  /* We use copies of these to avoid circular refcounts */

  GstBuffer *new_segment_buf, *caps_buf, *tag_buf;

  GstStructure *structure;

  GstFlowReturn r = GST_FLOW_OK;

  gboolean version_one_zero = TRUE;

  GValue array = { 0 };

  GValue value = { 0 };

  GST_DEBUG_OBJECT (this, "start");

  /* In version 0.2, we didn't need or send new segment or tags */

  if (this->version == GST_DP_VERSION_0_2)

    version_one_zero = FALSE;

  if (version_one_zero) {

    if (!this->new_segment_buf || !this->caps_buf) {

      GST_DEBUG_OBJECT (this, "1.0, missing new_segment or caps, returning");

      return GST_FLOW_OK;

    }

  } else {

    if (!this->caps_buf) {

      GST_DEBUG_OBJECT (this, "0.2, missing caps, returning");

      return GST_FLOW_OK;

    }

  }

  /* put copies of the buffers in a fixed list

   * Stamp the buffers with offset and offset_end as well.

   * We do this here so the offsets match the order the buffers go out in */

  g_value_init (&array, GST_TYPE_ARRAY);

  if (version_one_zero) {

    gst_gdp_stamp_buffer (this, this->new_segment_buf);

    GST_DEBUG_OBJECT (this, "1.0, appending copy of new segment buffer %p",

        this->new_segment_buf);

    new_segment_buf = gst_buffer_copy (this->new_segment_buf);

    g_value_init (&value, GST_TYPE_BUFFER);

    gst_value_set_buffer (&value, new_segment_buf);

    gst_value_array_append_value (&array, &value);

    g_value_unset (&value);

    gst_buffer_unref (new_segment_buf);

    if (this->tag_buf) {

      gst_gdp_stamp_buffer (this, this->tag_buf);

      GST_DEBUG_OBJECT (this, "1.0, appending copy of tag buffer %p",

          this->tag_buf);

      tag_buf = gst_buffer_copy (this->tag_buf);

      g_value_init (&value, GST_TYPE_BUFFER);

      gst_value_set_buffer (&value, tag_buf);

      gst_value_array_append_value (&array, &value);

      g_value_unset (&value);

      gst_buffer_unref (tag_buf);

    }

  }

  gst_gdp_stamp_buffer (this, this->caps_buf);

  GST_DEBUG_OBJECT (this, "appending copy of caps buffer %p", this->caps_buf);

  caps_buf = gst_buffer_copy (this->caps_buf);

  g_value_init (&value, GST_TYPE_BUFFER);

  gst_value_set_buffer (&value, caps_buf);

  gst_value_array_append_value (&array, &value);

  g_value_unset (&value);

  gst_buffer_unref (caps_buf);

  /* we also need to add GDP serializations of the streamheaders of the

   * incoming caps */

  structure = gst_caps_get_structure (this->caps, 0);

  if (gst_structure_has_field (structure, "streamheader")) {

    const GValue *sh;

    GArray *buffers;

    GstBuffer *buffer;

    int i;

    sh = gst_structure_get_value (structure, "streamheader");

    buffers = g_value_peek_pointer (sh);

    GST_DEBUG_OBJECT (this,

        "Need to serialize %d incoming streamheader buffers on ours",

        buffers->len);

    for (i = 0; i < buffers->len; ++i) {

      GValue *bufval;

      GstBuffer *outbuffer;

      bufval = &g_array_index (buffers, GValue, i);

      buffer = g_value_peek_pointer (bufval);

      outbuffer = gst_gdp_pay_buffer_from_buffer (this, buffer);

      if (!outbuffer) {

        g_value_unset (&array);

        goto no_buffer;

      }

      g_value_init (&value, GST_TYPE_BUFFER);

      gst_value_set_buffer (&value, outbuffer);

      gst_value_array_append_value (&array, &value);

      g_value_unset (&value);

      gst_buffer_unref (outbuffer);

    }

  } else {

    GST_DEBUG_OBJECT (this, "no streamheader to serialize");

  }

  GST_DEBUG_OBJECT (this, "%d serialized buffers on streamheaders",

      gst_value_array_get_size (&array));

  caps = gst_caps_from_string ("application/x-gdp");

  structure = gst_caps_get_structure (caps, 0);

  gst_structure_set_value (structure, "streamheader", &array);

  g_value_unset (&array);

  GST_DEBUG_OBJECT (this, "Setting caps on src pad %" GST_PTR_FORMAT, caps);

  gst_pad_set_caps (this->srcpad, caps);

  gst_buffer_set_caps (this->caps_buf, caps);

  gst_buffer_set_caps (this->new_segment_buf, caps);

  /* if these are our first ever buffers, send out new_segment first */

  if (!this->sent_streamheader) {

    GstEvent *event =

        gst_event_new_new_segment (TRUE, 1.0, GST_FORMAT_BYTES, 0, -1, 0);

    GST_DEBUG_OBJECT (this, "Sending out new_segment event %p", event);

    if (!gst_pad_push_event (this->srcpad, event)) {

      GST_WARNING_OBJECT (this, "pushing new segment failed");

      r = GST_FLOW_ERROR;

      goto done;

    }

  }

  /* push out these streamheader buffers, then flush our internal queue */

  GST_DEBUG_OBJECT (this, "Pushing GDP new_segment buffer %p with offset %"

      G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT, this->new_segment_buf,

      GST_BUFFER_OFFSET (this->new_segment_buf),

      GST_BUFFER_OFFSET_END (this->new_segment_buf));

  /* we stored these bufs with refcount 1, so make sure we keep a ref */

  r = gst_pad_push (this->srcpad, gst_buffer_ref (this->new_segment_buf));

  if (r != GST_FLOW_OK) {

    GST_WARNING_OBJECT (this, "pushing GDP newsegment buffer returned %d", r);

    goto done;

  }

  if (this->tag_buf) {

    GST_DEBUG_OBJECT (this, "Pushing GDP tag buffer %p", this->tag_buf);

    /* we stored these bufs with refcount 1, so make sure we keep a ref */

    r = gst_pad_push (this->srcpad, gst_buffer_ref (this->tag_buf));

    if (r != GST_FLOW_OK) {

      GST_WARNING_OBJECT (this, "pushing GDP tag buffer returned %d", r);

      goto done;

    }

  }

  GST_DEBUG_OBJECT (this, "Pushing GDP caps buffer %p", this->caps_buf);

  r = gst_pad_push (this->srcpad, gst_buffer_ref (this->caps_buf));

  if (r != GST_FLOW_OK) {

    GST_WARNING_OBJECT (this, "pushing GDP caps buffer returned %d", r);

    goto done;

  }

  this->sent_streamheader = TRUE;

  GST_DEBUG_OBJECT (this, "need to push %d queued buffers",

      g_list_length (this->queue));

  while (this->queue) {

    GstBuffer *buffer;

    buffer = GST_BUFFER_CAST (this->queue->data);

    GST_DEBUG_OBJECT (this, "Pushing queued GDP buffer %p", buffer);

    /* delete buffer from queue now */

    this->queue = g_list_delete_link (this->queue, this->queue);

    /* set caps and push */

    gst_buffer_set_caps (buffer, caps);

    r = gst_pad_push (this->srcpad, buffer);

    if (r != GST_FLOW_OK) {

      GST_WARNING_OBJECT (this, "pushing queued GDP buffer returned %d", r);

      goto done;

    }

  }

done:

  gst_caps_unref (caps);

  GST_DEBUG_OBJECT (this, "stop");

  return r;

  /* ERRORS */

no_buffer:

  {

    GST_ELEMENT_ERROR (this, STREAM, FORMAT, (NULL),

        ("failed to create GDP buffer from streamheader"));

    return GST_FLOW_ERROR;

  }

}

/* queue a buffer internally if we haven't sent streamheader buffers yet;

 * otherwise, just push on, this takes ownership of the buffer. */

static GstFlowReturn

gst_gdp_queue_buffer (GstGDPPay * this, GstBuffer * buffer)

{

  if (this->sent_streamheader) {

    GST_LOG_OBJECT (this, "Pushing GDP buffer %p, caps %" GST_PTR_FORMAT,

        buffer, this->caps);

    return gst_pad_push (this->srcpad, buffer);