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