gst/gdp/gstgdppay.c

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