gst/onvif/gstrtponviftimestamp.c

   1 /*
   2  * gstrtponviftimestamp.h
   3  *
   4  * Copyright (C) 2014 Axis Communications AB
   5  *  Author: Guillaume Desmottes <guillaume.desmottes@collabora.com>
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Lesser General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2.1 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Lesser General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Lesser General Public
  18  * License along with this library; if not, see <http://www.gnu.org/licenses/>.
  19  */
  20
  21 #ifdef HAVE_CONFIG_H
  22 #include "config.h"
  23 #endif
  24
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28
  29 #include <gst/rtp/gstrtpbuffer.h>
  30
  31 #include "gstrtponviftimestamp.h"
  32
  33 #define GST_NTP_OFFSET_EVENT_NAME "GstNtpOffset"
  34
  35 #define DEFAULT_NTP_OFFSET GST_CLOCK_TIME_NONE
  36 #define DEFAULT_CSEQ 0
  37 #define DEFAULT_SET_E_BIT FALSE
  38 #define DEFAULT_SET_T_BIT FALSE
  39 #define DEFAULT_DROP_OUT_OF_SEGMENT TRUE
  40
  41 GST_DEBUG_CATEGORY_STATIC (rtponviftimestamp_debug);
  42 #define GST_CAT_DEFAULT (rtponviftimestamp_debug)
  43
  44 static GstFlowReturn gst_rtp_onvif_timestamp_chain (GstPad * pad,
  45     GstObject * parent, GstBuffer * buf);
  46 static GstFlowReturn gst_rtp_onvif_timestamp_chain_list (GstPad * pad,
  47     GstObject * parent, GstBufferList * list);
  48
  49 static GstFlowReturn handle_and_push_buffer (GstRtpOnvifTimestamp * self,
  50     GstBuffer * buf);
  51 static GstFlowReturn handle_and_push_buffer_list (GstRtpOnvifTimestamp * self,
  52     GstBufferList * list);
  53
  54 static GstStaticPadTemplate sink_template_factory =
  55 GST_STATIC_PAD_TEMPLATE ("sink",
  56     GST_PAD_SINK,
  57     GST_PAD_ALWAYS,
  58     GST_STATIC_CAPS ("application/x-rtp")
  59     );
  60
  61 static GstStaticPadTemplate src_template_factory =
  62 GST_STATIC_PAD_TEMPLATE ("src",
  63     GST_PAD_SRC,
  64     GST_PAD_ALWAYS,
  65     GST_STATIC_CAPS ("application/x-rtp")
  66     );
  67
  68 enum
  69 {
  70   PROP_0,
  71   PROP_NTP_OFFSET,
  72   PROP_CSEQ,
  73   PROP_SET_E_BIT,
  74   PROP_SET_T_BIT,
  75   PROP_DROP_OUT_OF_SEGMENT
  76 };
  77
  78 /*static guint gst_rtp_onvif_timestamp_signals[LAST_SIGNAL] = { 0 }; */
  79
  80 G_DEFINE_TYPE (GstRtpOnvifTimestamp, gst_rtp_onvif_timestamp, GST_TYPE_ELEMENT);
  81 GST_ELEMENT_REGISTER_DEFINE (rtponviftimestamp, "rtponviftimestamp",
  82     GST_RANK_NONE, GST_TYPE_RTP_ONVIF_TIMESTAMP);
  83
  84 static void
  85 gst_rtp_onvif_timestamp_get_property (GObject * object,
  86     guint prop_id, GValue * value, GParamSpec * pspec)
  87 {
  88   GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (object);
  89
  90   switch (prop_id) {
  91     case PROP_NTP_OFFSET:
  92       g_value_set_uint64 (value, self->prop_ntp_offset);
  93       break;
  94     case PROP_CSEQ:
  95       g_value_set_uint (value, self->prop_cseq);
  96       break;
  97     case PROP_SET_E_BIT:
  98       g_value_set_boolean (value, self->prop_set_e_bit);
  99       break;
 100     case PROP_SET_T_BIT:
 101       g_value_set_boolean (value, self->prop_set_t_bit);
 102       break;
 103     case PROP_DROP_OUT_OF_SEGMENT:
 104       g_value_set_boolean (value, self->prop_drop_out_of_segment);
 105       break;
 106     default:
 107       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 108       break;
 109   }
 110 }
 111
 112 static void
 113 gst_rtp_onvif_timestamp_set_property (GObject * object,
 114     guint prop_id, const GValue * value, GParamSpec * pspec)
 115 {
 116   GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (object);
 117
 118   switch (prop_id) {
 119     case PROP_NTP_OFFSET:
 120       self->prop_ntp_offset = g_value_get_uint64 (value);
 121       break;
 122     case PROP_CSEQ:
 123       self->prop_cseq = g_value_get_uint (value);
 124       break;
 125     case PROP_SET_E_BIT:
 126       self->prop_set_e_bit = g_value_get_boolean (value);
 127       break;
 128     case PROP_SET_T_BIT:
 129       self->prop_set_t_bit = g_value_get_boolean (value);
 130       break;
 131     case PROP_DROP_OUT_OF_SEGMENT:
 132       self->prop_drop_out_of_segment = g_value_get_boolean (value);
 133       break;
 134     default:
 135       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 136       break;
 137   }
 138 }
 139
 140 /* send cached buffer or list, and events, if present */
 141 static GstFlowReturn
 142 send_cached_buffer_and_events (GstRtpOnvifTimestamp * self)
 143 {
 144   GstFlowReturn ret = GST_FLOW_OK;
 145
 146   g_assert (!(self->buffer && self->list));
 147
 148   if (self->buffer) {
 149     GST_DEBUG_OBJECT (self, "pushing %" GST_PTR_FORMAT, self->buffer);
 150     ret = handle_and_push_buffer (self, self->buffer);
 151     self->buffer = NULL;
 152   }
 153   if (self->list) {
 154     GST_DEBUG_OBJECT (self, "pushing %" GST_PTR_FORMAT, self->list);
 155     ret = handle_and_push_buffer_list (self, self->list);
 156     self->list = NULL;
 157   }
 158
 159   if (ret != GST_FLOW_OK)
 160     goto out;
 161
 162   while (!g_queue_is_empty (self->event_queue)) {
 163     GstEvent *event;
 164
 165     event = GST_EVENT_CAST (g_queue_pop_head (self->event_queue));
 166     GST_LOG_OBJECT (self->sinkpad, "sending %" GST_PTR_FORMAT, event);
 167     (void) gst_pad_send_event (self->sinkpad, event);
 168   }
 169
 170 out:
 171   return ret;
 172 }
 173
 174 static void
 175 purge_cached_buffer_and_events (GstRtpOnvifTimestamp * self)
 176 {
 177   g_assert (!(self->buffer && self->list));
 178
 179   if (self->buffer) {
 180     GST_DEBUG_OBJECT (self, "purging %" GST_PTR_FORMAT, self->buffer);
 181     gst_buffer_unref (self->buffer);
 182     self->buffer = NULL;
 183   }
 184   if (self->list) {
 185     GST_DEBUG_OBJECT (self, "purging %" GST_PTR_FORMAT, self->list);
 186     gst_buffer_list_unref (self->list);
 187     self->list = NULL;
 188   }
 189
 190   while (!g_queue_is_empty (self->event_queue)) {
 191     GstEvent *event;
 192
 193     event = GST_EVENT_CAST (g_queue_pop_head (self->event_queue));
 194     gst_event_unref (event);
 195   }
 196 }
 197
 198 static GstStateChangeReturn
 199 gst_rtp_onvif_timestamp_change_state (GstElement * element,
 200     GstStateChange transition)
 201 {
 202   GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (element);
 203   GstStateChangeReturn ret;
 204
 205   switch (transition) {
 206     case GST_STATE_CHANGE_READY_TO_PAUSED:
 207       self->ntp_offset = self->prop_ntp_offset;
 208       GST_DEBUG_OBJECT (self, "ntp-offset: %" GST_TIME_FORMAT,
 209           GST_TIME_ARGS (self->ntp_offset));
 210       self->set_d_bit = TRUE;
 211       self->set_e_bit = FALSE;
 212       self->set_t_bit = FALSE;
 213       break;
 214     default:
 215       break;
 216   }
 217
 218   ret = GST_ELEMENT_CLASS (gst_rtp_onvif_timestamp_parent_class)->change_state
 219       (element, transition);
 220
 221   if (ret == GST_STATE_CHANGE_FAILURE)
 222     return ret;
 223
 224   switch (transition) {
 225     case GST_STATE_CHANGE_PAUSED_TO_READY:
 226       purge_cached_buffer_and_events (self);
 227       gst_segment_init (&self->segment, GST_FORMAT_UNDEFINED);
 228       break;
 229     default:
 230       break;
 231   }
 232
 233   return ret;
 234 }
 235
 236 static void
 237 gst_rtp_onvif_timestamp_finalize (GObject * object)
 238 {
 239   GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (object);
 240
 241   g_queue_free (self->event_queue);
 242
 243   G_OBJECT_CLASS (gst_rtp_onvif_timestamp_parent_class)->finalize (object);
 244 }
 245
 246 static void
 247 gst_rtp_onvif_timestamp_class_init (GstRtpOnvifTimestampClass * klass)
 248 {
 249   GObjectClass *gobject_class;
 250   GstElementClass *gstelement_class;
 251
 252   gobject_class = G_OBJECT_CLASS (klass);
 253   gstelement_class = GST_ELEMENT_CLASS (klass);
 254
 255   gobject_class->get_property = gst_rtp_onvif_timestamp_get_property;
 256   gobject_class->set_property = gst_rtp_onvif_timestamp_set_property;
 257   gobject_class->finalize = gst_rtp_onvif_timestamp_finalize;
 258
 259   g_object_class_install_property (gobject_class, PROP_NTP_OFFSET,
 260       g_param_spec_uint64 ("ntp-offset", "NTP offset",
 261           "Offset between the pipeline running time and the absolute UTC time, "
 262           "in nano-seconds since 1900 (-1 for automatic computation)",
 263           0, G_MAXUINT64,
 264           DEFAULT_NTP_OFFSET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 265
 266   g_object_class_install_property (gobject_class, PROP_CSEQ,
 267       g_param_spec_uint ("cseq", "CSeq",
 268           "The RTSP CSeq which initiated the playback",
 269           0, G_MAXUINT32,
 270           DEFAULT_CSEQ, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 271
 272   g_object_class_install_property (gobject_class, PROP_SET_E_BIT,
 273       g_param_spec_boolean ("set-e-bit", "Set 'E' bit",
 274           "If the element should set the 'E' bit as defined in the ONVIF RTP "
 275           "extension. This increases latency by one packet",
 276           DEFAULT_SET_E_BIT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 277
 278   g_object_class_install_property (gobject_class, PROP_SET_T_BIT,
 279       g_param_spec_boolean ("set-t-bit", "Set 'T' bit",
 280           "If the element should set the 'T' bit as defined in the ONVIF RTP "
 281           "extension. This increases latency by one packet",
 282           DEFAULT_SET_T_BIT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 283
 284   g_object_class_install_property (gobject_class, PROP_DROP_OUT_OF_SEGMENT,
 285       g_param_spec_boolean ("drop-out-of-segment", "Drop out of segment",
 286           "Whether the element should drop buffers that fall outside the segment, "
 287           "not part of the specification but allows full reverse playback.",
 288           DEFAULT_DROP_OUT_OF_SEGMENT,
 289           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 290
 291   /* register pads */
 292   gst_element_class_add_static_pad_template (gstelement_class,
 293       &sink_template_factory);
 294   gst_element_class_add_static_pad_template (gstelement_class,
 295       &src_template_factory);
 296
 297   gst_element_class_set_static_metadata (gstelement_class,
 298       "ONVIF NTP timestamps RTP extension", "Effect/RTP",
 299       "Add absolute timestamps and flags of recorded data in a playback "
 300       "session", "Guillaume Desmottes <guillaume.desmottes@collabora.com>");
 301
 302   gstelement_class->change_state =
 303       GST_DEBUG_FUNCPTR (gst_rtp_onvif_timestamp_change_state);
 304
 305   GST_DEBUG_CATEGORY_INIT (rtponviftimestamp_debug, "rtponviftimestamp",
 306       0, "ONVIF NTP timestamps RTP extension");
 307 }
 308
 309 static gboolean
 310 parse_event_ntp_offset (GstRtpOnvifTimestamp * self, GstEvent * event,
 311     GstClockTime * offset, gboolean * discont)
 312 {
 313   const GstStructure *structure = gst_event_get_structure (event);
 314   GstClockTime event_offset;
 315   gboolean event_discont;
 316
 317   if (!gst_structure_get_clock_time (structure, "ntp-offset", &event_offset)) {
 318     GST_ERROR_OBJECT (self, "no ntp-offset in %" GST_PTR_FORMAT, event);
 319     return FALSE;
 320   }
 321   if (!gst_structure_get_boolean (structure, "discont", &event_discont)) {
 322     GST_ERROR_OBJECT (self, "no discontinue in %" GST_PTR_FORMAT, event);
 323     return FALSE;
 324   }
 325
 326   if (offset)
 327     *offset = event_offset;
 328
 329   if (discont)
 330     *discont = event_discont;
 331
 332   return TRUE;
 333 }
 334
 335 static gboolean
 336 gst_rtp_onvif_timestamp_sink_event (GstPad * pad, GstObject * parent,
 337     GstEvent * event)
 338 {
 339   GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (parent);
 340   gboolean drop = FALSE;
 341   gboolean ret = TRUE;
 342
 343   GST_DEBUG_OBJECT (pad, "handling event %s", GST_EVENT_TYPE_NAME (event));
 344
 345   /* handle serialized events, which, should not be enqueued */
 346   switch (GST_EVENT_TYPE (event)) {
 347     case GST_EVENT_CUSTOM_DOWNSTREAM:
 348       /* if the "set-e-bit" property is set, an offset event might mark the
 349        * stream as discontinued. We need to check if the currently cached buffer
 350        * needs the e-bit before it's pushed */
 351       if (self->buffer != NULL && self->prop_set_e_bit &&
 352           gst_event_has_name (event, GST_NTP_OFFSET_EVENT_NAME)) {
 353         gboolean discont;
 354         if (parse_event_ntp_offset (self, event, NULL, &discont)) {
 355           GST_DEBUG_OBJECT (self, "stream %s discontinued",
 356               (discont ? "is" : "is not"));
 357           self->set_e_bit = discont;
 358         } else {
 359           drop = TRUE;
 360           ret = FALSE;
 361           goto out;
 362         }
 363       }
 364       break;
 365     case GST_EVENT_EOS:
 366     {
 367       GstFlowReturn res;
 368
 369       /* Push pending buffers, if any */
 370       self->set_e_bit = TRUE;
 371       if (self->prop_set_t_bit)
 372         self->set_t_bit = TRUE;
 373       res = send_cached_buffer_and_events (self);
 374       if (res != GST_FLOW_OK) {
 375         drop = TRUE;
 376         ret = FALSE;
 377         goto out;
 378       }
 379       break;
 380     }
 381     case GST_EVENT_FLUSH_STOP:
 382       purge_cached_buffer_and_events (self);
 383       self->set_d_bit = TRUE;
 384       self->set_e_bit = FALSE;
 385       self->set_t_bit = FALSE;
 386       gst_segment_init (&self->segment, GST_FORMAT_UNDEFINED);
 387       break;
 388     default:
 389       break;
 390   }
 391
 392   /* enqueue serialized events if there is a cached buffer */
 393   if (GST_EVENT_IS_SERIALIZED (event) && (self->buffer || self->list)) {
 394     GST_DEBUG ("enqueueing serialized event");
 395     g_queue_push_tail (self->event_queue, event);
 396     event = NULL;
 397     goto out;
 398   }
 399
 400   /* handle rest of the events */
 401   switch (GST_EVENT_TYPE (event)) {
 402     case GST_EVENT_CUSTOM_DOWNSTREAM:
 403       /* update the ntp-offset after any cached buffer/buffer list has been
 404        * pushed. the d-bit of the next buffer/buffer list should be set if
 405        * the stream is discontinued */
 406       if (gst_event_has_name (event, GST_NTP_OFFSET_EVENT_NAME)) {
 407         GstClockTime offset;
 408         gboolean discont;
 409         if (parse_event_ntp_offset (self, event, &offset, &discont)) {
 410           GST_DEBUG_OBJECT (self, "new ntp-offset: %" GST_TIME_FORMAT
 411               ", stream %s discontinued", GST_TIME_ARGS (offset),
 412               (discont ? "is" : "is not"));
 413           self->ntp_offset = offset;
 414           self->set_d_bit = discont;
 415         } else {
 416           ret = FALSE;
 417         }
 418         drop = TRUE;
 419       }
 420       break;
 421     case GST_EVENT_SEGMENT:
 422       gst_event_copy_segment (event, &self->segment);
 423       break;
 424     default:
 425       break;
 426   }
 427
 428 out:
 429   if (drop)
 430     gst_event_unref (event);
 431   else if (event)
 432     ret = gst_pad_event_default (pad, parent, event);
 433
 434   return ret;
 435 }
 436
 437 static void
 438 gst_rtp_onvif_timestamp_init (GstRtpOnvifTimestamp * self)
 439 {
 440   self->sinkpad =
 441       gst_pad_new_from_static_template (&sink_template_factory, "sink");
 442   gst_pad_set_chain_function (self->sinkpad, gst_rtp_onvif_timestamp_chain);
 443   gst_pad_set_chain_list_function (self->sinkpad,
 444       gst_rtp_onvif_timestamp_chain_list);
 445   gst_pad_set_event_function (self->sinkpad,
 446       gst_rtp_onvif_timestamp_sink_event);
 447   gst_element_add_pad (GST_ELEMENT (self), self->sinkpad);
 448   GST_PAD_SET_PROXY_CAPS (self->sinkpad);
 449   GST_PAD_SET_PROXY_ALLOCATION (self->sinkpad);
 450
 451   self->srcpad =
 452       gst_pad_new_from_static_template (&src_template_factory, "src");
 453   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
 454
 455   self->prop_ntp_offset = DEFAULT_NTP_OFFSET;
 456   self->prop_set_e_bit = DEFAULT_SET_E_BIT;
 457   self->prop_set_t_bit = DEFAULT_SET_T_BIT;
 458   self->prop_drop_out_of_segment = DEFAULT_DROP_OUT_OF_SEGMENT;
 459
 460   gst_segment_init (&self->segment, GST_FORMAT_UNDEFINED);
 461
 462   self->event_queue = g_queue_new ();
 463   self->buffer = NULL;
 464   self->list = NULL;
 465 }
 466
 467 #define EXTENSION_ID 0xABAC
 468 #define EXTENSION_SIZE 3
 469
 470 static gboolean
 471 handle_buffer (GstRtpOnvifTimestamp * self, GstBuffer * buf)
 472 {
 473   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 474   guint8 *data;
 475   guint16 bits;
 476   guint wordlen;
 477   guint64 time;
 478   guint8 field = 0;
 479
 480   if (!GST_CLOCK_TIME_IS_VALID (self->ntp_offset)) {
 481     GstClock *clock = gst_element_get_clock (GST_ELEMENT (self));
 482
 483     if (clock) {
 484       GstClockTime clock_time = gst_clock_get_time (clock);
 485       guint64 real_time = g_get_real_time ();
 486       GstClockTime running_time = clock_time -
 487           gst_element_get_base_time (GST_ELEMENT (self));
 488
 489       /* convert microseconds to nanoseconds */
 490       real_time *= 1000;
 491
 492       /* add constant to convert from 1970 based time to 1900 based time */
 493       real_time += (G_GUINT64_CONSTANT (2208988800) * GST_SECOND);
 494
 495       self->ntp_offset = real_time - running_time;
 496
 497       GST_DEBUG_OBJECT (self, "new ntp-offset: %" GST_TIME_FORMAT,
 498           GST_TIME_ARGS (self->ntp_offset));
 499
 500       gst_object_unref (clock);
 501     } else {
 502       GST_ELEMENT_ERROR (self, STREAM, FAILED, ("No ntp-offset present"),
 503           ("Can not guess ntp-offset with no clock."));
 504       /* Received a buffer in PAUSED, so we can't guess the match
 505        * between the running time and the NTP clock yet.
 506        */
 507       return FALSE;
 508     }
 509   }
 510
 511   if (self->segment.format != GST_FORMAT_TIME) {
 512     GST_ELEMENT_ERROR (self, STREAM, FAILED,
 513         ("did not receive a time segment yet"), (NULL));
 514     return FALSE;
 515   }
 516
 517   if (!gst_rtp_buffer_map (buf, GST_MAP_READWRITE, &rtp)) {
 518     GST_ELEMENT_ERROR (self, STREAM, FAILED,
 519         ("Failed to map RTP buffer"), (NULL));
 520     return FALSE;
 521   }
 522
 523   if (!gst_rtp_buffer_set_extension_data (&rtp, EXTENSION_ID, EXTENSION_SIZE)) {
 524     GST_ELEMENT_ERROR (self, STREAM, FAILED, ("Failed to set extension data"),
 525         (NULL));
 526     gst_rtp_buffer_unmap (&rtp);
 527     return FALSE;
 528   }
 529
 530   if (!gst_rtp_buffer_get_extension_data (&rtp, &bits, (gpointer) & data,
 531           &wordlen)) {
 532     GST_ELEMENT_ERROR (self, STREAM, FAILED, ("Failed to get extension data"),
 533         (NULL));
 534     gst_rtp_buffer_unmap (&rtp);
 535     return FALSE;
 536   }
 537
 538   /* NTP timestamp */
 539   if (GST_BUFFER_PTS_IS_VALID (buf)) {
 540     time = gst_segment_to_stream_time (&self->segment, GST_FORMAT_TIME,
 541         GST_BUFFER_PTS (buf));
 542   } else if (GST_BUFFER_DTS_IS_VALID (buf)) {
 543     time = gst_segment_to_stream_time (&self->segment, GST_FORMAT_TIME,
 544         GST_BUFFER_DTS (buf));
 545   } else {
 546     GST_INFO_OBJECT (self,
 547         "Buffer doesn't contain any valid DTS or PTS timestamp");
 548     goto done;
 549   }
 550
 551   if (self->prop_drop_out_of_segment && time == GST_CLOCK_TIME_NONE) {
 552     GST_ERROR_OBJECT (self, "Failed to get stream time");
 553     gst_rtp_buffer_unmap (&rtp);
 554     return FALSE;
 555   }
 556
 557   /* add the offset (in seconds) */
 558   if (time != GST_CLOCK_TIME_NONE) {
 559     time += self->ntp_offset;
 560     /* convert to NTP time. upper 32 bits should contain the seconds
 561      * and the lower 32 bits, the fractions of a second. */
 562     time = gst_util_uint64_scale (time, (G_GINT64_CONSTANT (1) << 32),
 563         GST_SECOND);
 564   }
 565
 566   GST_DEBUG_OBJECT (self, "timestamp: %" G_GUINT64_FORMAT, time);
 567
 568   GST_WRITE_UINT64_BE (data, time);
 569
 570   /* The next byte is composed of: C E D T mbz (4 bits) */
 571
 572   /* Set C if the buffer does *not* have the DELTA_UNIT flag as it means
 573    * that's a key frame (or 'clean point'). */
 574   if (!GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT)) {
 575     GST_DEBUG_OBJECT (self, "set C flag");
 576     field |= (1 << 7);
 577   }
 578
 579   /* Set E if this the last buffer of a contiguous section of recording */
 580   if (self->set_e_bit) {
 581     GST_DEBUG_OBJECT (self, "set E flag");
 582     field |= (1 << 6);
 583     self->set_e_bit = FALSE;
 584   }
 585
 586   /* Set D if the buffer has the DISCONT flag */
 587   if (self->set_d_bit || GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DISCONT)) {
 588     GST_DEBUG_OBJECT (self, "set D flag");
 589     field |= (1 << 5);
 590     self->set_d_bit = FALSE;
 591   }
 592
 593   /* Set T if we have received EOS */
 594   if (self->set_t_bit) {
 595     GST_DEBUG_OBJECT (self, "set T flag");
 596     field |= (1 << 4);
 597     self->set_t_bit = FALSE;
 598   }
 599
 600   GST_WRITE_UINT8 (data + 8, field);
 601
 602   /* CSeq (low-order byte) */
 603   GST_WRITE_UINT8 (data + 9, (guchar) self->prop_cseq);
 604
 605   memset (data + 10, 0, 3);
 606
 607 done:
 608   gst_rtp_buffer_unmap (&rtp);
 609   return TRUE;
 610 }
 611
 612 /* @buf: (transfer full) */
 613 static GstFlowReturn
 614 handle_and_push_buffer (GstRtpOnvifTimestamp * self, GstBuffer * buf)
 615 {
 616   if (!handle_buffer (self, buf)) {
 617     gst_buffer_unref (buf);
 618     return GST_FLOW_ERROR;
 619   }
 620
 621   return gst_pad_push (self->srcpad, buf);
 622 }
 623
 624 static GstFlowReturn
 625 gst_rtp_onvif_timestamp_chain (GstPad * pad, GstObject * parent,
 626     GstBuffer * buf)
 627 {
 628   GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (parent);
 629   GstFlowReturn result = GST_FLOW_OK;
 630
 631   if (!self->prop_set_e_bit && !self->prop_set_t_bit) {
 632     /* Modify and push this buffer right away */
 633     return handle_and_push_buffer (self, buf);
 634   }
 635
 636   /* send any previously cached item(s), this leaves an empty queue */
 637   result = send_cached_buffer_and_events (self);
 638
 639   /* enqueue the new item, as the only item in the queue */
 640   self->buffer = buf;
 641   return result;
 642 }
 643
 644 /* @buf: (transfer full) */
 645 static GstFlowReturn
 646 handle_and_push_buffer_list (GstRtpOnvifTimestamp * self, GstBufferList * list)
 647 {
 648   GstBuffer *buf;
 649
 650   /* Set the extension on the *first* buffer */
 651   buf = gst_buffer_list_get (list, 0);
 652   if (!handle_buffer (self, buf)) {
 653     gst_buffer_list_unref (list);
 654     return GST_FLOW_ERROR;
 655   }
 656
 657   return gst_pad_push_list (self->srcpad, list);
 658 }
 659
 660 /* gst_pad_chain_list_default() refs the buffer when passing it to the chain
 661  * function, making it not writable. We implement our own chain_list function
 662  * to avoid having to copy each buffer. */
 663 static GstFlowReturn
 664 gst_rtp_onvif_timestamp_chain_list (GstPad * pad, GstObject * parent,
 665     GstBufferList * list)
 666 {
 667   GstRtpOnvifTimestamp *self = GST_RTP_ONVIF_TIMESTAMP (parent);
 668   GstFlowReturn result = GST_FLOW_OK;
 669
 670   if (!self->prop_set_e_bit && !self->prop_set_t_bit) {
 671     return handle_and_push_buffer_list (self, list);
 672   }
 673
 674   /* send any previously cached item(s), this leaves an empty queue */
 675   result = send_cached_buffer_and_events (self);
 676
 677   /* enqueue the new item, as the only item in the queue */
 678   self->list = list;
 679   return result;
 680 }