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