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  * </refsect2>
  30  */
  31
  32 #ifdef HAVE_CONFIG_H
  33 #include "config.h"
  34 #endif
  35
  36 #include <gst/dataprotocol/dataprotocol.h>
  37
  38 #include "gstgdppay.h"
  39
  40 /* elementfactory information */
  41 static const GstElementDetails gdp_pay_details =
  42 GST_ELEMENT_DETAILS ("GDP Payloader",
  43     "GDP/Payloader",
  44     "Payloads GStreamer Data Protocol buffers",
  45     "Thomas Vander Stichele <thomas at apestaart dot org>");
  46
  47 static GstStaticPadTemplate gdp_pay_sink_template =
  48 GST_STATIC_PAD_TEMPLATE ("sink",
  49     GST_PAD_SINK,
  50     GST_PAD_ALWAYS,
  51     GST_STATIC_CAPS_ANY);
  52
  53 static GstStaticPadTemplate gdp_pay_src_template =
  54 GST_STATIC_PAD_TEMPLATE ("src",
  55     GST_PAD_SRC,
  56     GST_PAD_ALWAYS,
  57     GST_STATIC_CAPS ("application/x-gdp"));
  58
  59 GST_DEBUG_CATEGORY (gst_gdp_pay_debug);
  60 #define GST_CAT_DEFAULT gst_gdp_pay_debug
  61
  62 #define DEFAULT_CRC_HEADER TRUE
  63 #define DEFAULT_CRC_PAYLOAD FALSE
  64 #define DEFAULT_VERSION GST_DP_VERSION_1_0
  65
  66 enum
  67 {
  68   PROP_0,
  69   PROP_CRC_HEADER,
  70   PROP_CRC_PAYLOAD,
  71   PROP_VERSION,
  72 };
  73
  74 #define _do_init(x) \
  75     GST_DEBUG_CATEGORY_INIT (gst_gdp_pay_debug, "gdppay", 0, \
  76     "GDP payloader");
  77
  78 GST_BOILERPLATE_FULL (GstGDPPay, gst_gdp_pay, GstElement,
  79     GST_TYPE_ELEMENT, _do_init);
  80
  81 static GstFlowReturn gst_gdp_pay_chain (GstPad * pad, GstBuffer * buffer);
  82 static gboolean gst_gdp_pay_sink_event (GstPad * pad, GstEvent * event);
  83 static GstStateChangeReturn gst_gdp_pay_change_state (GstElement *
  84     element, GstStateChange transition);
  85
  86 static void gst_gdp_pay_set_property (GObject * object, guint prop_id,
  87     const GValue * value, GParamSpec * pspec);
  88 static void gst_gdp_pay_get_property (GObject * object, guint prop_id,
  89     GValue * value, GParamSpec * pspec);
  90
  91 static void gst_gdp_pay_dispose (GObject * gobject);
  92
  93 static void
  94 gst_gdp_pay_base_init (gpointer g_class)
  95 {
  96   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
  97
  98   gst_element_class_set_details (element_class, &gdp_pay_details);
  99
 100   gst_element_class_add_pad_template (element_class,
 101       gst_static_pad_template_get (&gdp_pay_sink_template));
 102   gst_element_class_add_pad_template (element_class,
 103       gst_static_pad_template_get (&gdp_pay_src_template));
 104 }
 105
 106 static void
 107 gst_gdp_pay_class_init (GstGDPPayClass * klass)
 108 {
 109   GObjectClass *gobject_class;
 110   GstElementClass *gstelement_class;
 111
 112   gobject_class = (GObjectClass *) klass;
 113   gstelement_class = (GstElementClass *) klass;
 114
 115   parent_class = g_type_class_peek_parent (klass);
 116
 117   gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_gdp_pay_set_property);
 118   gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_gdp_pay_get_property);
 119   gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_gdp_pay_dispose);
 120   gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_gdp_pay_change_state);
 121
 122   g_object_class_install_property (gobject_class, PROP_CRC_HEADER,
 123       g_param_spec_boolean ("crc-header", "CRC Header",
 124           "Calculate and store a CRC checksum on the header",
 125           DEFAULT_CRC_HEADER, G_PARAM_READWRITE));
 126   g_object_class_install_property (gobject_class, PROP_CRC_PAYLOAD,
 127       g_param_spec_boolean ("crc-payload", "CRC Payload",
 128           "Calculate and store a CRC checksum on the payload",
 129           DEFAULT_CRC_PAYLOAD, G_PARAM_READWRITE));
 130   g_object_class_install_property (gobject_class, PROP_VERSION,
 131       g_param_spec_enum ("version", "Version",
 132           "Version of the GStreamer Data Protocol",
 133           GST_TYPE_DP_VERSION, DEFAULT_VERSION, G_PARAM_READWRITE));
 134 }
 135
 136 static void
 137 gst_gdp_pay_init (GstGDPPay * gdppay, GstGDPPayClass * g_class)
 138 {
 139   gdppay->sinkpad =
 140       gst_pad_new_from_static_template (&gdp_pay_sink_template, "sink");
 141   gst_pad_set_chain_function (gdppay->sinkpad,
 142       GST_DEBUG_FUNCPTR (gst_gdp_pay_chain));
 143   gst_pad_set_event_function (gdppay->sinkpad,
 144       GST_DEBUG_FUNCPTR (gst_gdp_pay_sink_event));
 145   gst_element_add_pad (GST_ELEMENT (gdppay), gdppay->sinkpad);
 146
 147   gdppay->srcpad =
 148       gst_pad_new_from_static_template (&gdp_pay_src_template, "src");
 149   gst_element_add_pad (GST_ELEMENT (gdppay), gdppay->srcpad);
 150
 151   gdppay->offset = 0;
 152
 153   gdppay->crc_header = DEFAULT_CRC_HEADER;
 154   gdppay->crc_payload = DEFAULT_CRC_PAYLOAD;
 155   gdppay->header_flag = gdppay->crc_header | gdppay->crc_payload;
 156   gdppay->version = DEFAULT_VERSION;
 157 }
 158
 159 static void
 160 gst_gdp_pay_dispose (GObject * gobject)
 161 {
 162   GstGDPPay *this = GST_GDP_PAY (gobject);
 163
 164   if (this->caps_buf) {
 165     gst_buffer_unref (this->caps_buf);
 166     this->caps_buf = NULL;
 167   }
 168   if (this->new_segment_buf) {
 169     gst_buffer_unref (this->new_segment_buf);
 170     this->new_segment_buf = NULL;
 171   }
 172   GST_CALL_PARENT (G_OBJECT_CLASS, dispose, (gobject));
 173 }
 174
 175 /* set OFFSET and OFFSET_END with running count */
 176 static void
 177 gst_gdp_stamp_buffer (GstGDPPay * this, GstBuffer * buffer)
 178 {
 179   GST_BUFFER_OFFSET (buffer) = this->offset;
 180   GST_BUFFER_OFFSET_END (buffer) = this->offset + GST_BUFFER_SIZE (buffer);
 181   this->offset = GST_BUFFER_OFFSET_END (buffer);
 182 }
 183
 184 static GstBuffer *
 185 gst_gdp_buffer_from_caps (GstGDPPay * this, GstCaps * caps)
 186 {
 187   GstBuffer *headerbuf;
 188   GstBuffer *payloadbuf;
 189   guint8 *header, *payload;
 190   guint len;
 191
 192   if (!this->packetizer->packet_from_caps (caps, this->header_flag, &len,
 193           &header, &payload)) {
 194     GST_WARNING_OBJECT (this, "could not create GDP header from caps");
 195     return NULL;
 196   }
 197
 198   GST_LOG_OBJECT (this, "creating GDP header and payload buffer from caps");
 199   headerbuf = gst_buffer_new ();
 200   gst_buffer_set_data (headerbuf, header, len);
 201   GST_BUFFER_MALLOCDATA (headerbuf) = header;
 202
 203   payloadbuf = gst_buffer_new ();
 204   gst_buffer_set_data (payloadbuf, payload,
 205       gst_dp_header_payload_length (header));
 206   GST_BUFFER_MALLOCDATA (payloadbuf) = payload;
 207
 208   return gst_buffer_join (headerbuf, payloadbuf);
 209 }
 210
 211 static GstBuffer *
 212 gst_gdp_pay_buffer_from_buffer (GstGDPPay * this, GstBuffer * buffer)
 213 {
 214   GstBuffer *headerbuf;
 215   guint8 *header;
 216   guint len;
 217
 218   if (!this->packetizer->header_from_buffer (buffer, this->header_flag, &len,
 219           &header)) {
 220     GST_WARNING_OBJECT (this, "could not create GDP header from buffer");
 221     return NULL;
 222   }
 223
 224   GST_LOG_OBJECT (this, "creating GDP header and payload buffer from buffer");
 225   headerbuf = gst_buffer_new ();
 226   gst_buffer_set_data (headerbuf, header, len);
 227   GST_BUFFER_MALLOCDATA (headerbuf) = header;
 228
 229   /* we do not want to lose the ref on the incoming buffer */
 230   gst_buffer_ref (buffer);
 231   return gst_buffer_join (headerbuf, buffer);
 232 }
 233
 234 static GstBuffer *
 235 gst_gdp_buffer_from_event (GstGDPPay * this, GstEvent * event)
 236 {
 237   GstBuffer *headerbuf;
 238   GstBuffer *payloadbuf;
 239   guint8 *header, *payload;
 240   guint len;
 241   gboolean ret;
 242
 243   ret =
 244       this->packetizer->packet_from_event (event, this->header_flag, &len,
 245       &header, &payload);
 246
 247   if (!ret) {
 248     GST_WARNING_OBJECT (this, "could not create GDP header from event %s (%d)",
 249         gst_event_type_get_name (event->type), event->type);
 250     return NULL;
 251   }
 252
 253   GST_LOG_OBJECT (this, "creating GDP header and payload buffer from event");
 254   headerbuf = gst_buffer_new ();
 255   gst_buffer_set_data (headerbuf, header, len);
 256   GST_BUFFER_MALLOCDATA (headerbuf) = header;
 257
 258   payloadbuf = gst_buffer_new ();
 259   gst_buffer_set_data (payloadbuf, payload,
 260       gst_dp_header_payload_length (header));
 261   GST_BUFFER_MALLOCDATA (payloadbuf) = payload;
 262
 263   return gst_buffer_join (headerbuf, payloadbuf);
 264 }
 265
 266
 267 /* set our caps with streamheader, based on the latest newsegment and caps,
 268  * and (possibly) GDP-serialized buffers of the streamheaders on the src pad */
 269 static GstFlowReturn
 270 gst_gdp_pay_reset_streamheader (GstGDPPay * this)
 271 {
 272   GstCaps *caps;
 273   GstStructure *structure;
 274   GstBuffer *new_segment_buf = NULL, *caps_buf, *tag_buf = NULL;
 275   GstFlowReturn r = GST_FLOW_OK;
 276   gboolean version_one_zero = TRUE;
 277
 278   GValue array = { 0 };
 279   GValue value = { 0 };
 280
 281   /* In version 0.2, we didn't need or send new segment or tags */
 282   if (this->version == GST_DP_VERSION_0_2)
 283     version_one_zero = FALSE;
 284
 285   if (version_one_zero) {
 286     if (!this->new_segment_buf || !this->caps_buf)
 287       return GST_FLOW_OK;
 288   } else {
 289     if (!this->caps_buf)
 290       return GST_FLOW_OK;
 291   }
 292
 293   /* put copies of the buffers in a fixed list */
 294   g_value_init (&array, GST_TYPE_ARRAY);
 295
 296   if (version_one_zero) {
 297     new_segment_buf = gst_buffer_copy (this->new_segment_buf);
 298     g_value_init (&value, GST_TYPE_BUFFER);
 299     gst_value_set_buffer (&value, new_segment_buf);
 300     gst_value_array_append_value (&array, &value);
 301     g_value_unset (&value);
 302
 303     if (this->tag_buf) {
 304       tag_buf = gst_buffer_copy (this->tag_buf);
 305       g_value_init (&value, GST_TYPE_BUFFER);
 306       gst_value_set_buffer (&value, tag_buf);
 307       gst_value_array_append_value (&array, &value);
 308       g_value_unset (&value);
 309     }
 310   }
 311
 312   caps_buf = gst_buffer_copy (this->caps_buf);
 313   g_value_init (&value, GST_TYPE_BUFFER);
 314   gst_value_set_buffer (&value, caps_buf);
 315   gst_value_array_append_value (&array, &value);
 316   g_value_unset (&value);
 317
 318   /* we also need to add GDP serializations of the streamheaders of the
 319    * incoming caps */
 320   structure = gst_caps_get_structure (this->caps, 0);
 321   if (gst_structure_has_field (structure, "streamheader")) {
 322     const GValue *sh;
 323     GArray *buffers;
 324     GstBuffer *buffer;
 325     int i;
 326
 327     sh = gst_structure_get_value (structure, "streamheader");
 328     buffers = g_value_peek_pointer (sh);
 329     GST_DEBUG_OBJECT (this,
 330         "Need to serialize %d incoming streamheader buffers on ours",
 331         buffers->len);
 332     for (i = 0; i < buffers->len; ++i) {
 333       GValue *bufval;
 334       GstBuffer *outbuffer;
 335
 336       bufval = &g_array_index (buffers, GValue, i);
 337       buffer = g_value_peek_pointer (bufval);
 338       outbuffer = gst_gdp_pay_buffer_from_buffer (this, buffer);
 339       if (outbuffer) {
 340         g_value_init (&value, GST_TYPE_BUFFER);
 341         gst_value_set_buffer (&value, outbuffer);
 342         gst_value_array_append_value (&array, &value);
 343         g_value_unset (&value);
 344       }
 345       /* FIXME: if one or more in this loop fail to produce and outbuffer,
 346        * should we error out ? Once ? Every time ? */
 347     }
 348   }
 349
 350   caps = gst_caps_from_string ("application/x-gdp");
 351   structure = gst_caps_get_structure (caps, 0);
 352
 353   gst_structure_set_value (structure, "streamheader", &array);
 354   g_value_unset (&array);
 355
 356   /* Unref our copies */
 357   if (new_segment_buf)
 358     gst_buffer_unref (new_segment_buf);
 359   gst_buffer_unref (caps_buf);
 360
 361   GST_DEBUG_OBJECT (this, "Setting caps on src pad %" GST_PTR_FORMAT, caps);
 362   gst_pad_set_caps (this->srcpad, caps);
 363   gst_buffer_set_caps (this->caps_buf, caps);
 364   gst_buffer_set_caps (this->new_segment_buf, caps);
 365
 366   /* if these are our first ever buffers, send out new_segment first */
 367   if (!this->sent_streamheader) {
 368     GstEvent *event =
 369         gst_event_new_new_segment (TRUE, 1.0, GST_FORMAT_BYTES, 0, -1, 0);
 370     GST_DEBUG_OBJECT (this, "Sending out new_segment event %p", event);
 371     if (!gst_pad_push_event (this->srcpad, event)) {
 372       GST_WARNING_OBJECT (this, "pushing new segment failed");
 373       return GST_FLOW_ERROR;
 374     }
 375   }
 376
 377   /* push out these streamheader buffers, then flush our internal queue */
 378   GST_DEBUG_OBJECT (this, "Pushing GDP new_segment buffer %p",
 379       this->new_segment_buf);
 380   /* we stored these bufs with refcount 1, so make sure we keep a ref */
 381   r = gst_pad_push (this->srcpad, gst_buffer_ref (this->new_segment_buf));
 382   if (r != GST_FLOW_OK) {
 383     GST_WARNING_OBJECT (this, "pushing GDP newsegment buffer returned %d", r);
 384     return r;
 385   }
 386   if (this->tag_buf) {
 387     GST_DEBUG_OBJECT (this, "Pushing GDP tag buffer %p", this->tag_buf);
 388     /* we stored these bufs with refcount 1, so make sure we keep a ref */
 389     r = gst_pad_push (this->srcpad, gst_buffer_ref (this->tag_buf));
 390     if (r != GST_FLOW_OK) {
 391       GST_WARNING_OBJECT (this, "pushing GDP tag buffer returned %d", r);
 392       return r;
 393     }
 394   }
 395   GST_DEBUG_OBJECT (this, "Pushing GDP caps buffer %p", this->new_segment_buf);
 396   r = gst_pad_push (this->srcpad, gst_buffer_ref (this->caps_buf));
 397   if (r != GST_FLOW_OK) {
 398     GST_WARNING_OBJECT (this, "pushing GDP caps buffer returned %d", r);
 399     return r;
 400   }
 401   this->sent_streamheader = TRUE;
 402   GST_DEBUG_OBJECT (this, "need to push %d queued buffers",
 403       g_list_length (this->queue));
 404   if (this->queue) {
 405     GList *l;
 406
 407     for (l = this->queue; l; l = g_list_next (l)) {
 408       GST_DEBUG_OBJECT (this, "Pushing queued GDP buffer %p", l->data);
 409       gst_buffer_set_caps (l->data, caps);
 410       r = gst_pad_push (this->srcpad, l->data);
 411       if (r != GST_FLOW_OK) {
 412         GST_WARNING_OBJECT (this, "pushing queued GDP buffer returned %d", r);
 413         return r;
 414       }
 415     }
 416   }
 417
 418   return r;
 419 }
 420
 421 /* queue a buffer internally if we haven't sent streamheader buffers yet;
 422  * otherwise, just push on */
 423 static GstFlowReturn
 424 gst_gdp_queue_buffer (GstGDPPay * this, GstBuffer * buffer)
 425 {
 426   if (this->sent_streamheader) {
 427     GST_LOG_OBJECT (this, "Pushing GDP buffer %p", buffer);
 428     GST_LOG_OBJECT (this, "set caps %" GST_PTR_FORMAT, this->caps);
 429     return gst_pad_push (this->srcpad, buffer);
 430   }
 431
 432   /* store it on an internal queue */
 433   this->queue = g_list_append (this->queue, buffer);
 434   GST_DEBUG_OBJECT (this, "queued buffer %p, now %d buffers queued",
 435       buffer, g_list_length (this->queue));
 436   return GST_FLOW_OK;
 437 }
 438
 439 static GstFlowReturn
 440 gst_gdp_pay_chain (GstPad * pad, GstBuffer * buffer)
 441 {
 442   GstGDPPay *this;
 443   GstCaps *caps;
 444   GstBuffer *outbuffer;
 445   GstFlowReturn ret;
 446
 447   this = GST_GDP_PAY (gst_pad_get_parent (pad));
 448
 449   /* we should have received a new_segment before, otherwise it's a bug.
 450    * fake one in that case */
 451   if (!this->new_segment_buf) {
 452     GstEvent *event;
 453
 454     GST_WARNING_OBJECT (this,
 455         "did not receive new-segment before first buffer");
 456     event = gst_event_new_new_segment (TRUE, 1.0, GST_FORMAT_BYTES, 0, -1, 0);
 457     outbuffer = gst_gdp_buffer_from_event (this, event);
 458     gst_event_unref (event);
 459
 460     /* GDP 0.2 doesn't know about new-segment, so this is not fatal */
 461     if (!outbuffer) {
 462       GST_ELEMENT_WARNING (this, STREAM, ENCODE, (NULL),
 463           ("Could not create GDP buffer from new segment event"));
 464 /*
 465       ret = GST_FLOW_ERROR;
 466       goto done;
 467 */
 468     } else {
 469
 470       gst_gdp_stamp_buffer (this, outbuffer);
 471       GST_BUFFER_TIMESTAMP (outbuffer) = GST_BUFFER_TIMESTAMP (buffer);
 472       GST_BUFFER_DURATION (outbuffer) = 0;
 473       GST_DEBUG_OBJECT (this, "Storing buffer %p as new_segment_buf",
 474           outbuffer);
 475       this->new_segment_buf = outbuffer;
 476     }
 477   }
 478
 479   /* make sure we've received caps before */
 480   caps = gst_buffer_get_caps (buffer);
 481   if (!this->caps && !caps) {
 482     GST_WARNING_OBJECT (this, "first received buffer does not have caps set");
 483     if (caps)
 484       gst_caps_unref (caps);
 485     ret = GST_FLOW_NOT_NEGOTIATED;
 486     goto done;
 487   }
 488
 489   /* if the caps have changed, process caps first */
 490   if (caps && !gst_caps_is_equal (this->caps, caps)) {
 491     GST_LOG_OBJECT (this, "caps changed to %p, %" GST_PTR_FORMAT, caps, caps);
 492     gst_caps_replace (&(this->caps), caps);
 493     outbuffer = gst_gdp_buffer_from_caps (this, caps);
 494     if (!outbuffer) {
 495       GST_ELEMENT_ERROR (this, STREAM, ENCODE, (NULL),
 496           ("Could not create GDP buffer from caps %" GST_PTR_FORMAT, caps));
 497       gst_caps_unref (caps);
 498       ret = GST_FLOW_ERROR;
 499       goto done;
 500     }
 501
 502     gst_gdp_stamp_buffer (this, outbuffer);
 503     GST_BUFFER_TIMESTAMP (outbuffer) = GST_BUFFER_TIMESTAMP (buffer);
 504     GST_BUFFER_DURATION (outbuffer) = 0;
 505     GST_BUFFER_FLAG_SET (outbuffer, GST_BUFFER_FLAG_IN_CAPS);
 506     this->caps_buf = outbuffer;
 507     gst_gdp_pay_reset_streamheader (this);
 508   }
 509
 510   /* create a GDP header packet,
 511    * then create a GST buffer of the header packet and the buffer contents */
 512   outbuffer = gst_gdp_pay_buffer_from_buffer (this, buffer);
 513   if (!outbuffer) {
 514     GST_ELEMENT_ERROR (this, STREAM, ENCODE, (NULL),
 515         ("Could not create GDP buffer from buffer"));
 516     ret = GST_FLOW_ERROR;
 517     goto done;
 518   }
 519
 520   gst_gdp_stamp_buffer (this, outbuffer);
 521   GST_BUFFER_TIMESTAMP (outbuffer) = GST_BUFFER_TIMESTAMP (buffer);
 522   GST_BUFFER_DURATION (outbuffer) = GST_BUFFER_DURATION (buffer);
 523
 524   ret = gst_gdp_queue_buffer (this, outbuffer);
 525
 526 done:
 527   gst_buffer_unref (buffer);
 528   gst_object_unref (this);
 529   return ret;
 530 }
 531
 532 static gboolean
 533 gst_gdp_pay_sink_event (GstPad * pad, GstEvent * event)
 534 {
 535   GstBuffer *outbuffer;
 536   GstGDPPay *this = GST_GDP_PAY (gst_pad_get_parent (pad));
 537   GstFlowReturn flowret;
 538   gboolean ret = TRUE;
 539
 540   GST_DEBUG_OBJECT (this, "received event %p of type %s (%d)",
 541       event, gst_event_type_get_name (event->type), event->type);
 542
 543   /* now turn the event into a buffer */
 544   outbuffer = gst_gdp_buffer_from_event (this, event);
 545   if (!outbuffer) {
 546     GST_ELEMENT_WARNING (this, STREAM, ENCODE, (NULL),
 547         ("Could not create GDP buffer from received event (type %s)",
 548             gst_event_type_get_name (event->type)));
 549     ret = FALSE;
 550     goto done;
 551   }
 552   gst_gdp_stamp_buffer (this, outbuffer);
 553   GST_BUFFER_TIMESTAMP (outbuffer) = GST_EVENT_TIMESTAMP (event);
 554   GST_BUFFER_DURATION (outbuffer) = 0;
 555
 556   /* if we got a new segment, we should put it on our streamheader,
 557    * and not send it on */
 558   switch (GST_EVENT_TYPE (event)) {
 559     case GST_EVENT_NEWSEGMENT:
 560       GST_DEBUG_OBJECT (this, "received new_segment event");
 561       if (this->new_segment_buf) {
 562         gst_buffer_unref (this->new_segment_buf);
 563       }
 564       GST_DEBUG_OBJECT (this, "Storing buffer %p as new_segment_buf",
 565           outbuffer);
 566       this->new_segment_buf = outbuffer;
 567       gst_gdp_pay_reset_streamheader (this);
 568       break;
 569     case GST_EVENT_TAG:
 570       GST_DEBUG_OBJECT (this, "received tag event");
 571       if (this->tag_buf) {
 572         gst_buffer_unref (this->tag_buf);
 573       }
 574       GST_DEBUG_OBJECT (this, "Storing buffer %p as tag_buf", outbuffer);
 575       this->tag_buf = outbuffer;
 576       gst_gdp_pay_reset_streamheader (this);
 577       break;
 578     default:
 579       GST_DEBUG_OBJECT (this, "queuing GDP buffer %p of event %p", outbuffer,
 580           event);
 581       flowret = gst_gdp_queue_buffer (this, outbuffer);
 582       if (flowret != GST_FLOW_OK) {
 583         GST_WARNING_OBJECT (this, "queueing GDP event buffer returned %d",
 584             flowret);
 585         ret = FALSE;
 586         goto done;
 587       }
 588       break;
 589   }
 590
 591   /* if we have EOS, we should send on EOS ourselves */
 592   if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
 593     GST_DEBUG_OBJECT (this, "Sending on EOS event %p", event);
 594     return gst_pad_push_event (this->srcpad, event);
 595   };
 596
 597 done:
 598   gst_object_unref (this);
 599   gst_event_unref (event);
 600   return ret;
 601 }
 602
 603 static void
 604 gst_gdp_pay_set_property (GObject * object, guint prop_id,
 605     const GValue * value, GParamSpec * pspec)
 606 {
 607   GstGDPPay *this;
 608
 609   g_return_if_fail (GST_IS_GDP_PAY (object));
 610   this = GST_GDP_PAY (object);
 611
 612   switch (prop_id) {
 613     case PROP_CRC_HEADER:
 614       this->crc_header =
 615           g_value_get_boolean (value) ? GST_DP_HEADER_FLAG_CRC_HEADER : 0;
 616       this->header_flag = this->crc_header | this->crc_payload;
 617       break;
 618     case PROP_CRC_PAYLOAD:
 619       this->crc_payload =
 620           g_value_get_boolean (value) ? GST_DP_HEADER_FLAG_CRC_PAYLOAD : 0;
 621       this->header_flag = this->crc_header | this->crc_payload;
 622       break;
 623     case PROP_VERSION:
 624       this->version = g_value_get_enum (value);
 625       break;
 626     default:
 627       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 628       break;
 629   }
 630 }
 631
 632 static void
 633 gst_gdp_pay_get_property (GObject * object, guint prop_id,
 634     GValue * value, GParamSpec * pspec)
 635 {
 636   GstGDPPay *this;
 637
 638   g_return_if_fail (GST_IS_GDP_PAY (object));
 639   this = GST_GDP_PAY (object);
 640
 641   switch (prop_id) {
 642     case PROP_CRC_HEADER:
 643       g_value_set_boolean (value, this->crc_header);
 644       break;
 645     case PROP_CRC_PAYLOAD:
 646       g_value_set_boolean (value, this->crc_payload);
 647       break;
 648     case PROP_VERSION:
 649       g_value_set_enum (value, this->version);
 650       break;
 651     default:
 652       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 653       break;
 654   }
 655 }
 656
 657 static GstStateChangeReturn
 658 gst_gdp_pay_change_state (GstElement * element, GstStateChange transition)
 659 {
 660   GstStateChangeReturn ret;
 661   GstGDPPay *this = GST_GDP_PAY (element);
 662
 663   switch (transition) {
 664     case GST_STATE_CHANGE_READY_TO_PAUSED:
 665       this->packetizer = gst_dp_packetizer_new (this->version);
 666       break;
 667     default:
 668       break;
 669   }
 670
 671   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 672
 673   switch (transition) {
 674     case GST_STATE_CHANGE_PAUSED_TO_READY:
 675       if (this->packetizer) {
 676         gst_dp_packetizer_free (this->packetizer);
 677         this->packetizer = NULL;
 678       }
 679       break;
 680     case GST_STATE_CHANGE_READY_TO_NULL:
 681       if (this->caps) {
 682         gst_caps_unref (this->caps);
 683         this->caps = NULL;
 684       }
 685       break;
 686     default:
 687       break;
 688   }
 689
 690   return ret;
 691 }
 692
 693 gboolean
 694 gst_gdp_pay_plugin_init (GstPlugin * plugin)
 695 {
 696   if (!gst_element_register (plugin, "gdppay", GST_RANK_NONE, GST_TYPE_GDP_PAY))
 697     return FALSE;
 698
 699   return TRUE;
 700 }