gst/gdp/gstgdppay.c

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