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