gst/gdp/gstgdppay.c

   1 /* GStreamer
   2  * Copyright (C) 2006 Thomas Vander Stichele <thomas at apestaart dot org>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  17  * Boston, MA 02111-1307, USA.
  18  */
  19
  20 /**
  21  * SECTION:element-gdppay
  22  * @see_also: gdpdepay
  23  *
  24  * <refsect2>
  25  * <para>
  26  * This element payloads GStreamer buffers and events using the
  27  * GStreamer Data Protocol.
  28  * </para>
  29  * <para>
  30  * <programlisting>
  31  * gst-launch -v -m videotestsrc num-buffers=50 ! gdppay ! filesink location=test.gdp
  32  * </programlisting>
  33  * This pipeline creates a serialized video stream that can be played back
  34  * with the example shown in gdpdepay.
  35  * </para>
  36  * </refsect2>
  37  */
  38
  39 #ifdef HAVE_CONFIG_H
  40 #include "config.h"
  41 #endif
  42
  43 #include <gst/dataprotocol/dataprotocol.h>
  44
  45 #include "gstgdppay.h"
  46
  47 /* elementfactory information */
  48 static const GstElementDetails gdp_pay_details =
  49 GST_ELEMENT_DETAILS ("GDP Payloader",
  50     "GDP/Payloader",
  51     "Payloads GStreamer Data Protocol buffers",
  52     "Thomas Vander Stichele <thomas at apestaart dot org>");
  53
  54 static GstStaticPadTemplate gdp_pay_sink_template =
  55 GST_STATIC_PAD_TEMPLATE ("sink",
  56     GST_PAD_SINK,
  57     GST_PAD_ALWAYS,
  58     GST_STATIC_CAPS_ANY);
  59
  60 static GstStaticPadTemplate gdp_pay_src_template =
  61 GST_STATIC_PAD_TEMPLATE ("src",
  62     GST_PAD_SRC,
  63     GST_PAD_ALWAYS,
  64     GST_STATIC_CAPS ("application/x-gdp"));
  65
  66 GST_DEBUG_CATEGORY_STATIC (gst_gdp_pay_debug);
  67 #define GST_CAT_DEFAULT gst_gdp_pay_debug
  68
  69 #define DEFAULT_CRC_HEADER TRUE
  70 #define DEFAULT_CRC_PAYLOAD FALSE
  71 #define DEFAULT_VERSION GST_DP_VERSION_1_0
  72
  73 enum
  74 {
  75   PROP_0,
  76   PROP_CRC_HEADER,
  77   PROP_CRC_PAYLOAD,
  78   PROP_VERSION,
  79 };
  80
  81 #define _do_init(x) \
  82     GST_DEBUG_CATEGORY_INIT (gst_gdp_pay_debug, "gdppay", 0, \
  83     "GDP payloader");
  84
  85 GST_BOILERPLATE_FULL (GstGDPPay, gst_gdp_pay, GstElement,
  86     GST_TYPE_ELEMENT, _do_init);
  87
  88 static void gst_gdp_pay_reset (GstGDPPay * this);
  89 static GstFlowReturn gst_gdp_pay_chain (GstPad * pad, GstBuffer * buffer);
  90
  91 static gboolean gst_gdp_pay_src_event (GstPad * pad, GstEvent * event);
  92 static gboolean gst_gdp_pay_sink_event (GstPad * pad, GstEvent * event);
  93
  94 static GstStateChangeReturn gst_gdp_pay_change_state (GstElement *
  95     element, GstStateChange transition);
  96
  97 static void gst_gdp_pay_set_property (GObject * object, guint prop_id,
  98     const GValue * value, GParamSpec * pspec);
  99 static void gst_gdp_pay_get_property (GObject * object, guint prop_id,
 100     GValue * value, GParamSpec * pspec);
 101
 102 static void gst_gdp_pay_finalize (GObject * gobject);
 103
 104 static void
 105 gst_gdp_pay_base_init (gpointer g_class)
 106 {
 107   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
 108
 109   gst_element_class_set_details (element_class, &gdp_pay_details);
 110
 111   gst_element_class_add_pad_template (element_class,
 112       gst_static_pad_template_get (&gdp_pay_sink_template));
 113   gst_element_class_add_pad_template (element_class,
 114       gst_static_pad_template_get (&gdp_pay_src_template));
 115 }
 116
 117 static void
 118 gst_gdp_pay_class_init (GstGDPPayClass * klass)
 119 {
 120   GObjectClass *gobject_class;
 121   GstElementClass *gstelement_class;
 122
 123   gobject_class = (GObjectClass *) klass;
 124   gstelement_class = (GstElementClass *) klass;
 125
 126   gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_gdp_pay_set_property);
 127   gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_gdp_pay_get_property);
 128   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_gdp_pay_finalize);
 129
 130   g_object_class_install_property (gobject_class, PROP_CRC_HEADER,
 131       g_param_spec_boolean ("crc-header", "CRC Header",
 132           "Calculate and store a CRC checksum on the header",
 133           DEFAULT_CRC_HEADER, G_PARAM_READWRITE));
 134   g_object_class_install_property (gobject_class, PROP_CRC_PAYLOAD,
 135       g_param_spec_boolean ("crc-payload", "CRC Payload",
 136           "Calculate and store a CRC checksum on the payload",
 137           DEFAULT_CRC_PAYLOAD, G_PARAM_READWRITE));
 138   g_object_class_install_property (gobject_class, PROP_VERSION,
 139       g_param_spec_enum ("version", "Version",
 140           "Version of the GStreamer Data Protocol",
 141           GST_TYPE_DP_VERSION, DEFAULT_VERSION, G_PARAM_READWRITE));
 142
 143   gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_gdp_pay_change_state);
 144 }
 145
 146 static void
 147 gst_gdp_pay_init (GstGDPPay * gdppay, GstGDPPayClass * g_class)
 148 {
 149   gdppay->sinkpad =
 150       gst_pad_new_from_static_template (&gdp_pay_sink_template, "sink");
 151   gst_pad_set_chain_function (gdppay->sinkpad,
 152       GST_DEBUG_FUNCPTR (gst_gdp_pay_chain));
 153   gst_pad_set_event_function (gdppay->sinkpad,
 154       GST_DEBUG_FUNCPTR (gst_gdp_pay_sink_event));
 155   gst_element_add_pad (GST_ELEMENT (gdppay), gdppay->sinkpad);
 156
 157   gdppay->srcpad =
 158       gst_pad_new_from_static_template (&gdp_pay_src_template, "src");
 159   gst_pad_set_event_function (gdppay->srcpad,
 160       GST_DEBUG_FUNCPTR (gst_gdp_pay_src_event));
 161   gst_element_add_pad (GST_ELEMENT (gdppay), gdppay->srcpad);
 162
 163   gdppay->crc_header = DEFAULT_CRC_HEADER;
 164   gdppay->crc_payload = DEFAULT_CRC_PAYLOAD;
 165   gdppay->header_flag = gdppay->crc_header | gdppay->crc_payload;
 166   gdppay->version = DEFAULT_VERSION;
 167   gdppay->offset = 0;
 168
 169   gdppay->packetizer = gst_dp_packetizer_new (gdppay->version);
 170 }
 171
 172 static void
 173 gst_gdp_pay_finalize (GObject * gobject)
 174 {
 175   GstGDPPay *this = GST_GDP_PAY (gobject);
 176
 177   gst_gdp_pay_reset (this);
 178   gst_dp_packetizer_free (this->packetizer);
 179
 180   GST_CALL_PARENT (G_OBJECT_CLASS, finalize, (gobject));
 181 }
 182
 183 static void
 184 gst_gdp_pay_reset (GstGDPPay * this)
 185 {
 186   GST_DEBUG_OBJECT (this, "Resetting GDP object");
 187   /* clear the queued buffers */
 188   while (this->queue) {
 189     GstBuffer *buffer;
 190
 191     buffer = GST_BUFFER_CAST (this->queue->data);
 192     GST_DEBUG_OBJECT (this, "Pushing queued GDP buffer %p", buffer);
 193
 194     /* delete buffer from queue now */
 195     this->queue = g_list_delete_link (this->queue, this->queue);
 196   }
 197   if (this->caps) {
 198     gst_caps_unref (this->caps);
 199     this->caps = NULL;
 200   }
 201   if (this->caps_buf) {
 202     gst_buffer_unref (this->caps_buf);
 203     this->caps_buf = NULL;
 204   }
 205   if (this->tag_buf) {
 206     gst_buffer_unref (this->tag_buf);
 207     this->tag_buf = NULL;
 208   }
 209   if (this->new_segment_buf) {
 210     gst_buffer_unref (this->new_segment_buf);
 211     this->new_segment_buf = NULL;
 212   }
 213   this->sent_streamheader = FALSE;
 214   this->offset = 0;
 215 }
 216
 217 /* set OFFSET and OFFSET_END with running count */
 218 static void
 219 gst_gdp_stamp_buffer (GstGDPPay * this, GstBuffer * buffer)
 220 {
 221   GST_BUFFER_OFFSET (buffer) = this->offset;
 222   GST_BUFFER_OFFSET_END (buffer) = this->offset + GST_BUFFER_SIZE (buffer);
 223   this->offset = GST_BUFFER_OFFSET_END (buffer);
 224 }
 225
 226 static GstBuffer *
 227 gst_gdp_buffer_from_caps (GstGDPPay * this, GstCaps * caps)
 228 {
 229   GstBuffer *headerbuf;
 230   GstBuffer *payloadbuf;
 231   guint8 *header, *payload;
 232   guint len;
 233
 234   if (!this->packetizer->packet_from_caps (caps, this->header_flag, &len,
 235           &header, &payload))
 236     goto packet_failed;
 237
 238   GST_LOG_OBJECT (this, "creating GDP header and payload buffer from caps");
 239   headerbuf = gst_buffer_new ();
 240   gst_buffer_set_data (headerbuf, header, len);
 241   GST_BUFFER_MALLOCDATA (headerbuf) = header;
 242
 243   payloadbuf = gst_buffer_new ();
 244   gst_buffer_set_data (payloadbuf, payload,
 245       gst_dp_header_payload_length (header));
 246   GST_BUFFER_MALLOCDATA (payloadbuf) = payload;
 247
 248   return gst_buffer_join (headerbuf, payloadbuf);
 249
 250   /* ERRORS */
 251 packet_failed:
 252   {
 253     GST_WARNING_OBJECT (this, "could not create GDP header from caps");
 254     return NULL;
 255   }
 256 }
 257
 258 static GstBuffer *
 259 gst_gdp_pay_buffer_from_buffer (GstGDPPay * this, GstBuffer * buffer)
 260 {
 261   GstBuffer *headerbuf;
 262   guint8 *header;
 263   guint len;
 264
 265   if (!this->packetizer->header_from_buffer (buffer, this->header_flag, &len,
 266           &header))
 267     goto no_buffer;
 268
 269   GST_LOG_OBJECT (this, "creating GDP header and payload buffer from buffer");
 270   headerbuf = gst_buffer_new ();
 271   gst_buffer_set_data (headerbuf, header, len);
 272   GST_BUFFER_MALLOCDATA (headerbuf) = header;
 273
 274   /* we do not want to lose the ref on the incoming buffer */
 275   gst_buffer_ref (buffer);
 276
 277   return gst_buffer_join (headerbuf, buffer);
 278
 279   /* ERRORS */
 280 no_buffer:
 281   {
 282     GST_WARNING_OBJECT (this, "could not create GDP header from buffer");
 283     return NULL;
 284   }
 285 }
 286
 287 static GstBuffer *
 288 gst_gdp_buffer_from_event (GstGDPPay * this, GstEvent * event)
 289 {
 290   GstBuffer *headerbuf;
 291   GstBuffer *payloadbuf;
 292   guint8 *header, *payload;
 293   guint len;
 294   gboolean ret;
 295
 296   ret =
 297       this->packetizer->packet_from_event (event, this->header_flag, &len,
 298       &header, &payload);
 299   if (!ret)
 300     goto no_event;
 301
 302   GST_LOG_OBJECT (this, "creating GDP header and payload buffer from event");
 303   headerbuf = gst_buffer_new ();
 304   gst_buffer_set_data (headerbuf, header, len);
 305   GST_BUFFER_MALLOCDATA (headerbuf) = header;
 306
 307   payloadbuf = gst_buffer_new ();
 308   gst_buffer_set_data (payloadbuf, payload,
 309       gst_dp_header_payload_length (header));
 310   GST_BUFFER_MALLOCDATA (payloadbuf) = payload;
 311
 312   return gst_buffer_join (headerbuf, payloadbuf);
 313
 314   /* ERRORS */
 315 no_event:
 316   {
 317     GST_WARNING_OBJECT (this, "could not create GDP header from event %s (%d)",
 318         gst_event_type_get_name (event->type), event->type);
 319     return NULL;
 320   }
 321 }
 322
 323
 324 /* set our caps with streamheader, based on the latest newsegment and caps,
 325  * and (possibly) GDP-serialized buffers of the streamheaders on the src pad */
 326 static GstFlowReturn
 327 gst_gdp_pay_reset_streamheader (GstGDPPay * this)
 328 {
 329   GstCaps *caps;
 330
 331   /* We use copies of these to avoid circular refcounts */
 332   GstBuffer *new_segment_buf, *caps_buf, *tag_buf;
 333   GstStructure *structure;
 334   GstFlowReturn r = GST_FLOW_OK;
 335   gboolean version_one_zero = TRUE;
 336
 337   GValue array = { 0 };
 338   GValue value = { 0 };
 339
 340   GST_DEBUG_OBJECT (this, "start");
 341   /* In version 0.2, we didn't need or send new segment or tags */
 342   if (this->version == GST_DP_VERSION_0_2)
 343     version_one_zero = FALSE;
 344
 345   if (version_one_zero) {
 346     if (!this->new_segment_buf || !this->caps_buf) {
 347       GST_DEBUG_OBJECT (this, "1.0, missing new_segment or caps, returning");
 348       return GST_FLOW_OK;
 349     }
 350   } else {
 351     if (!this->caps_buf) {
 352       GST_DEBUG_OBJECT (this, "0.2, missing caps, returning");
 353       return GST_FLOW_OK;
 354     }
 355   }
 356
 357   /* put copies of the buffers in a fixed list
 358    * Stamp the buffers with offset and offset_end as well.
 359    * We do this here so the offsets match the order the buffers go out in */
 360   g_value_init (&array, GST_TYPE_ARRAY);
 361
 362   if (version_one_zero) {
 363     gst_gdp_stamp_buffer (this, this->new_segment_buf);
 364     GST_DEBUG_OBJECT (this, "1.0, appending copy of new segment buffer %p",
 365         this->new_segment_buf);
 366     new_segment_buf = gst_buffer_copy (this->new_segment_buf);
 367     g_value_init (&value, GST_TYPE_BUFFER);
 368     gst_value_set_buffer (&value, new_segment_buf);
 369     gst_value_array_append_value (&array, &value);
 370     g_value_unset (&value);
 371     gst_buffer_unref (new_segment_buf);
 372
 373     if (this->tag_buf) {
 374       gst_gdp_stamp_buffer (this, this->tag_buf);
 375       GST_DEBUG_OBJECT (this, "1.0, appending copy of tag buffer %p",
 376           this->tag_buf);
 377       tag_buf = gst_buffer_copy (this->tag_buf);
 378       g_value_init (&value, GST_TYPE_BUFFER);
 379       gst_value_set_buffer (&value, tag_buf);
 380       gst_value_array_append_value (&array, &value);
 381       g_value_unset (&value);
 382       gst_buffer_unref (tag_buf);
 383     }
 384   }
 385
 386   gst_gdp_stamp_buffer (this, this->caps_buf);
 387   GST_DEBUG_OBJECT (this, "appending copy of caps buffer %p", this->caps_buf);
 388   caps_buf = gst_buffer_copy (this->caps_buf);
 389   g_value_init (&value, GST_TYPE_BUFFER);
 390   gst_value_set_buffer (&value, caps_buf);
 391   gst_value_array_append_value (&array, &value);
 392   g_value_unset (&value);
 393   gst_buffer_unref (caps_buf);
 394
 395   /* we also need to add GDP serializations of the streamheaders of the
 396    * incoming caps */
 397   structure = gst_caps_get_structure (this->caps, 0);
 398   if (gst_structure_has_field (structure, "streamheader")) {
 399     const GValue *sh;
 400     GArray *buffers;
 401     GstBuffer *buffer;
 402     int i;
 403
 404     sh = gst_structure_get_value (structure, "streamheader");
 405     buffers = g_value_peek_pointer (sh);
 406     GST_DEBUG_OBJECT (this,
 407         "Need to serialize %d incoming streamheader buffers on ours",
 408         buffers->len);
 409     for (i = 0; i < buffers->len; ++i) {
 410       GValue *bufval;
 411       GstBuffer *outbuffer;
 412
 413       bufval = &g_array_index (buffers, GValue, i);
 414       buffer = g_value_peek_pointer (bufval);
 415       outbuffer = gst_gdp_pay_buffer_from_buffer (this, buffer);
 416       if (!outbuffer) {
 417         g_value_unset (&array);
 418         goto no_buffer;
 419       }
 420
 421       g_value_init (&value, GST_TYPE_BUFFER);
 422       gst_value_set_buffer (&value, outbuffer);
 423       gst_value_array_append_value (&array, &value);
 424       g_value_unset (&value);
 425
 426       gst_buffer_unref (outbuffer);
 427     }
 428   } else {
 429     GST_DEBUG_OBJECT (this, "no streamheader to serialize");
 430   }
 431
 432   GST_DEBUG_OBJECT (this, "%d serialized buffers on streamheaders",
 433       gst_value_array_get_size (&array));
 434   caps = gst_caps_from_string ("application/x-gdp");
 435   structure = gst_caps_get_structure (caps, 0);
 436
 437   gst_structure_set_value (structure, "streamheader", &array);
 438   g_value_unset (&array);
 439
 440   GST_DEBUG_OBJECT (this, "Setting caps on src pad %" GST_PTR_FORMAT, caps);
 441   gst_pad_set_caps (this->srcpad, caps);
 442   gst_buffer_set_caps (this->caps_buf, caps);
 443   gst_buffer_set_caps (this->new_segment_buf, caps);
 444
 445   /* if these are our first ever buffers, send out new_segment first */
 446   if (!this->sent_streamheader) {
 447     GstEvent *event =
 448         gst_event_new_new_segment (TRUE, 1.0, GST_FORMAT_BYTES, 0, -1, 0);
 449     GST_DEBUG_OBJECT (this, "Sending out new_segment event %p", event);
 450     if (!gst_pad_push_event (this->srcpad, event)) {
 451       GST_WARNING_OBJECT (this, "pushing new segment failed");
 452       r = GST_FLOW_ERROR;
 453       goto done;
 454     }
 455   }
 456
 457   /* push out these streamheader buffers, then flush our internal queue */
 458   GST_DEBUG_OBJECT (this, "Pushing GDP new_segment buffer %p with offset %"
 459       G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT, this->new_segment_buf,
 460       GST_BUFFER_OFFSET (this->new_segment_buf),
 461       GST_BUFFER_OFFSET_END (this->new_segment_buf));
 462   /* we stored these bufs with refcount 1, so make sure we keep a ref */
 463   r = gst_pad_push (this->srcpad, gst_buffer_ref (this->new_segment_buf));
 464   if (r != GST_FLOW_OK) {
 465     GST_WARNING_OBJECT (this, "pushing GDP newsegment buffer returned %d", r);
 466     goto done;
 467   }
 468   if (this->tag_buf) {
 469     GST_DEBUG_OBJECT (this, "Pushing GDP tag buffer %p", this->tag_buf);
 470     /* we stored these bufs with refcount 1, so make sure we keep a ref */
 471     r = gst_pad_push (this->srcpad, gst_buffer_ref (this->tag_buf));
 472     if (r != GST_FLOW_OK) {
 473       GST_WARNING_OBJECT (this, "pushing GDP tag buffer returned %d", r);
 474       goto done;
 475     }
 476   }
 477   GST_DEBUG_OBJECT (this, "Pushing GDP caps buffer %p", this->caps_buf);
 478   r = gst_pad_push (this->srcpad, gst_buffer_ref (this->caps_buf));
 479   if (r != GST_FLOW_OK) {
 480     GST_WARNING_OBJECT (this, "pushing GDP caps buffer returned %d", r);
 481     goto done;
 482   }
 483   this->sent_streamheader = TRUE;
 484   GST_DEBUG_OBJECT (this, "need to push %d queued buffers",
 485       g_list_length (this->queue));
 486   while (this->queue) {
 487     GstBuffer *buffer;
 488
 489     buffer = GST_BUFFER_CAST (this->queue->data);
 490     GST_DEBUG_OBJECT (this, "Pushing queued GDP buffer %p", buffer);
 491
 492     /* delete buffer from queue now */
 493     this->queue = g_list_delete_link (this->queue, this->queue);
 494
 495     /* set caps and push */
 496     gst_buffer_set_caps (buffer, caps);
 497     r = gst_pad_push (this->srcpad, buffer);
 498     if (r != GST_FLOW_OK) {
 499       GST_WARNING_OBJECT (this, "pushing queued GDP buffer returned %d", r);
 500       goto done;
 501     }
 502   }
 503
 504 done:
 505   gst_caps_unref (caps);
 506   GST_DEBUG_OBJECT (this, "stop");
 507   return r;
 508
 509   /* ERRORS */
 510 no_buffer:
 511   {
 512     GST_ELEMENT_ERROR (this, STREAM, FORMAT, (NULL),
 513         ("failed to create GDP buffer from streamheader"));
 514     return GST_FLOW_ERROR;
 515   }
 516 }
 517
 518 /* queue a buffer internally if we haven't sent streamheader buffers yet;
 519  * otherwise, just push on, this takes ownership of the buffer. */
 520 static GstFlowReturn
 521 gst_gdp_queue_buffer (GstGDPPay * this, GstBuffer * buffer)
 522 {
 523   if (this->sent_streamheader) {
 524     GST_LOG_OBJECT (this, "Pushing GDP buffer %p, caps %" GST_PTR_FORMAT,
 525         buffer, this->caps);
 526     return gst_pad_push (this->srcpad, buffer);
 527   }
 528
 529   /* store it on an internal queue. buffer remains reffed. */
 530   this->queue = g_list_append (this->queue, buffer);
 531   GST_DEBUG_OBJECT (this, "streamheader not sent yet, "
 532       "queued buffer %p, now %d buffers queued",
 533       buffer, g_list_length (this->queue));
 534
 535   return GST_FLOW_OK;
 536 }
 537
 538 static GstFlowReturn
 539 gst_gdp_pay_chain (GstPad * pad, GstBuffer * buffer)
 540 {
 541   GstGDPPay *this;
 542   GstCaps *caps;
 543   GstBuffer *outbuffer;
 544   GstFlowReturn ret;
 545
 546   this = GST_GDP_PAY (gst_pad_get_parent (pad));
 547
 548   /* we should have received a new_segment before, otherwise it's a bug.
 549    * fake one in that case */
 550   if (!this->new_segment_buf) {
 551     GstEvent *event;
 552
 553     GST_WARNING_OBJECT (this,
 554         "did not receive new-segment before first buffer");
 555     event = gst_event_new_new_segment (TRUE, 1.0, GST_FORMAT_BYTES, 0, -1, 0);
 556     outbuffer = gst_gdp_buffer_from_event (this, event);
 557     gst_event_unref (event);
 558
 559     /* GDP 0.2 doesn't know about new-segment, so this is not fatal */
 560     if (!outbuffer) {
 561       GST_ELEMENT_WARNING (this, STREAM, ENCODE, (NULL),
 562           ("Could not create GDP buffer from new segment event"));
 563     } else {
 564       GST_BUFFER_TIMESTAMP (outbuffer) = GST_BUFFER_TIMESTAMP (buffer);
 565       GST_BUFFER_DURATION (outbuffer) = 0;
 566       GST_DEBUG_OBJECT (this, "Storing buffer %p as new_segment_buf",
 567           outbuffer);
 568       this->new_segment_buf = outbuffer;
 569     }
 570   }
 571
 572   /* make sure we've received caps before */
 573   caps = gst_buffer_get_caps (buffer);
 574   if (!this->caps && !caps)
 575     goto no_caps;
 576
 577   /* if the caps have changed, process caps first */
 578   if (caps && !gst_caps_is_equal (this->caps, caps)) {
 579     GST_LOG_OBJECT (this, "caps changed to %p, %" GST_PTR_FORMAT, caps, caps);
 580     gst_caps_replace (&(this->caps), caps);
 581     outbuffer = gst_gdp_buffer_from_caps (this, caps);
 582     if (!outbuffer)
 583       goto no_caps_buffer;
 584
 585     GST_BUFFER_TIMESTAMP (outbuffer) = GST_BUFFER_TIMESTAMP (buffer);
 586     GST_BUFFER_DURATION (outbuffer) = 0;
 587     GST_BUFFER_FLAG_SET (outbuffer, GST_BUFFER_FLAG_IN_CAPS);
 588     this->caps_buf = outbuffer;
 589     gst_gdp_pay_reset_streamheader (this);
 590   }
 591
 592   if (caps)
 593     gst_caps_unref (caps);
 594
 595   /* create a GDP header packet,
 596    * then create a GST buffer of the header packet and the buffer contents */
 597   outbuffer = gst_gdp_pay_buffer_from_buffer (this, buffer);
 598   if (!outbuffer)
 599     goto no_buffer;
 600
 601   gst_gdp_stamp_buffer (this, outbuffer);
 602   GST_BUFFER_TIMESTAMP (outbuffer) = GST_BUFFER_TIMESTAMP (buffer);
 603   GST_BUFFER_DURATION (outbuffer) = GST_BUFFER_DURATION (buffer);
 604
 605   ret = gst_gdp_queue_buffer (this, outbuffer);
 606
 607 done:
 608   gst_buffer_unref (buffer);
 609   gst_object_unref (this);
 610   return ret;
 611
 612   /* ERRORS */
 613 no_caps:
 614   {
 615     /* when returning a fatal error as a GstFlowReturn we must post an error
 616      * message */
 617     GST_ELEMENT_ERROR (this, STREAM, FORMAT, (NULL),
 618         ("first received buffer does not have caps set"));
 619     if (caps)
 620       gst_caps_unref (caps);
 621     ret = GST_FLOW_NOT_NEGOTIATED;
 622     goto done;
 623   }
 624 no_caps_buffer:
 625   {
 626     GST_ELEMENT_ERROR (this, STREAM, ENCODE, (NULL),
 627         ("Could not create GDP buffer from caps %" GST_PTR_FORMAT, caps));
 628     gst_caps_unref (caps);
 629     ret = GST_FLOW_ERROR;
 630     goto done;
 631   }
 632 no_buffer:
 633   {
 634     GST_ELEMENT_ERROR (this, STREAM, ENCODE, (NULL),
 635         ("Could not create GDP buffer from buffer"));
 636     ret = GST_FLOW_ERROR;
 637     goto done;
 638   }
 639 }
 640
 641 static gboolean
 642 gst_gdp_pay_sink_event (GstPad * pad, GstEvent * event)
 643 {
 644   GstBuffer *outbuffer;
 645   GstGDPPay *this = GST_GDP_PAY (gst_pad_get_parent (pad));
 646   GstFlowReturn flowret;
 647   gboolean ret = TRUE;
 648
 649   GST_DEBUG_OBJECT (this, "received event %p of type %s (%d)",
 650       event, gst_event_type_get_name (event->type), event->type);
 651
 652   /* now turn the event into a buffer */
 653   outbuffer = gst_gdp_buffer_from_event (this, event);
 654   if (!outbuffer)
 655     goto no_outbuffer;
 656
 657   GST_BUFFER_TIMESTAMP (outbuffer) = GST_EVENT_TIMESTAMP (event);
 658   GST_BUFFER_DURATION (outbuffer) = 0;
 659
 660   /* if we got a new segment or tag event, we should put it on our streamheader,
 661    * and not send it on */
 662   switch (GST_EVENT_TYPE (event)) {
 663     case GST_EVENT_NEWSEGMENT:
 664       GST_DEBUG_OBJECT (this, "Storing buffer %p as new_segment_buf",
 665           outbuffer);
 666
 667       if (this->new_segment_buf)
 668         gst_buffer_unref (this->new_segment_buf);
 669       this->new_segment_buf = outbuffer;
 670
 671       gst_gdp_pay_reset_streamheader (this);
 672       break;
 673     case GST_EVENT_TAG:
 674       GST_DEBUG_OBJECT (this, "Storing buffer %p as tag_buf", outbuffer);
 675
 676       if (this->tag_buf)
 677         gst_buffer_unref (this->tag_buf);
 678       this->tag_buf = outbuffer;
 679
 680       gst_gdp_pay_reset_streamheader (this);
 681       break;
 682     default:
 683       GST_DEBUG_OBJECT (this, "queuing GDP buffer %p of event %p", outbuffer,
 684           event);
 685       flowret = gst_gdp_queue_buffer (this, outbuffer);
 686       if (flowret != GST_FLOW_OK)
 687         goto push_error;
 688       break;
 689   }
 690
 691   /* if we have EOS, we should send on EOS ourselves */
 692   if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
 693     GST_DEBUG_OBJECT (this, "Sending on EOS event %p", event);
 694     /* ref, we unref later again */
 695     ret = gst_pad_push_event (this->srcpad, gst_event_ref (event));
 696   }
 697
 698 done:
 699   gst_event_unref (event);
 700   gst_object_unref (this);
 701
 702   return ret;
 703
 704   /* ERRORS */
 705 no_outbuffer:
 706   {
 707     GST_ELEMENT_WARNING (this, STREAM, ENCODE, (NULL),
 708         ("Could not create GDP buffer from received event (type %s)",
 709             gst_event_type_get_name (event->type)));
 710     ret = FALSE;
 711     goto done;
 712   }
 713 push_error:
 714   {
 715     GST_WARNING_OBJECT (this, "queueing GDP event buffer returned %d", flowret);
 716     ret = FALSE;
 717     goto done;
 718   }
 719 }
 720
 721 static gboolean
 722 gst_gdp_pay_src_event (GstPad * pad, GstEvent * event)
 723 {
 724   GstGDPPay *this;
 725   gboolean res = TRUE;
 726
 727   this = GST_GDP_PAY (gst_pad_get_parent (pad));
 728
 729   switch (GST_EVENT_TYPE (event)) {
 730     case GST_EVENT_SEEK:
 731       /* we refuse seek for now. */
 732       gst_event_unref (event);
 733       res = FALSE;
 734       break;
 735     case GST_EVENT_QOS:
 736     case GST_EVENT_NAVIGATION:
 737     default:
 738       /* everything else is passed */
 739       res = gst_pad_push_event (this->sinkpad, event);
 740       break;
 741   }
 742   gst_object_unref (this);
 743
 744   return res;
 745 }
 746
 747 static void
 748 gst_gdp_pay_set_property (GObject * object, guint prop_id,
 749     const GValue * value, GParamSpec * pspec)
 750 {
 751   GstGDPPay *this;
 752
 753   g_return_if_fail (GST_IS_GDP_PAY (object));
 754   this = GST_GDP_PAY (object);
 755
 756   switch (prop_id) {
 757     case PROP_CRC_HEADER:
 758       this->crc_header =
 759           g_value_get_boolean (value) ? GST_DP_HEADER_FLAG_CRC_HEADER : 0;
 760       this->header_flag = this->crc_header | this->crc_payload;
 761       break;
 762     case PROP_CRC_PAYLOAD:
 763       this->crc_payload =
 764           g_value_get_boolean (value) ? GST_DP_HEADER_FLAG_CRC_PAYLOAD : 0;
 765       this->header_flag = this->crc_header | this->crc_payload;
 766       break;
 767     case PROP_VERSION:
 768       this->version = g_value_get_enum (value);
 769       break;
 770     default:
 771       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 772       break;
 773   }
 774 }
 775
 776 static void
 777 gst_gdp_pay_get_property (GObject * object, guint prop_id,
 778     GValue * value, GParamSpec * pspec)
 779 {
 780   GstGDPPay *this;
 781
 782   g_return_if_fail (GST_IS_GDP_PAY (object));
 783   this = GST_GDP_PAY (object);
 784
 785   switch (prop_id) {
 786     case PROP_CRC_HEADER:
 787       g_value_set_boolean (value, this->crc_header);
 788       break;
 789     case PROP_CRC_PAYLOAD:
 790       g_value_set_boolean (value, this->crc_payload);
 791       break;
 792     case PROP_VERSION:
 793       g_value_set_enum (value, this->version);
 794       break;
 795     default:
 796       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 797       break;
 798   }
 799 }
 800
 801 static GstStateChangeReturn
 802 gst_gdp_pay_change_state (GstElement * element, GstStateChange transition)
 803 {
 804   GstStateChangeReturn ret;
 805   GstGDPPay *this = GST_GDP_PAY (element);
 806
 807   switch (transition) {
 808     case GST_STATE_CHANGE_READY_TO_PAUSED:
 809       break;
 810     default:
 811       break;
 812   }
 813
 814   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 815
 816   switch (transition) {
 817     case GST_STATE_CHANGE_PAUSED_TO_READY:
 818       gst_gdp_pay_reset (this);
 819       break;
 820     default:
 821       break;
 822   }
 823
 824   return ret;
 825 }
 826
 827 gboolean
 828 gst_gdp_pay_plugin_init (GstPlugin * plugin)
 829 {
 830   if (!gst_element_register (plugin, "gdppay", GST_RANK_NONE, GST_TYPE_GDP_PAY))
 831     return FALSE;
 832
 833   return TRUE;
 834 }