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