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