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