subprojects/gst-plugins-bad/gst/mpegtsdemux/mpegtspacketizer.c

   1 /*
   2  * mpegtspacketizer.c -
   3  * Copyright (C) 2007, 2008 Alessandro Decina, Zaheer Merali
   4  *
   5  * Authors:
   6  *   Zaheer Merali <zaheerabbas at merali dot org>
   7  *   Alessandro Decina <alessandro@nnva.org>
   8  *
   9  * This library is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Library General Public
  11  * License as published by the Free Software Foundation; either
  12  * version 2 of the License, or (at your option) any later version.
  13  *
  14  * This library is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * Library General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Library General Public
  20  * License along with this library; if not, write to the
  21  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  22  * Boston, MA 02110-1301, USA.
  23  */
  24 #ifdef HAVE_CONFIG_H
  25 #include "config.h"
  26 #endif
  27
  28 #include <string.h>
  29 #include <stdlib.h>
  30
  31 /* Skew calculation pameters */
  32 #define MAX_TIME        (2 * GST_SECOND)
  33
  34 /* maximal PCR time */
  35 #define PCR_MAX_VALUE (((((guint64)1)<<33) * 300) + 298)
  36 #define PCR_GST_MAX_VALUE (PCR_MAX_VALUE * GST_MSECOND / (PCR_MSECOND))
  37 #define PTS_DTS_MAX_VALUE (((guint64)1) << 33)
  38
  39 #include "mpegtspacketizer.h"
  40 #include "gstmpegdesc.h"
  41
  42 GST_DEBUG_CATEGORY_STATIC (mpegts_packetizer_debug);
  43 #define GST_CAT_DEFAULT mpegts_packetizer_debug
  44
  45 static void _init_local (void);
  46 G_DEFINE_TYPE_EXTENDED (MpegTSPacketizer2, mpegts_packetizer, G_TYPE_OBJECT, 0,
  47     _init_local ());
  48
  49 #define ABSDIFF(a,b) ((a) < (b) ? (b) - (a) : (a) - (b))
  50
  51 #define PACKETIZER_GROUP_LOCK(p) g_mutex_lock(&((p)->group_lock))
  52 #define PACKETIZER_GROUP_UNLOCK(p) g_mutex_unlock(&((p)->group_lock))
  53
  54 static void mpegts_packetizer_dispose (GObject * object);
  55 static void mpegts_packetizer_finalize (GObject * object);
  56 static GstClockTime calculate_skew (MpegTSPacketizer2 * packetizer,
  57     MpegTSPCR * pcr, guint64 pcrtime, GstClockTime time);
  58 static void _close_current_group (MpegTSPCR * pcrtable);
  59 static void record_pcr (MpegTSPacketizer2 * packetizer, MpegTSPCR * pcrtable,
  60     guint64 pcr, guint64 offset);
  61
  62 #define CONTINUITY_UNSET 255
  63 #define VERSION_NUMBER_UNSET 255
  64 #define TABLE_ID_UNSET 0xFF
  65 #define PACKET_SYNC_BYTE 0x47
  66
  67 static inline MpegTSPCR *
  68 get_pcr_table (MpegTSPacketizer2 * packetizer, guint16 pid)
  69 {
  70   MpegTSPCR *res;
  71
  72   res = packetizer->observations[packetizer->pcrtablelut[pid]];
  73
  74   if (G_UNLIKELY (res == NULL)) {
  75     /* If we don't have a PCR table for the requested PID, create one .. */
  76     res = g_new0 (MpegTSPCR, 1);
  77     /* Add it to the last table position */
  78     packetizer->observations[packetizer->lastobsid] = res;
  79     /* Update the pcrtablelut */
  80     packetizer->pcrtablelut[pid] = packetizer->lastobsid;
  81     /* And increment the last know slot */
  82     packetizer->lastobsid++;
  83
  84     /* Finally set the default values */
  85     res->pid = pid;
  86     res->base_time = GST_CLOCK_TIME_NONE;
  87     res->base_pcrtime = GST_CLOCK_TIME_NONE;
  88     res->last_pcrtime = GST_CLOCK_TIME_NONE;
  89     res->window_pos = 0;
  90     res->window_filling = TRUE;
  91     res->window_min = 0;
  92     res->skew = 0;
  93     res->prev_send_diff = GST_CLOCK_TIME_NONE;
  94     res->prev_out_time = GST_CLOCK_TIME_NONE;
  95     res->pcroffset = 0;
  96
  97     res->current = g_slice_new0 (PCROffsetCurrent);
  98   }
  99
 100   return res;
 101 }
 102
 103 static void
 104 pcr_offset_group_free (PCROffsetGroup * group)
 105 {
 106   g_free (group->values);
 107   g_slice_free (PCROffsetGroup, group);
 108 }
 109
 110 static void
 111 flush_observations (MpegTSPacketizer2 * packetizer)
 112 {
 113   gint i;
 114
 115   for (i = 0; i < packetizer->lastobsid; i++) {
 116     g_list_free_full (packetizer->observations[i]->groups,
 117         (GDestroyNotify) pcr_offset_group_free);
 118     if (packetizer->observations[i]->current)
 119       g_slice_free (PCROffsetCurrent, packetizer->observations[i]->current);
 120     g_free (packetizer->observations[i]);
 121     packetizer->observations[i] = NULL;
 122   }
 123   memset (packetizer->pcrtablelut, 0xff, 0x2000);
 124   packetizer->lastobsid = 0;
 125 }
 126
 127 GstClockTime
 128 mpegts_packetizer_get_current_time (MpegTSPacketizer2 * packetizer,
 129     guint16 pcr_pid)
 130 {
 131   MpegTSPCR *pcrtable = get_pcr_table (packetizer, pcr_pid);
 132
 133   if (pcrtable == NULL)
 134     return GST_CLOCK_TIME_NONE;
 135
 136   return mpegts_packetizer_pts_to_ts (packetizer, pcrtable->last_pcrtime,
 137       pcr_pid);
 138 }
 139
 140 static inline MpegTSPacketizerStreamSubtable *
 141 find_subtable (GSList * subtables, guint8 table_id, guint16 subtable_extension)
 142 {
 143   GSList *tmp;
 144
 145   /* FIXME: Make this an array ! */
 146   for (tmp = subtables; tmp; tmp = tmp->next) {
 147     MpegTSPacketizerStreamSubtable *sub =
 148         (MpegTSPacketizerStreamSubtable *) tmp->data;
 149     if (sub->table_id == table_id
 150         && sub->subtable_extension == subtable_extension)
 151       return sub;
 152   }
 153
 154   return NULL;
 155 }
 156
 157 static gboolean
 158 seen_section_before (MpegTSPacketizerStream * stream, guint8 table_id,
 159     guint16 subtable_extension, guint8 version_number, guint8 section_number,
 160     guint8 last_section_number, guint8 * data_start, gsize to_read)
 161 {
 162   MpegTSPacketizerStreamSubtable *subtable;
 163
 164   /* Check if we've seen this table_id/subtable_extension first */
 165   subtable = find_subtable (stream->subtables, table_id, subtable_extension);
 166   if (!subtable) {
 167     GST_DEBUG ("Haven't seen subtable");
 168     return FALSE;
 169   }
 170   /* If we have, check it has the same version_number */
 171   if (subtable->version_number != version_number) {
 172     GST_DEBUG ("Different version number");
 173     return FALSE;
 174   }
 175   /* Did the number of sections change ? */
 176   if (subtable->last_section_number != last_section_number) {
 177     GST_DEBUG ("Different last_section_number");
 178     return FALSE;
 179   }
 180   /* Finally return whether we saw that section or not */
 181   if (!MPEGTS_BIT_IS_SET (subtable->seen_section, section_number)) {
 182     GST_DEBUG ("Different section_number");
 183     return FALSE;
 184   }
 185
 186   if (stream->section_data) {
 187     /* Everything else is the same, fall back to memcmp */
 188     return (memcmp (stream->section_data, data_start, to_read) != 0);
 189   }
 190
 191   return FALSE;
 192 }
 193
 194 static MpegTSPacketizerStreamSubtable *
 195 mpegts_packetizer_stream_subtable_new (guint8 table_id,
 196     guint16 subtable_extension, guint8 last_section_number)
 197 {
 198   MpegTSPacketizerStreamSubtable *subtable;
 199
 200   subtable = g_new0 (MpegTSPacketizerStreamSubtable, 1);
 201   subtable->version_number = VERSION_NUMBER_UNSET;
 202   subtable->table_id = table_id;
 203   subtable->subtable_extension = subtable_extension;
 204   subtable->last_section_number = last_section_number;
 205   return subtable;
 206 }
 207
 208 static MpegTSPacketizerStream *
 209 mpegts_packetizer_stream_new (guint16 pid)
 210 {
 211   MpegTSPacketizerStream *stream;
 212
 213   stream = (MpegTSPacketizerStream *) g_new0 (MpegTSPacketizerStream, 1);
 214   stream->continuity_counter = CONTINUITY_UNSET;
 215   stream->subtables = NULL;
 216   stream->table_id = TABLE_ID_UNSET;
 217   stream->pid = pid;
 218   return stream;
 219 }
 220
 221 static void
 222 mpegts_packetizer_clear_section (MpegTSPacketizerStream * stream)
 223 {
 224   stream->continuity_counter = CONTINUITY_UNSET;
 225   stream->section_length = 0;
 226   stream->section_offset = 0;
 227   stream->table_id = TABLE_ID_UNSET;
 228   g_free (stream->section_data);
 229   stream->section_data = NULL;
 230 }
 231
 232 static void
 233 mpegts_packetizer_stream_subtable_free (MpegTSPacketizerStreamSubtable *
 234     subtable)
 235 {
 236   g_free (subtable);
 237 }
 238
 239 static void
 240 mpegts_packetizer_stream_free (MpegTSPacketizerStream * stream)
 241 {
 242   mpegts_packetizer_clear_section (stream);
 243   g_slist_foreach (stream->subtables,
 244       (GFunc) mpegts_packetizer_stream_subtable_free, NULL);
 245   g_slist_free (stream->subtables);
 246   g_free (stream);
 247 }
 248
 249 static void
 250 mpegts_packetizer_class_init (MpegTSPacketizer2Class * klass)
 251 {
 252   GObjectClass *gobject_class;
 253
 254   gobject_class = G_OBJECT_CLASS (klass);
 255
 256   gobject_class->dispose = mpegts_packetizer_dispose;
 257   gobject_class->finalize = mpegts_packetizer_finalize;
 258 }
 259
 260 static void
 261 mpegts_packetizer_init (MpegTSPacketizer2 * packetizer)
 262 {
 263   g_mutex_init (&packetizer->group_lock);
 264
 265   packetizer->adapter = gst_adapter_new ();
 266   packetizer->offset = 0;
 267   packetizer->empty = TRUE;
 268   packetizer->streams = g_new0 (MpegTSPacketizerStream *, 8192);
 269   packetizer->packet_size = 0;
 270   packetizer->calculate_skew = FALSE;
 271   packetizer->calculate_offset = FALSE;
 272
 273   packetizer->map_data = NULL;
 274   packetizer->map_size = 0;
 275   packetizer->map_offset = 0;
 276   packetizer->need_sync = FALSE;
 277
 278   memset (packetizer->pcrtablelut, 0xff, 0x2000);
 279   memset (packetizer->observations, 0x0, sizeof (packetizer->observations));
 280   packetizer->lastobsid = 0;
 281
 282   packetizer->nb_seen_offsets = 0;
 283   packetizer->refoffset = -1;
 284   packetizer->last_in_time = GST_CLOCK_TIME_NONE;
 285   packetizer->pcr_discont_threshold = GST_SECOND;
 286   packetizer->last_pts = GST_CLOCK_TIME_NONE;
 287   packetizer->last_dts = GST_CLOCK_TIME_NONE;
 288   packetizer->extra_shift = 0;
 289 }
 290
 291 static void
 292 mpegts_packetizer_dispose (GObject * object)
 293 {
 294   MpegTSPacketizer2 *packetizer = GST_MPEGTS_PACKETIZER (object);
 295
 296   if (!packetizer->disposed) {
 297     if (packetizer->packet_size)
 298       packetizer->packet_size = 0;
 299     if (packetizer->streams) {
 300       int i;
 301       for (i = 0; i < 8192; i++) {
 302         if (packetizer->streams[i])
 303           mpegts_packetizer_stream_free (packetizer->streams[i]);
 304       }
 305       g_free (packetizer->streams);
 306     }
 307
 308     gst_adapter_clear (packetizer->adapter);
 309     g_object_unref (packetizer->adapter);
 310     g_mutex_clear (&packetizer->group_lock);
 311     packetizer->disposed = TRUE;
 312     packetizer->offset = 0;
 313     packetizer->empty = TRUE;
 314
 315     flush_observations (packetizer);
 316   }
 317
 318   if (G_OBJECT_CLASS (mpegts_packetizer_parent_class)->dispose)
 319     G_OBJECT_CLASS (mpegts_packetizer_parent_class)->dispose (object);
 320 }
 321
 322 static void
 323 mpegts_packetizer_finalize (GObject * object)
 324 {
 325   if (G_OBJECT_CLASS (mpegts_packetizer_parent_class)->finalize)
 326     G_OBJECT_CLASS (mpegts_packetizer_parent_class)->finalize (object);
 327 }
 328
 329 static inline guint64
 330 mpegts_packetizer_compute_pcr (const guint8 * data)
 331 {
 332   guint32 pcr1;
 333   guint16 pcr2;
 334   guint64 pcr, pcr_ext;
 335
 336   pcr1 = GST_READ_UINT32_BE (data);
 337   pcr2 = GST_READ_UINT16_BE (data + 4);
 338   pcr = ((guint64) pcr1) << 1;
 339   pcr |= (pcr2 & 0x8000) >> 15;
 340   pcr_ext = (pcr2 & 0x01ff);
 341   return pcr * 300 + pcr_ext % 300;
 342 }
 343
 344 static gboolean
 345 mpegts_packetizer_parse_adaptation_field_control (MpegTSPacketizer2 *
 346     packetizer, MpegTSPacketizerPacket * packet)
 347 {
 348   guint8 length, afcflags;
 349   guint8 *data;
 350
 351   length = *packet->data++;
 352
 353   /* an adaptation field with length 0 is valid and
 354    * can be used to insert a single stuffing byte */
 355   if (!length) {
 356     packet->afc_flags = 0;
 357     return TRUE;
 358   }
 359
 360   if ((packet->scram_afc_cc & 0x30) == 0x20) {
 361     /* no payload, adaptation field of 183 bytes */
 362     if (length > 183) {
 363       GST_WARNING ("PID 0x%04x afc == 0x%02x and length %d > 183",
 364           packet->pid, packet->scram_afc_cc & 0x30, length);
 365       return FALSE;
 366     }
 367     if (length != 183) {
 368       GST_WARNING ("PID 0x%04x afc == 0x%02x and length %d != 183",
 369           packet->pid, packet->scram_afc_cc & 0x30, length);
 370       GST_MEMDUMP ("Unknown payload", packet->data + length,
 371           packet->data_end - packet->data - length);
 372     }
 373   } else if (length == 183) {
 374     /* Note: According to the specification, the adaptation field length
 375      * must be 183 if there is no payload data and < 183 if the packet
 376      * contains an adaptation field and payload data.
 377      * Some payloaders always set the flag for payload data, even if the
 378      * adaptation field length is 183. This just means a zero length
 379      * payload so we clear the payload flag here and continue.
 380      */
 381     GST_WARNING ("PID 0x%04x afc == 0x%02x and length %d == 183 (ignored)",
 382         packet->pid, packet->scram_afc_cc & 0x30, length);
 383     packet->scram_afc_cc &= ~0x10;
 384   } else if (length > 182) {
 385     GST_WARNING ("PID 0x%04x afc == 0x%02x and length %d > 182",
 386         packet->pid, packet->scram_afc_cc & 0x30, length);
 387     return FALSE;
 388   }
 389
 390   if (packet->data + length > packet->data_end) {
 391     GST_DEBUG
 392         ("PID 0x%04x afc length %d overflows the buffer current %d max %d",
 393         packet->pid, length, (gint) (packet->data - packet->data_start),
 394         (gint) (packet->data_end - packet->data_start));
 395     return FALSE;
 396   }
 397
 398   data = packet->data;
 399   packet->data += length;
 400
 401   afcflags = packet->afc_flags = *data++;
 402
 403   GST_DEBUG ("flags: %s%s%s%s%s%s%s%s%s",
 404       afcflags & 0x80 ? "discontinuity " : "",
 405       afcflags & 0x40 ? "random_access " : "",
 406       afcflags & 0x20 ? "elementary_stream_priority " : "",
 407       afcflags & 0x10 ? "PCR " : "",
 408       afcflags & 0x08 ? "OPCR " : "",
 409       afcflags & 0x04 ? "splicing_point " : "",
 410       afcflags & 0x02 ? "transport_private_data " : "",
 411       afcflags & 0x01 ? "extension " : "", afcflags == 0x00 ? "<none>" : "");
 412
 413   /* PCR */
 414   if (afcflags & MPEGTS_AFC_PCR_FLAG) {
 415     MpegTSPCR *pcrtable = NULL;
 416     packet->pcr = mpegts_packetizer_compute_pcr (data);
 417     data += 6;
 418     GST_DEBUG ("pcr 0x%04x %" G_GUINT64_FORMAT " (%" GST_TIME_FORMAT
 419         ") offset:%" G_GUINT64_FORMAT, packet->pid, packet->pcr,
 420         GST_TIME_ARGS (PCRTIME_TO_GSTTIME (packet->pcr)), packet->offset);
 421
 422     PACKETIZER_GROUP_LOCK (packetizer);
 423     if (packetizer->calculate_skew
 424         && GST_CLOCK_TIME_IS_VALID (packetizer->last_in_time)) {
 425       pcrtable = get_pcr_table (packetizer, packet->pid);
 426       calculate_skew (packetizer, pcrtable, packet->pcr,
 427           packetizer->last_in_time);
 428     }
 429     if (packetizer->calculate_offset) {
 430       if (!pcrtable)
 431         pcrtable = get_pcr_table (packetizer, packet->pid);
 432       record_pcr (packetizer, pcrtable, packet->pcr, packet->offset);
 433     }
 434     PACKETIZER_GROUP_UNLOCK (packetizer);
 435   }
 436 #ifndef GST_DISABLE_GST_DEBUG
 437   /* OPCR */
 438   if (afcflags & MPEGTS_AFC_OPCR_FLAG) {
 439     /* Note: We don't use/need opcr for the time being */
 440     guint64 opcr = mpegts_packetizer_compute_pcr (data);
 441     data += 6;
 442     GST_DEBUG ("opcr %" G_GUINT64_FORMAT " (%" GST_TIME_FORMAT ")",
 443         opcr, GST_TIME_ARGS (PCRTIME_TO_GSTTIME (opcr)));
 444   }
 445
 446   if (afcflags & MPEGTS_AFC_SPLICING_POINT_FLAG) {
 447     GST_DEBUG ("splice_countdown: %u", *data++);
 448   }
 449
 450   if (afcflags & MPEGTS_AFC_TRANSPORT_PRIVATE_DATA_FLAG) {
 451     guint8 len = *data++;
 452     GST_MEMDUMP ("private data", data, len);
 453     data += len;
 454   }
 455
 456   if (afcflags & MPEGTS_AFC_EXTENSION_FLAG) {
 457     guint8 extlen = *data++;
 458     guint8 flags = *data++;
 459     GST_DEBUG ("extension size:%d flags : %s%s%s", extlen,
 460         flags & 0x80 ? "ltw " : "",
 461         flags & 0x40 ? "piecewise_rate " : "",
 462         flags & 0x20 ? "seamless_splice " : "");
 463     if (flags & 0x80) {
 464       GST_DEBUG ("legal time window: valid_flag:%d offset:%d", *data >> 7,
 465           GST_READ_UINT16_BE (data) & 0x7fff);
 466       data += 2;
 467     }
 468   }
 469 #endif
 470
 471   return TRUE;
 472 }
 473
 474 static MpegTSPacketizerPacketReturn
 475 mpegts_packetizer_parse_packet (MpegTSPacketizer2 * packetizer,
 476     MpegTSPacketizerPacket * packet)
 477 {
 478   guint8 *data;
 479   guint8 tmp;
 480
 481   data = packet->data_start;
 482   data += 1;
 483   tmp = *data;
 484
 485   /* transport_error_indicator 1 */
 486   if (G_UNLIKELY (tmp & 0x80))
 487     return PACKET_BAD;
 488
 489   /* payload_unit_start_indicator 1 */
 490   packet->payload_unit_start_indicator = tmp & 0x40;
 491
 492   /* transport_priority 1 */
 493   /* PID 13 */
 494   packet->pid = GST_READ_UINT16_BE (data) & 0x1FFF;
 495   data += 2;
 496
 497   packet->scram_afc_cc = tmp = *data++;
 498   /* transport_scrambling_control 2 */
 499   if (G_UNLIKELY (tmp & 0xc0))
 500     return PACKET_BAD;
 501
 502   packet->data = data;
 503
 504   packet->afc_flags = 0;
 505   packet->pcr = G_MAXUINT64;
 506
 507   if (FLAGS_HAS_AFC (tmp)) {
 508     if (!mpegts_packetizer_parse_adaptation_field_control (packetizer, packet))
 509       return FALSE;
 510   }
 511
 512   if (FLAGS_HAS_PAYLOAD (packet->scram_afc_cc))
 513     packet->payload = packet->data;
 514   else
 515     packet->payload = NULL;
 516
 517   return PACKET_OK;
 518 }
 519
 520 static GstMpegtsSection *
 521 mpegts_packetizer_parse_section_header (MpegTSPacketizer2 * packetizer,
 522     MpegTSPacketizerStream * stream)
 523 {
 524   MpegTSPacketizerStreamSubtable *subtable;
 525   GstMpegtsSection *res;
 526
 527   subtable =
 528       find_subtable (stream->subtables, stream->table_id,
 529       stream->subtable_extension);
 530   if (subtable) {
 531     GST_DEBUG ("Found previous subtable_extension:0x%04x",
 532         stream->subtable_extension);
 533     if (G_UNLIKELY (stream->version_number != subtable->version_number)) {
 534       /* If the version number changed, reset the subtable */
 535       subtable->version_number = stream->version_number;
 536       subtable->last_section_number = stream->last_section_number;
 537       memset (subtable->seen_section, 0, 32);
 538     }
 539   } else {
 540     GST_DEBUG ("Appending new subtable_extension: 0x%04x",
 541         stream->subtable_extension);
 542     subtable = mpegts_packetizer_stream_subtable_new (stream->table_id,
 543         stream->subtable_extension, stream->last_section_number);
 544     subtable->version_number = stream->version_number;
 545
 546     stream->subtables = g_slist_prepend (stream->subtables, subtable);
 547   }
 548
 549   GST_MEMDUMP ("Full section data", stream->section_data,
 550       stream->section_length);
 551   /* TODO ? : Replace this by an efficient version (where we provide all
 552    * pre-parsed header data) */
 553   res =
 554       gst_mpegts_section_new (stream->pid, stream->section_data,
 555       stream->section_length);
 556   stream->section_data = NULL;
 557   mpegts_packetizer_clear_section (stream);
 558
 559   if (res) {
 560     /* NOTE : Due to the new mpegts-si system, There is a insanely low probability
 561      * that we might have gotten a section that was corrupted (i.e. wrong crc)
 562      * and that we consider it as seen.
 563      *
 564      * The reason why we consider this as acceptable is because all the previous
 565      * checks were already done:
 566      * * transport layer checks (DVB)
 567      * * 0x47 validation
 568      * * continuity counter validation
 569      * * subtable validation
 570      * * section_number validation
 571      * * section_length validation
 572      *
 573      * The probability of this happening vs the overhead of doing CRC checks
 574      * on all sections (including those we would not use) is just not worth it.
 575      * */
 576     MPEGTS_BIT_SET (subtable->seen_section, stream->section_number);
 577     res->offset = stream->offset;
 578   }
 579
 580   return res;
 581 }
 582
 583 void
 584 mpegts_packetizer_clear (MpegTSPacketizer2 * packetizer)
 585 {
 586   guint i;
 587   MpegTSPCR *pcrtable;
 588
 589   packetizer->packet_size = 0;
 590
 591   if (packetizer->streams) {
 592     int i;
 593     for (i = 0; i < 8192; i++) {
 594       if (packetizer->streams[i]) {
 595         mpegts_packetizer_stream_free (packetizer->streams[i]);
 596       }
 597     }
 598     memset (packetizer->streams, 0, 8192 * sizeof (MpegTSPacketizerStream *));
 599   }
 600
 601   gst_adapter_clear (packetizer->adapter);
 602   packetizer->offset = 0;
 603   packetizer->empty = TRUE;
 604   packetizer->need_sync = FALSE;
 605   packetizer->map_data = NULL;
 606   packetizer->map_size = 0;
 607   packetizer->map_offset = 0;
 608   packetizer->last_in_time = GST_CLOCK_TIME_NONE;
 609   packetizer->last_pts = GST_CLOCK_TIME_NONE;
 610   packetizer->last_dts = GST_CLOCK_TIME_NONE;
 611
 612   pcrtable = packetizer->observations[packetizer->pcrtablelut[0x1fff]];
 613   if (pcrtable)
 614     pcrtable->base_time = GST_CLOCK_TIME_NONE;
 615
 616   /* Close current PCR group */
 617   PACKETIZER_GROUP_LOCK (packetizer);
 618
 619   for (i = 0; i < MAX_PCR_OBS_CHANNELS; i++) {
 620     if (packetizer->observations[i])
 621       _close_current_group (packetizer->observations[i]);
 622     else
 623       break;
 624   }
 625   PACKETIZER_GROUP_UNLOCK (packetizer);
 626 }
 627
 628 void
 629 mpegts_packetizer_flush (MpegTSPacketizer2 * packetizer, gboolean hard)
 630 {
 631   guint i;
 632   MpegTSPCR *pcrtable;
 633   GST_DEBUG ("Flushing");
 634
 635   if (packetizer->streams) {
 636     for (i = 0; i < 8192; i++) {
 637       if (packetizer->streams[i]) {
 638         mpegts_packetizer_clear_section (packetizer->streams[i]);
 639       }
 640     }
 641   }
 642   gst_adapter_clear (packetizer->adapter);
 643
 644   packetizer->offset = 0;
 645   packetizer->empty = TRUE;
 646   packetizer->need_sync = FALSE;
 647   packetizer->map_data = NULL;
 648   packetizer->map_size = 0;
 649   packetizer->map_offset = 0;
 650   packetizer->last_in_time = GST_CLOCK_TIME_NONE;
 651   packetizer->last_pts = GST_CLOCK_TIME_NONE;
 652   packetizer->last_dts = GST_CLOCK_TIME_NONE;
 653
 654   pcrtable = packetizer->observations[packetizer->pcrtablelut[0x1fff]];
 655   if (pcrtable)
 656     pcrtable->base_time = GST_CLOCK_TIME_NONE;
 657
 658   /* Close current PCR group */
 659   PACKETIZER_GROUP_LOCK (packetizer);
 660   for (i = 0; i < MAX_PCR_OBS_CHANNELS; i++) {
 661     if (packetizer->observations[i])
 662       _close_current_group (packetizer->observations[i]);
 663     else
 664       break;
 665   }
 666   PACKETIZER_GROUP_UNLOCK (packetizer);
 667
 668   if (hard) {
 669     /* For pull mode seeks in tsdemux the observation must be preserved */
 670     flush_observations (packetizer);
 671   }
 672 }
 673
 674 void
 675 mpegts_packetizer_remove_stream (MpegTSPacketizer2 * packetizer, gint16 pid)
 676 {
 677   MpegTSPacketizerStream *stream = packetizer->streams[pid];
 678   if (stream) {
 679     GST_INFO ("Removing stream for PID 0x%04x", pid);
 680     mpegts_packetizer_stream_free (stream);
 681     packetizer->streams[pid] = NULL;
 682   }
 683 }
 684
 685 MpegTSPacketizer2 *
 686 mpegts_packetizer_new (void)
 687 {
 688   MpegTSPacketizer2 *packetizer;
 689
 690   packetizer =
 691       GST_MPEGTS_PACKETIZER (g_object_new (GST_TYPE_MPEGTS_PACKETIZER, NULL));
 692
 693   return packetizer;
 694 }
 695
 696 void
 697 mpegts_packetizer_push (MpegTSPacketizer2 * packetizer, GstBuffer * buffer)
 698 {
 699   GstClockTime ts;
 700   if (G_UNLIKELY (packetizer->empty)) {
 701     packetizer->empty = FALSE;
 702     packetizer->offset = GST_BUFFER_OFFSET (buffer);
 703   }
 704
 705   GST_DEBUG ("Pushing %" G_GSIZE_FORMAT " byte from offset %"
 706       G_GUINT64_FORMAT, gst_buffer_get_size (buffer),
 707       GST_BUFFER_OFFSET (buffer));
 708   gst_adapter_push (packetizer->adapter, buffer);
 709   /* If the buffer has a valid timestamp, store it - preferring DTS,
 710    * which is where upstream arrival times should be stored */
 711   ts = GST_BUFFER_DTS_OR_PTS (buffer);
 712   if (GST_CLOCK_TIME_IS_VALID (ts))
 713     packetizer->last_in_time = ts;
 714   packetizer->last_pts = GST_BUFFER_PTS (buffer);
 715   packetizer->last_dts = GST_BUFFER_DTS (buffer);
 716 }
 717
 718 static void
 719 mpegts_packetizer_flush_bytes (MpegTSPacketizer2 * packetizer, gsize size)
 720 {
 721   if (size > 0) {
 722     GST_LOG ("flushing %" G_GSIZE_FORMAT " bytes from adapter", size);
 723     gst_adapter_flush (packetizer->adapter, size);
 724   }
 725
 726   packetizer->map_data = NULL;
 727   packetizer->map_size = 0;
 728   packetizer->map_offset = 0;
 729 }
 730
 731 static gboolean
 732 mpegts_packetizer_map (MpegTSPacketizer2 * packetizer, gsize size)
 733 {
 734   gsize available;
 735
 736   if (packetizer->map_size - packetizer->map_offset >= size)
 737     return TRUE;
 738
 739   mpegts_packetizer_flush_bytes (packetizer, packetizer->map_offset);
 740
 741   available = gst_adapter_available (packetizer->adapter);
 742   if (available < size)
 743     return FALSE;
 744
 745   packetizer->map_data =
 746       (guint8 *) gst_adapter_map (packetizer->adapter, available);
 747   if (!packetizer->map_data)
 748     return FALSE;
 749
 750   packetizer->map_size = available;
 751   packetizer->map_offset = 0;
 752
 753   GST_LOG ("mapped %" G_GSIZE_FORMAT " bytes from adapter", available);
 754
 755   return TRUE;
 756 }
 757
 758 static gboolean
 759 mpegts_try_discover_packet_size (MpegTSPacketizer2 * packetizer)
 760 {
 761   guint8 *data;
 762   gsize size, i, j;
 763
 764   static const guint psizes[] = {
 765     MPEGTS_NORMAL_PACKETSIZE,
 766     MPEGTS_M2TS_PACKETSIZE,
 767     MPEGTS_DVB_ASI_PACKETSIZE,
 768     MPEGTS_ATSC_PACKETSIZE
 769   };
 770
 771   if (!mpegts_packetizer_map (packetizer, 4 * MPEGTS_MAX_PACKETSIZE))
 772     return FALSE;
 773
 774   size = packetizer->map_size - packetizer->map_offset;
 775   data = packetizer->map_data + packetizer->map_offset;
 776
 777   for (i = 0; i + 3 * MPEGTS_MAX_PACKETSIZE < size; i++) {
 778     /* find a sync byte */
 779     if (data[i] != PACKET_SYNC_BYTE)
 780       continue;
 781
 782     /* check for 4 consecutive sync bytes with each possible packet size */
 783     for (j = 0; j < G_N_ELEMENTS (psizes); j++) {
 784       guint packet_size = psizes[j];
 785
 786       if (data[i + packet_size] == PACKET_SYNC_BYTE &&
 787           data[i + 2 * packet_size] == PACKET_SYNC_BYTE &&
 788           data[i + 3 * packet_size] == PACKET_SYNC_BYTE) {
 789         packetizer->packet_size = packet_size;
 790         goto out;
 791       }
 792     }
 793   }
 794
 795 out:
 796   packetizer->map_offset += i;
 797
 798   if (packetizer->packet_size == 0) {
 799     GST_DEBUG ("Could not determine packet size in %" G_GSIZE_FORMAT
 800         " bytes buffer, flush %" G_GSIZE_FORMAT " bytes", size, i);
 801     mpegts_packetizer_flush_bytes (packetizer, packetizer->map_offset);
 802     return FALSE;
 803   }
 804
 805   GST_INFO ("have packetsize detected: %u bytes", packetizer->packet_size);
 806
 807   if (packetizer->packet_size == MPEGTS_M2TS_PACKETSIZE &&
 808       packetizer->map_offset >= 4)
 809     packetizer->map_offset -= 4;
 810
 811   return TRUE;
 812 }
 813
 814 static gboolean
 815 mpegts_packetizer_sync (MpegTSPacketizer2 * packetizer)
 816 {
 817   gboolean found = FALSE;
 818   guint8 *data;
 819   guint packet_size;
 820   gsize size, sync_offset, i;
 821
 822   packet_size = packetizer->packet_size;
 823
 824   if (!mpegts_packetizer_map (packetizer, 3 * packet_size))
 825     return FALSE;
 826
 827   size = packetizer->map_size - packetizer->map_offset;
 828   data = packetizer->map_data + packetizer->map_offset;
 829
 830   if (packet_size == MPEGTS_M2TS_PACKETSIZE)
 831     sync_offset = 4;
 832   else
 833     sync_offset = 0;
 834
 835   for (i = sync_offset; i + 2 * packet_size < size; i++) {
 836     if (data[i] == PACKET_SYNC_BYTE &&
 837         data[i + packet_size] == PACKET_SYNC_BYTE &&
 838         data[i + 2 * packet_size] == PACKET_SYNC_BYTE) {
 839       found = TRUE;
 840       break;
 841     }
 842   }
 843
 844   packetizer->map_offset += i - sync_offset;
 845
 846   if (!found)
 847     mpegts_packetizer_flush_bytes (packetizer, packetizer->map_offset);
 848
 849   return found;
 850 }
 851
 852 MpegTSPacketizerPacketReturn
 853 mpegts_packetizer_next_packet (MpegTSPacketizer2 * packetizer,
 854     MpegTSPacketizerPacket * packet)
 855 {
 856   guint8 *packet_data;
 857   guint packet_size;
 858   gsize sync_offset;
 859
 860   packet_size = packetizer->packet_size;
 861   if (G_UNLIKELY (!packet_size)) {
 862     if (!mpegts_try_discover_packet_size (packetizer))
 863       return PACKET_NEED_MORE;
 864     packet_size = packetizer->packet_size;
 865   }
 866
 867   /* M2TS packets don't start with the sync byte, all other variants do */
 868   if (packet_size == MPEGTS_M2TS_PACKETSIZE)
 869     sync_offset = 4;
 870   else
 871     sync_offset = 0;
 872
 873   while (1) {
 874     if (packetizer->need_sync) {
 875       if (!mpegts_packetizer_sync (packetizer))
 876         return PACKET_NEED_MORE;
 877       packetizer->need_sync = FALSE;
 878     }
 879
 880     if (!mpegts_packetizer_map (packetizer, packet_size))
 881       return PACKET_NEED_MORE;
 882
 883     packet_data = &packetizer->map_data[packetizer->map_offset + sync_offset];
 884
 885     /* Check sync byte */
 886     if (G_UNLIKELY (*packet_data != PACKET_SYNC_BYTE)) {
 887       GST_DEBUG ("lost sync");
 888       packetizer->need_sync = TRUE;
 889     } else {
 890       /* ALL mpeg-ts variants contain 188 bytes of data. Those with bigger
 891        * packet sizes contain either extra data (timesync, FEC, ..) either
 892        * before or after the data */
 893       packet->data_start = packet_data;
 894       packet->data_end = packet->data_start + 188;
 895       packet->offset = packetizer->offset;
 896       GST_LOG ("offset %" G_GUINT64_FORMAT, packet->offset);
 897       packetizer->offset += packet_size;
 898       GST_MEMDUMP ("data_start", packet->data_start, 16);
 899
 900       return mpegts_packetizer_parse_packet (packetizer, packet);
 901     }
 902   }
 903 }
 904
 905 MpegTSPacketizerPacketReturn
 906 mpegts_packetizer_process_next_packet (MpegTSPacketizer2 * packetizer)
 907 {
 908   MpegTSPacketizerPacket packet;
 909   MpegTSPacketizerPacketReturn ret;
 910
 911   ret = mpegts_packetizer_next_packet (packetizer, &packet);
 912   if (ret != PACKET_NEED_MORE)
 913     mpegts_packetizer_clear_packet (packetizer, &packet);
 914
 915   return ret;
 916 }
 917
 918 void
 919 mpegts_packetizer_clear_packet (MpegTSPacketizer2 * packetizer,
 920     MpegTSPacketizerPacket * packet)
 921 {
 922   guint8 packet_size = packetizer->packet_size;
 923
 924   if (packetizer->map_data) {
 925     packetizer->map_offset += packet_size;
 926     if (packetizer->map_size - packetizer->map_offset < packet_size)
 927       mpegts_packetizer_flush_bytes (packetizer, packetizer->map_offset);
 928   }
 929 }
 930
 931 gboolean
 932 mpegts_packetizer_has_packets (MpegTSPacketizer2 * packetizer)
 933 {
 934   if (G_UNLIKELY (!packetizer->packet_size)) {
 935     if (!mpegts_try_discover_packet_size (packetizer))
 936       return FALSE;
 937   }
 938   return gst_adapter_available (packetizer->adapter) >= packetizer->packet_size;
 939 }
 940
 941 /*
 942  * Ideally it should just return a section if:
 943  * * The section is complete
 944  * * The section is valid (sanity checks for length for example)
 945  * * The section applies now (current_next_indicator)
 946  * * The section is an update or was never seen
 947  *
 948  * The section should be a new GstMpegtsSection:
 949  * * properly initialized
 950  * * With pid, table_id AND section_type set (move logic from mpegtsbase)
 951  * * With data copied into it (yes, minor overhead)
 952  *
 953  * In all other cases it should just return NULL
 954  *
 955  * If more than one section is available, the 'remaining' field will
 956  * be set to the beginning of a valid GList containing other sections.
 957  * */
 958 GstMpegtsSection *
 959 mpegts_packetizer_push_section (MpegTSPacketizer2 * packetizer,
 960     MpegTSPacketizerPacket * packet, GList ** remaining)
 961 {
 962   GstMpegtsSection *section;
 963   GstMpegtsSection *res = NULL;
 964   MpegTSPacketizerStream *stream;
 965   gboolean long_packet;
 966   guint8 pointer = 0, table_id;
 967   guint16 subtable_extension;
 968   gsize to_read;
 969   guint section_length;
 970   /* data points to the current read location
 971    * data_start points to the beginning of the data to accumulate */
 972   guint8 *data, *data_start;
 973   guint8 packet_cc;
 974   GList *others = NULL;
 975   guint8 version_number, section_number, last_section_number;
 976
 977   data = packet->data;
 978   packet_cc = FLAGS_CONTINUITY_COUNTER (packet->scram_afc_cc);
 979
 980   /* Get our filter */
 981   stream = packetizer->streams[packet->pid];
 982   if (G_UNLIKELY (stream == NULL)) {
 983     if (!packet->payload_unit_start_indicator) {
 984       /* Early exit (we need to start with a section start) */
 985       GST_DEBUG ("PID 0x%04x  waiting for section start", packet->pid);
 986       goto out;
 987     }
 988     stream = mpegts_packetizer_stream_new (packet->pid);
 989     packetizer->streams[packet->pid] = stream;
 990   }
 991
 992   GST_MEMDUMP ("Full packet data", packet->data,
 993       packet->data_end - packet->data);
 994
 995   /* This function is split into several parts:
 996    *
 997    * Pre checks (packet-wide). Determines where we go next
 998    * accumulate_data: store data and check if section is complete
 999    * section_start: handle beginning of a section, if needed loop back to
1000    *                accumulate_data
1001    *
1002    * The trigger that makes the loop stop and return is if:
1003    * 1) We do not have enough data for the current packet
1004    * 2) There is remaining data after a packet which is only made
1005    *    of stuffing bytes (0xff).
1006    *
1007    * Pre-loop checks, related to the whole incoming packet:
1008    *
1009    * If there is a CC-discont:
1010    *  If it is a PUSI, skip the pointer and handle section_start
1011    *  If not a PUSI, reset and return nothing
1012    * If there is not a CC-discont:
1013    *  If it is a PUSI
1014    *    If pointer, accumulate that data and check for complete section
1015    *    (loop)
1016    *  If it is not a PUSI
1017    *    Accumulate the expected data and check for complete section
1018    *    (loop)
1019    *
1020    **/
1021
1022   if (packet->payload_unit_start_indicator) {
1023     pointer = *data++;
1024     /* If the pointer is zero, we're guaranteed to be able to handle it */
1025     if (pointer == 0) {
1026       GST_LOG
1027           ("PID 0x%04x PUSI and pointer == 0, skipping straight to section_start parsing",
1028           packet->pid);
1029       mpegts_packetizer_clear_section (stream);
1030       goto section_start;
1031     }
1032   }
1033
1034   if (stream->continuity_counter == CONTINUITY_UNSET ||
1035       (stream->continuity_counter + 1) % 16 != packet_cc) {
1036     if (stream->continuity_counter != CONTINUITY_UNSET)
1037       GST_WARNING ("PID 0x%04x section discontinuity (%d vs %d)", packet->pid,
1038           stream->continuity_counter, packet_cc);
1039     mpegts_packetizer_clear_section (stream);
1040     /* If not a PUSI, not much we can do */
1041     if (!packet->payload_unit_start_indicator) {
1042       GST_LOG ("PID 0x%04x continuity discont/unset and not PUSI, bailing out",
1043           packet->pid);
1044       goto out;
1045     }
1046     /* If PUSI, skip pointer data and carry on to section start */
1047     data += pointer;
1048     pointer = 0;
1049     GST_LOG ("discont, but PUSI, skipped %d bytes and doing section start",
1050         pointer);
1051     goto section_start;
1052   }
1053
1054   GST_LOG ("Accumulating data from beginning of packet");
1055
1056   data_start = data;
1057
1058 accumulate_data:
1059   /* If not the beginning of a new section, accumulate what we have */
1060   stream->continuity_counter = packet_cc;
1061   to_read = MIN (stream->section_length - stream->section_offset,
1062       packet->data_end - data_start);
1063   memcpy (stream->section_data + stream->section_offset, data_start, to_read);
1064   stream->section_offset += to_read;
1065   /* Point data to after the data we accumulated */
1066   data = data_start + to_read;
1067   GST_DEBUG ("Appending data (need %d, have %d)", stream->section_length,
1068       stream->section_offset);
1069
1070   /* Check if we have enough */
1071   if (stream->section_offset < stream->section_length) {
1072     GST_DEBUG ("PID 0x%04x, section not complete (Got %d, need %d)",
1073         stream->pid, stream->section_offset, stream->section_length);
1074     goto out;
1075   }
1076
1077   /* Small sanity check. We should have collected *exactly* the right amount */
1078   if (G_UNLIKELY (stream->section_offset != stream->section_length))
1079     GST_WARNING ("PID 0x%04x Accumulated too much data (%d vs %d) !",
1080         stream->pid, stream->section_offset, stream->section_length);
1081   GST_DEBUG ("PID 0x%04x Section complete", stream->pid);
1082
1083   if ((section = mpegts_packetizer_parse_section_header (packetizer, stream))) {
1084     if (res)
1085       others = g_list_append (others, section);
1086     else
1087       res = section;
1088   }
1089
1090 section_start:
1091   subtable_extension = 0;
1092   version_number = 0;
1093   last_section_number = 0;
1094   section_number = 0;
1095   table_id = 0;
1096
1097   /* FIXME : We need at least 3 bytes (or 8 for long packets) with current algorithm :(
1098    * We might end up losing sections that start across two packets (srsl...) */
1099   if (data > packet->data_end - 3 || *data == 0xff) {
1100     /* flush stuffing bytes and leave */
1101     mpegts_packetizer_clear_section (stream);
1102     goto out;
1103   }
1104
1105   /* We have more data to process ... */
1106   GST_DEBUG ("PID 0x%04x, More section present in packet (remaining bytes:%"
1107       G_GSIZE_FORMAT ")", stream->pid, (gsize) (packet->data_end - data));
1108   GST_MEMDUMP ("section_start", data, packet->data_end - data);
1109   data_start = data;
1110   /* Beginning of a new section */
1111   /*
1112    * section_syntax_indicator means that the header is of the following format:
1113    * * table_id (8bit)
1114    * * section_syntax_indicator (1bit) == 0
1115    * * reserved/private fields (3bit)
1116    * * section_length (12bit)
1117    * * data (of size section_length)
1118    * * NO CRC !
1119    */
1120   long_packet = data[1] & 0x80;
1121
1122   /* Fast path for short packets */
1123   if (!long_packet) {
1124     /* We can create the section now (function will check for size) */
1125     GST_DEBUG ("Short packet");
1126     section_length = (GST_READ_UINT16_BE (data + 1) & 0xfff) + 3;
1127     /* Only do fast-path if we have enough byte */
1128     if (data + section_length <= packet->data_end) {
1129       if ((section =
1130               gst_mpegts_section_new (packet->pid, g_memdup2 (data,
1131                       section_length), section_length))) {
1132         GST_DEBUG ("PID 0x%04x Short section complete !", packet->pid);
1133         section->offset = packet->offset;
1134         if (res)
1135           others = g_list_append (others, section);
1136         else
1137           res = section;
1138       }
1139       /* Advance reader and potentially read another section */
1140       data += section_length;
1141       if (data < packet->data_end && *data != 0xff)
1142         goto section_start;
1143       /* If not, exit */
1144       goto out;
1145     }
1146     /* We don't have enough bytes to do short section shortcut */
1147   }
1148
1149   /* Beginning of a new section, do as much pre-parsing as possible */
1150   /* table_id                        : 8  bit */
1151   table_id = *data++;
1152
1153   /* section_syntax_indicator        : 1  bit
1154    * other_fields (reserved)         : 3  bit
1155    * section_length                  : 12 bit */
1156   section_length = (GST_READ_UINT16_BE (data) & 0x0FFF) + 3;
1157   data += 2;
1158
1159   if (long_packet) {
1160     /* Do we have enough data for a long packet? */
1161     if (data > packet->data_end - 5)
1162       goto out;
1163
1164     /* subtable_extension (always present, we are in a long section) */
1165     /* subtable extension              : 16 bit */
1166     subtable_extension = GST_READ_UINT16_BE (data);
1167     data += 2;
1168
1169     /* reserved                      : 2  bit
1170      * version_number                : 5  bit
1171      * current_next_indicator        : 1  bit */
1172     /* Bail out now if current_next_indicator == 0 */
1173     if (G_UNLIKELY (!(*data & 0x01))) {
1174       GST_DEBUG
1175           ("PID 0x%04x table_id 0x%02x section does not apply (current_next_indicator == 0)",
1176           packet->pid, table_id);
1177       goto out;
1178     }
1179
1180     version_number = *data++ >> 1 & 0x1f;
1181     /* section_number                : 8  bit */
1182     section_number = *data++;
1183     /* last_section_number                : 8  bit */
1184     last_section_number = *data++;
1185   } else {
1186     subtable_extension = 0;
1187     version_number = 0;
1188     section_number = 0;
1189     last_section_number = 0;
1190   }
1191   GST_DEBUG
1192       ("PID 0x%04x length:%d table_id:0x%02x subtable_extension:0x%04x version_number:%d section_number:%d(last:%d)",
1193       packet->pid, section_length, table_id, subtable_extension, version_number,
1194       section_number, last_section_number);
1195
1196   to_read = MIN (section_length, packet->data_end - data_start);
1197
1198   /* Check as early as possible whether we already saw this section
1199    * i.e. that we saw a subtable with:
1200    * * same subtable_extension (might be zero)
1201    * * same version_number
1202    * * same last_section_number
1203    * * same section_number was seen
1204    */
1205   if (seen_section_before (stream, table_id, subtable_extension,
1206           version_number, section_number, last_section_number, data_start,
1207           to_read)) {
1208     GST_DEBUG
1209         ("PID 0x%04x Already processed table_id:0x%02x subtable_extension:0x%04x, version_number:%d, section_number:%d",
1210         packet->pid, table_id, subtable_extension, version_number,
1211         section_number);
1212     /* skip data and see if we have more sections after */
1213     data = data_start + to_read;
1214     if (data == packet->data_end || *data == 0xff)
1215       goto out;
1216     goto section_start;
1217   }
1218   if (G_UNLIKELY (section_number > last_section_number)) {
1219     GST_WARNING
1220         ("PID 0x%04x corrupted packet (section_number:%d > last_section_number:%d)",
1221         packet->pid, section_number, last_section_number);
1222     goto out;
1223   }
1224
1225
1226   /* Copy over already parsed values */
1227   stream->table_id = table_id;
1228   stream->section_length = section_length;
1229   stream->version_number = version_number;
1230   stream->subtable_extension = subtable_extension;
1231   stream->section_number = section_number;
1232   stream->last_section_number = last_section_number;
1233   stream->offset = packet->offset;
1234
1235   /* Create enough room to store chunks of sections */
1236   stream->section_data = g_malloc (stream->section_length);
1237   stream->section_offset = 0;
1238
1239   /* Finally, accumulate and check if we parsed enough */
1240   goto accumulate_data;
1241
1242 out:
1243   packet->data = data;
1244   *remaining = others;
1245
1246   GST_DEBUG ("result: %p", res);
1247
1248   return res;
1249 }
1250
1251 static void
1252 _init_local (void)
1253 {
1254   GST_DEBUG_CATEGORY_INIT (mpegts_packetizer_debug, "mpegtspacketizer", 0,
1255       "MPEG transport stream parser");
1256 }
1257
1258
1259 static void
1260 mpegts_packetizer_resync (MpegTSPCR * pcr, GstClockTime time,
1261     GstClockTime gstpcrtime, gboolean reset_skew)
1262 {
1263   pcr->base_time = time;
1264   pcr->base_pcrtime = gstpcrtime;
1265   pcr->prev_out_time = GST_CLOCK_TIME_NONE;
1266   pcr->prev_send_diff = GST_CLOCK_TIME_NONE;
1267   if (reset_skew) {
1268     pcr->window_filling = TRUE;
1269     pcr->window_pos = 0;
1270     pcr->window_min = 0;
1271     pcr->window_size = 0;
1272     pcr->skew = 0;
1273   }
1274 }
1275
1276
1277 /* Code mostly copied from -good/gst/rtpmanager/rtpjitterbuffer.c */
1278
1279 /* For the clock skew we use a windowed low point averaging algorithm as can be
1280  * found in Fober, Orlarey and Letz, 2005, "Real Time Clock Skew Estimation
1281  * over Network Delays":
1282  * http://www.grame.fr/Ressources/pub/TR-050601.pdf
1283  * http://citeseerx.ist.psu.edu/viewdoc/summary?doi=10.1.1.102.1546
1284  *
1285  * The idea is that the jitter is composed of:
1286  *
1287  *  J = N + n
1288  *
1289  *   D   : a constant network delay.
1290  *   n   : random added noise. The noise is concentrated around 0
1291  *
1292  * In the receiver we can track the elapsed time at the sender with:
1293  *
1294  *  send_diff(i) = (Tsi - Ts0);
1295  *
1296  *   Tsi : The time at the sender at packet i
1297  *   Ts0 : The time at the sender at the first packet
1298  *
1299  * This is the difference between the RTP timestamp in the first received packet
1300  * and the current packet.
1301  *
1302  * At the receiver we have to deal with the jitter introduced by the network.
1303  *
1304  *  recv_diff(i) = (Tri - Tr0)
1305  *
1306  *   Tri : The time at the receiver at packet i
1307  *   Tr0 : The time at the receiver at the first packet
1308  *
1309  * Both of these values contain a jitter Ji, a jitter for packet i, so we can
1310  * write:
1311  *
1312  *  recv_diff(i) = (Cri + D + ni) - (Cr0 + D + n0))
1313  *
1314  *    Cri    : The time of the clock at the receiver for packet i
1315  *    D + ni : The jitter when receiving packet i
1316  *
1317  * We see that the network delay is irrelevant here as we can eliminate D:
1318  *
1319  *  recv_diff(i) = (Cri + ni) - (Cr0 + n0))
1320  *
1321  * The drift is now expressed as:
1322  *
1323  *  Drift(i) = recv_diff(i) - send_diff(i);
1324  *
1325  * We now keep the W latest values of Drift and find the minimum (this is the
1326  * one with the lowest network jitter and thus the one which is least affected
1327  * by it). We average this lowest value to smooth out the resulting network skew.
1328  *
1329  * Both the window and the weighting used for averaging influence the accuracy
1330  * of the drift estimation. Finding the correct parameters turns out to be a
1331  * compromise between accuracy and inertia.
1332  *
1333  * We use a 2 second window or up to 512 data points, which is statistically big
1334  * enough to catch spikes (FIXME, detect spikes).
1335  * We also use a rather large weighting factor (125) to smoothly adapt. During
1336  * startup, when filling the window, we use a parabolic weighting factor, the
1337  * more the window is filled, the faster we move to the detected possible skew.
1338  *
1339  * Returns: @time adjusted with the clock skew.
1340  */
1341 static GstClockTime
1342 calculate_skew (MpegTSPacketizer2 * packetizer,
1343     MpegTSPCR * pcr, guint64 pcrtime, GstClockTime time)
1344 {
1345   guint64 send_diff, recv_diff;
1346   gint64 delta;
1347   gint64 old;
1348   gint pos, i;
1349   GstClockTime gstpcrtime, out_time;
1350 #ifndef GST_DISABLE_GST_DEBUG
1351   guint64 slope;
1352 #endif
1353
1354   gstpcrtime = PCRTIME_TO_GSTTIME (pcrtime) + pcr->pcroffset;
1355
1356   /* first time, lock on to time and gstpcrtime */
1357   if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (pcr->base_time))) {
1358     pcr->base_time = time;
1359     pcr->prev_out_time = GST_CLOCK_TIME_NONE;
1360     GST_DEBUG ("Taking new base time %" GST_TIME_FORMAT, GST_TIME_ARGS (time));
1361   }
1362
1363   if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (pcr->base_pcrtime))) {
1364     pcr->base_pcrtime = gstpcrtime;
1365     pcr->prev_send_diff = -1;
1366     GST_DEBUG ("Taking new base pcrtime %" GST_TIME_FORMAT,
1367         GST_TIME_ARGS (gstpcrtime));
1368   }
1369
1370   /* Handle PCR wraparound and resets */
1371   if (GST_CLOCK_TIME_IS_VALID (pcr->last_pcrtime) &&
1372       gstpcrtime < pcr->last_pcrtime) {
1373     if (pcr->last_pcrtime - gstpcrtime > PCR_GST_MAX_VALUE / 2) {
1374       /* PCR wraparound */
1375       GST_DEBUG ("PCR wrap");
1376       pcr->pcroffset += PCR_GST_MAX_VALUE;
1377       gstpcrtime = PCRTIME_TO_GSTTIME (pcrtime) + pcr->pcroffset;
1378       send_diff = gstpcrtime - pcr->base_pcrtime;
1379     } else if (GST_CLOCK_TIME_IS_VALID (time)
1380         && pcr->last_pcrtime - gstpcrtime > 15 * GST_SECOND) {
1381       /* Time jumped backward by > 15 seconds, and we have a timestamp
1382        * to use to close the discont. Assume a reset */
1383       GST_DEBUG ("PCR reset");
1384       /* Calculate PCR we would have expected for the given input time,
1385        * essentially applying the reverse correction process
1386        *
1387        * We want to find the PCR offset to apply
1388        *   pcroffset = (corrected) gstpcrtime - (received) gstpcrtime
1389        *
1390        * send_diff = (corrected) gstpcrtime - pcr->base_pcrtime
1391        * recv_diff = time - pcr->base_time
1392        * out_time = pcr->base_time + send_diff
1393        *
1394        * We are assuming that send_diff == recv_diff
1395        *   (corrected) gstpcrtime - pcr->base_pcrtime = time - pcr->base_time
1396        * Giving us:
1397        *   (corrected) gstpcrtime = time - pcr->base_time + pcr->base_pcrtime
1398        *
1399        * And therefore:
1400        *   pcroffset = time - pcr->base_time + pcr->base_pcrtime - (received) gstpcrtime
1401        **/
1402       pcr->pcroffset += time - pcr->base_time + pcr->base_pcrtime - gstpcrtime;
1403       gstpcrtime = PCRTIME_TO_GSTTIME (pcrtime) + pcr->pcroffset;
1404       send_diff = gstpcrtime - pcr->base_pcrtime;
1405       GST_DEBUG ("Introduced offset is now %" GST_TIME_FORMAT
1406           " corrected pcr time %" GST_TIME_FORMAT,
1407           GST_TIME_ARGS (pcr->pcroffset), GST_TIME_ARGS (gstpcrtime));
1408     } else {
1409       /* Small jumps backward, assume some arrival jitter and skip it */
1410       send_diff = 0;
1411
1412       /* The following code are the different ways we deal with small-ish
1413        * jitter, ranging in severity from "can be ignored" to "this needs a full
1414        * resync" */
1415
1416       if (time == pcr->base_time) {
1417         /* If this comes from a non-fully-timestamped source (i.e. adaptive
1418          * streams), then cope with the fact that some producers generate utter
1419          * PCR garbage on fragment ends.
1420          *
1421          * We detect this comes from a non-fully-timestamped source by the fact
1422          * that the buffer time never changes */
1423         GST_DEBUG ("Ignoring PCR resets on non-fully timestamped stream");
1424       } else if (pcr->last_pcrtime - gstpcrtime < GST_SECOND) {
1425         GST_WARNING
1426             ("(small) backward timestamps at server or no buffer timestamps. Ignoring.");
1427         /* This will trigger the no_skew logic before but leave other state
1428          * intact */
1429         time = GST_CLOCK_TIME_NONE;
1430       } else {
1431         /* A bigger backward step than packet out-of-order can account for. Reset base PCR time
1432          * to be resynched the next time we see a PCR */
1433         GST_WARNING
1434             ("backward timestamps at server or no buffer timestamps. Resync base PCR");
1435         pcr->base_pcrtime = GST_CLOCK_TIME_NONE;
1436       }
1437     }
1438   } else
1439     send_diff = gstpcrtime - pcr->base_pcrtime;
1440
1441   GST_DEBUG ("gstpcr %" GST_TIME_FORMAT ", buftime %" GST_TIME_FORMAT
1442       ", base %" GST_TIME_FORMAT ", send_diff %" GST_TIME_FORMAT,
1443       GST_TIME_ARGS (gstpcrtime), GST_TIME_ARGS (time),
1444       GST_TIME_ARGS (pcr->base_pcrtime), GST_TIME_ARGS (send_diff));
1445
1446   /* keep track of the last extended pcrtime */
1447   pcr->last_pcrtime = gstpcrtime;
1448
1449   /* we don't have an arrival timestamp so we can't do skew detection. we
1450    * should still apply a timestamp based on RTP timestamp and base_time */
1451   if (!GST_CLOCK_TIME_IS_VALID (time)
1452       || !GST_CLOCK_TIME_IS_VALID (pcr->base_time))
1453     goto no_skew;
1454
1455   /* elapsed time at receiver, includes the jitter */
1456   recv_diff = time - pcr->base_time;
1457
1458   /* Ignore packets received at 100% the same time (i.e. from the same input buffer) */
1459   if (G_UNLIKELY (time == pcr->prev_in_time
1460           && GST_CLOCK_TIME_IS_VALID (pcr->prev_in_time)))
1461     goto no_skew;
1462
1463   /* measure the diff */
1464   delta = ((gint64) recv_diff) - ((gint64) send_diff);
1465
1466 #ifndef GST_DISABLE_GST_DEBUG
1467   /* measure the slope, this gives a rought estimate between the sender speed
1468    * and the receiver speed. This should be approximately 8, higher values
1469    * indicate a burst (especially when the connection starts) */
1470   slope = recv_diff > 0 ? (send_diff * 8) / recv_diff : 8;
1471 #endif
1472
1473   GST_DEBUG ("time %" GST_TIME_FORMAT ", base %" GST_TIME_FORMAT
1474       ", recv_diff %" GST_TIME_FORMAT ", slope %" G_GUINT64_FORMAT,
1475       GST_TIME_ARGS (time), GST_TIME_ARGS (pcr->base_time),
1476       GST_TIME_ARGS (recv_diff), slope);
1477
1478   /* if the difference between the sender timeline and the receiver timeline
1479    * changed too quickly we have to resync because the server likely restarted
1480    * its timestamps. */
1481   if (ABS (delta - pcr->skew) > packetizer->pcr_discont_threshold) {
1482     GST_WARNING ("delta - skew: %" GST_STIME_FORMAT " too big, reset skew",
1483         GST_STIME_ARGS (delta - pcr->skew));
1484     mpegts_packetizer_resync (pcr, time, gstpcrtime, TRUE);
1485     send_diff = 0;
1486     delta = 0;
1487   }
1488
1489   pos = pcr->window_pos;
1490
1491   if (G_UNLIKELY (pcr->window_filling)) {
1492     /* we are filling the window */
1493     GST_DEBUG ("filling %d, delta %" G_GINT64_FORMAT, pos, delta);
1494     pcr->window[pos++] = delta;
1495     /* calc the min delta we observed */
1496     if (G_UNLIKELY (pos == 1 || delta < pcr->window_min))
1497       pcr->window_min = delta;
1498
1499     if (G_UNLIKELY (send_diff >= MAX_TIME || pos >= MAX_WINDOW)) {
1500       pcr->window_size = pos;
1501
1502       /* window filled */
1503       GST_DEBUG ("min %" G_GINT64_FORMAT, pcr->window_min);
1504
1505       /* the skew is now the min */
1506       pcr->skew = pcr->window_min;
1507       pcr->window_filling = FALSE;
1508     } else {
1509       gint perc_time, perc_window, perc;
1510
1511       /* figure out how much we filled the window, this depends on the amount of
1512        * time we have or the max number of points we keep. */
1513       perc_time = send_diff * 100 / MAX_TIME;
1514       perc_window = pos * 100 / MAX_WINDOW;
1515       perc = MAX (perc_time, perc_window);
1516
1517       /* make a parabolic function, the closer we get to the MAX, the more value
1518        * we give to the scaling factor of the new value */
1519       perc = perc * perc;
1520
1521       /* quickly go to the min value when we are filling up, slowly when we are
1522        * just starting because we're not sure it's a good value yet. */
1523       pcr->skew =
1524           (perc * pcr->window_min + ((10000 - perc) * pcr->skew)) / 10000;
1525       pcr->window_size = pos + 1;
1526     }
1527   } else {
1528     /* pick old value and store new value. We keep the previous value in order
1529      * to quickly check if the min of the window changed */
1530     old = pcr->window[pos];
1531     pcr->window[pos++] = delta;
1532
1533     if (G_UNLIKELY (delta <= pcr->window_min)) {
1534       /* if the new value we inserted is smaller or equal to the current min,
1535        * it becomes the new min */
1536       pcr->window_min = delta;
1537     } else if (G_UNLIKELY (old == pcr->window_min)) {
1538       gint64 min = G_MAXINT64;
1539
1540       /* if we removed the old min, we have to find a new min */
1541       for (i = 0; i < pcr->window_size; i++) {
1542         /* we found another value equal to the old min, we can stop searching now */
1543         if (pcr->window[i] == old) {
1544           min = old;
1545           break;
1546         }
1547         if (pcr->window[i] < min)
1548           min = pcr->window[i];
1549       }
1550       pcr->window_min = min;
1551     }
1552     /* average the min values */
1553     pcr->skew = (pcr->window_min + (124 * pcr->skew)) / 125;
1554     GST_DEBUG ("delta %" G_GINT64_FORMAT ", new min: %" G_GINT64_FORMAT,
1555         delta, pcr->window_min);
1556   }
1557   /* wrap around in the window */
1558   if (G_UNLIKELY (pos >= pcr->window_size))
1559     pos = 0;
1560
1561   pcr->window_pos = pos;
1562
1563 no_skew:
1564   /* the output time is defined as the base timestamp plus the PCR time
1565    * adjusted for the clock skew .*/
1566   if (pcr->base_time != -1) {
1567     out_time = pcr->base_time + send_diff;
1568     /* skew can be negative and we don't want to make invalid timestamps */
1569     if (pcr->skew < 0 && out_time < -pcr->skew) {
1570       out_time = 0;
1571     } else {
1572       out_time += pcr->skew;
1573     }
1574     /* check if timestamps are not going backwards, we can only check this if we
1575      * have a previous out time and a previous send_diff */
1576     if (G_LIKELY (pcr->prev_out_time != -1 && pcr->prev_send_diff != -1)) {
1577       /* now check for backwards timestamps */
1578       if (G_UNLIKELY (
1579               /* if the server timestamps went up and the out_time backwards */
1580               (send_diff > pcr->prev_send_diff
1581                   && out_time < pcr->prev_out_time) ||
1582               /* if the server timestamps went backwards and the out_time forwards */
1583               (send_diff < pcr->prev_send_diff
1584                   && out_time > pcr->prev_out_time) ||
1585               /* if the server timestamps did not change */
1586               send_diff == pcr->prev_send_diff)) {
1587         GST_DEBUG ("backwards timestamps, using previous time");
1588         out_time = GSTTIME_TO_MPEGTIME (out_time);
1589       }
1590     }
1591   } else {
1592     /* We simply use the pcrtime without applying any skew compensation */
1593     out_time = time;
1594   }
1595
1596   pcr->prev_out_time = out_time;
1597   pcr->prev_in_time = time;
1598   pcr->prev_send_diff = send_diff;
1599
1600   GST_DEBUG ("skew %" G_GINT64_FORMAT ", out %" GST_TIME_FORMAT,
1601       pcr->skew, GST_TIME_ARGS (out_time));
1602
1603   return out_time;
1604 }
1605
1606 static void
1607 _reevaluate_group_pcr_offset (MpegTSPCR * pcrtable, PCROffsetGroup * group)
1608 {
1609   PCROffsetGroup *prev = NULL;
1610 #ifndef GST_DISABLE_GST_DEBUG
1611   PCROffsetGroup *first = pcrtable->groups->data;
1612 #endif
1613   PCROffsetCurrent *current = pcrtable->current;
1614   GList *tmp;
1615
1616   /* Go over all ESTIMATED groups until the target group */
1617   for (tmp = pcrtable->groups; tmp; tmp = tmp->next) {
1618     PCROffsetGroup *cur = (PCROffsetGroup *) tmp->data;
1619
1620     /* Skip groups that don't need re-evaluation */
1621     if (!(cur->flags & PCR_GROUP_FLAG_ESTIMATED)) {
1622       GST_DEBUG ("Skipping group %p pcr_offset (currently %" GST_TIME_FORMAT
1623           ")", cur, GST_TIME_ARGS (PCRTIME_TO_GSTTIME (cur->pcr_offset)));
1624       prev = cur;
1625       continue;
1626     }
1627
1628     /* This should not happen ! The first group is *always* correct (zero) */
1629     if (G_UNLIKELY (prev == NULL)) {
1630       GST_ERROR ("First PCR Group was not estimated (bug). Setting to zero");
1631       cur->pcr_offset = 0;
1632       cur->flags &= ~PCR_GROUP_FLAG_ESTIMATED;
1633       return;
1634     }
1635
1636     /* Finally do the estimation of this group's PCR offset based on the
1637      * previous group information */
1638
1639     GST_DEBUG ("Re-evaluating group %p pcr_offset (currently %" GST_TIME_FORMAT
1640         ")", group, GST_TIME_ARGS (PCRTIME_TO_GSTTIME (cur->pcr_offset)));
1641
1642     GST_DEBUG ("cur->first_pcr:%" GST_TIME_FORMAT " prev->first_pcr:%"
1643         GST_TIME_FORMAT, GST_TIME_ARGS (PCRTIME_TO_GSTTIME (cur->first_pcr)),
1644         GST_TIME_ARGS (PCRTIME_TO_GSTTIME (prev->first_pcr)));
1645
1646     if (G_UNLIKELY (cur->first_pcr < prev->first_pcr)) {
1647       guint64 prevbr, lastbr;
1648       guint64 prevpcr;
1649       guint64 prevoffset, lastoffset;
1650
1651       /* Take the previous group pcr_offset and figure out how much to add
1652        * to it for the current group */
1653
1654       /* Right now we do a dumb bitrate estimation
1655        * estimate bitrate (prev - first) : bitrate from the start
1656        * estimate bitrate (prev) : bitrate of previous group
1657        * estimate bitrate (last - first) : bitrate from previous group
1658        *
1659        * We will use raw (non-corrected/non-absolute) PCR values in a first time
1660        * to detect wraparound/resets/gaps...
1661        *
1662        * We will use the corrected/absolute PCR values to calculate
1663        * bitrate and estimate the target group pcr_offset.
1664        * */
1665
1666       /* If the current window estimator is over the previous group, used those
1667        * values as the latest (since they are more recent) */
1668       if (current->group == prev && current->pending[current->last].offset) {
1669         prevoffset =
1670             current->pending[current->last].offset + prev->first_offset;
1671         prevpcr = current->pending[current->last].pcr + prev->first_pcr;
1672         /* prevbr: bitrate(prev) */
1673         prevbr =
1674             gst_util_uint64_scale (PCR_SECOND,
1675             current->pending[current->last].offset,
1676             current->pending[current->last].pcr);
1677         GST_DEBUG ("Previous group bitrate (%" G_GUINT64_FORMAT " / %"
1678             GST_TIME_FORMAT ") : %" G_GUINT64_FORMAT,
1679             current->pending[current->last].offset,
1680             GST_TIME_ARGS (PCRTIME_TO_GSTTIME (current->pending[current->last].
1681                     pcr)), prevbr);
1682       } else if (prev->values[prev->last_value].offset) {
1683         prevoffset = prev->values[prev->last_value].offset + prev->first_offset;
1684         prevpcr = prev->values[prev->last_value].pcr + prev->first_pcr;
1685         /* prevbr: bitrate(prev) (FIXME : Cache) */
1686         prevbr =
1687             gst_util_uint64_scale (PCR_SECOND,
1688             prev->values[prev->last_value].offset,
1689             prev->values[prev->last_value].pcr);
1690         GST_DEBUG ("Previous group bitrate (%" G_GUINT64_FORMAT " / %"
1691             GST_TIME_FORMAT ") : %" G_GUINT64_FORMAT,
1692             prev->values[prev->last_value].offset,
1693             GST_TIME_ARGS (PCRTIME_TO_GSTTIME (prev->values[prev->last_value].
1694                     pcr)), prevbr);
1695       } else {
1696         GST_DEBUG ("Using overall bitrate");
1697         prevoffset = prev->values[prev->last_value].offset + prev->first_offset;
1698         prevpcr = prev->values[prev->last_value].pcr + prev->first_pcr;
1699         prevbr = gst_util_uint64_scale (PCR_SECOND,
1700             prev->first_offset, prev->pcr_offset);
1701       }
1702       lastoffset = cur->values[cur->last_value].offset + cur->first_offset;
1703
1704       GST_DEBUG ("Offset first:%" G_GUINT64_FORMAT " prev:%" G_GUINT64_FORMAT
1705           " cur:%" G_GUINT64_FORMAT, first->first_offset, prevoffset,
1706           lastoffset);
1707       GST_DEBUG ("PCR first:%" GST_TIME_FORMAT " prev:%" GST_TIME_FORMAT
1708           " cur:%" GST_TIME_FORMAT,
1709           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (first->first_pcr)),
1710           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (prevpcr)),
1711           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (cur->values[cur->last_value].pcr +
1712                   cur->first_pcr)));
1713
1714       if (prevpcr - cur->first_pcr > (PCR_MAX_VALUE * 9 / 10)) {
1715         gfloat diffprev;
1716         guint64 guess_offset;
1717
1718         /* Let's assume there is a PCR wraparound between the previous and current
1719          * group.
1720          * [ prev ]... PCR_MAX | 0 ...[ current ]
1721          * The estimated pcr_offset would therefore be:
1722          * current.first + (PCR_MAX_VALUE - prev.first)
1723          *
1724          * 1) Check if bitrate(prev) would be consistent with bitrate (cur - prev)
1725          */
1726         guess_offset = PCR_MAX_VALUE - prev->first_pcr + cur->first_pcr;
1727         lastbr = gst_util_uint64_scale (PCR_SECOND, lastoffset - prevoffset,
1728             guess_offset + cur->values[cur->last_value].pcr - (prevpcr -
1729                 prev->first_pcr));
1730         GST_DEBUG ("Wraparound prev-cur (guess_offset:%" GST_TIME_FORMAT
1731             ") bitrate:%" G_GUINT64_FORMAT,
1732             GST_TIME_ARGS (PCRTIME_TO_GSTTIME (guess_offset)), lastbr);
1733         diffprev = (float) 100.0 *(ABSDIFF (prevbr, lastbr)) / (float) prevbr;
1734         GST_DEBUG ("Difference with previous bitrate:%f", diffprev);
1735         if (diffprev < 10.0) {
1736           GST_DEBUG ("Difference < 10.0, Setting pcr_offset to %"
1737               G_GUINT64_FORMAT, guess_offset);
1738           cur->pcr_offset = guess_offset;
1739           if (diffprev < 1.0) {
1740             GST_DEBUG ("Difference < 1.0, Removing ESTIMATED flags");
1741             cur->flags &= ~PCR_GROUP_FLAG_ESTIMATED;
1742           }
1743         }
1744         /* Indicate the the previous group is before a wrapover */
1745         prev->flags |= PCR_GROUP_FLAG_WRAPOVER;
1746       } else {
1747         guint64 resetprev;
1748         /* Let's assume there was a PCR reset between the previous and current
1749          * group
1750          * [ prev ] ... x | x - reset ... [ current ]
1751          *
1752          * The estimated pcr_offset would then be
1753          * = current.first - (x - reset) + (x - prev.first) + 100ms (for safety)
1754          * = current.first + reset - prev.first + 100ms (for safety)
1755          */
1756         /* In order to calculate the reset, we estimate what the PCR would have
1757          * been by using prevbr */
1758         /* FIXME : Which bitrate should we use ???  */
1759         GST_DEBUG ("Using prevbr:%" G_GUINT64_FORMAT " and taking offsetdiff:%"
1760             G_GUINT64_FORMAT, prevbr, cur->first_offset - prev->first_offset);
1761         resetprev =
1762             gst_util_uint64_scale (PCR_SECOND,
1763             cur->first_offset - prev->first_offset, prevbr);
1764         GST_DEBUG ("Estimated full PCR for offset %" G_GUINT64_FORMAT
1765             ", using prevbr:%"
1766             GST_TIME_FORMAT, cur->first_offset,
1767             GST_TIME_ARGS (PCRTIME_TO_GSTTIME (resetprev)));
1768         cur->pcr_offset = prev->pcr_offset + resetprev + 100 * PCR_MSECOND;
1769         GST_DEBUG ("Adjusted group PCR_offset to %" GST_TIME_FORMAT,
1770             GST_TIME_ARGS (PCRTIME_TO_GSTTIME (cur->pcr_offset)));
1771         /* Indicate the the previous group is before a reset */
1772         prev->flags |= PCR_GROUP_FLAG_RESET;
1773       }
1774     } else {
1775       /* FIXME : Detect gaps if bitrate difference is really too big ? */
1776       cur->pcr_offset = prev->pcr_offset + cur->first_pcr - prev->first_pcr;
1777       GST_DEBUG ("Assuming there is no gap, setting pcr_offset to %"
1778           GST_TIME_FORMAT,
1779           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (cur->pcr_offset)));
1780       /* Remove the reset and wrapover flag (if it was previously there) */
1781       prev->flags &= ~PCR_GROUP_FLAG_RESET;
1782       prev->flags &= ~PCR_GROUP_FLAG_WRAPOVER;
1783     }
1784
1785
1786     /* Remember prev for the next group evaluation */
1787     prev = cur;
1788   }
1789 }
1790
1791 static PCROffsetGroup *
1792 _new_group (guint64 pcr, guint64 offset, guint64 pcr_offset, guint flags)
1793 {
1794   PCROffsetGroup *group = g_slice_new0 (PCROffsetGroup);
1795
1796   GST_DEBUG ("Input PCR %" GST_TIME_FORMAT " offset:%" G_GUINT64_FORMAT
1797       " pcr_offset:%" G_GUINT64_FORMAT " flags:%d",
1798       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (pcr)), offset, pcr_offset, flags);
1799
1800   group->flags = flags;
1801   group->values = g_new0 (PCROffset, DEFAULT_ALLOCATED_OFFSET);
1802   /* The first pcr/offset diff is always 0/0 */
1803   group->values[0].pcr = group->values[0].offset = 0;
1804   group->nb_allocated = DEFAULT_ALLOCATED_OFFSET;
1805
1806   /* Store the full values */
1807   group->first_pcr = pcr;
1808   group->first_offset = offset;
1809   group->pcr_offset = pcr_offset;
1810
1811   GST_DEBUG ("Created group starting with pcr:%" GST_TIME_FORMAT " offset:%"
1812       G_GUINT64_FORMAT " pcr_offset:%" GST_TIME_FORMAT,
1813       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->first_pcr)),
1814       group->first_offset,
1815       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->pcr_offset)));
1816
1817   return group;
1818 }
1819
1820 static void
1821 _insert_group_after (MpegTSPCR * pcrtable, PCROffsetGroup * group,
1822     PCROffsetGroup * prev)
1823 {
1824   if (prev == NULL) {
1825     /* First group */
1826     pcrtable->groups = g_list_prepend (pcrtable->groups, group);
1827   } else {
1828     GList *tmp, *toinsert, *prevlist = NULL, *nextlist = NULL;
1829     /* Insert before next and prev */
1830     for (tmp = pcrtable->groups; tmp; tmp = tmp->next) {
1831       if (tmp->data == prev) {
1832         prevlist = tmp;
1833         nextlist = tmp->next;
1834         break;
1835       }
1836     }
1837     if (!prevlist) {
1838       /* The non NULL prev given isn't in the list */
1839       GST_WARNING ("Request to insert before a group which isn't in the list");
1840       pcrtable->groups = g_list_prepend (pcrtable->groups, group);
1841     } else {
1842       toinsert = g_list_append (NULL, group);
1843       toinsert->next = nextlist;
1844       toinsert->prev = prevlist;
1845       prevlist->next = toinsert;
1846       if (nextlist)
1847         nextlist->prev = toinsert;
1848     }
1849   }
1850 }
1851
1852 static void
1853 _use_group (MpegTSPCR * pcrtable, PCROffsetGroup * group)
1854 {
1855   PCROffsetCurrent *current = pcrtable->current;
1856
1857   memset (current, 0, sizeof (PCROffsetCurrent));
1858   current->group = group;
1859   current->pending[0] = group->values[group->last_value];
1860   current->last_value = current->pending[0];
1861   current->write = 1;
1862   current->prev = group->values[group->last_value];
1863   current->first_pcr = group->first_pcr;
1864   current->first_offset = group->first_offset;
1865 }
1866
1867 /* Create a new group with the specified values after prev
1868  * Set current to that new group */
1869 static void
1870 _set_current_group (MpegTSPCR * pcrtable,
1871     PCROffsetGroup * prev, guint64 pcr, guint64 offset, gboolean contiguous)
1872 {
1873   PCROffsetGroup *group;
1874   guint flags = 0;
1875   guint64 pcr_offset = 0;
1876
1877   /* Handle wraparound/gap (only if contiguous with previous group) */
1878   if (contiguous) {
1879     guint64 lastpcr = prev->first_pcr + prev->values[prev->last_value].pcr;
1880
1881     /* Set CLOSED flag on previous group and remember pcr_offset */
1882     prev->flags |= PCR_GROUP_FLAG_CLOSED;
1883     pcr_offset = prev->pcr_offset;
1884
1885     /* Wraparound ? */
1886     if (lastpcr > pcr) {
1887       /* In offset-mode, a PCR wraparound is only actually consistent if
1888        * we have a very high confidence (99% right now, might need to change
1889        * later) */
1890       if (lastpcr - pcr > (PCR_MAX_VALUE * 99 / 100)) {
1891         GST_WARNING ("WRAPAROUND detected. diff %" GST_TIME_FORMAT,
1892             GST_TIME_ARGS (PCRTIME_TO_GSTTIME (lastpcr - pcr)));
1893         /* The previous group closed at PCR_MAX_VALUE */
1894         pcr_offset += PCR_MAX_VALUE - prev->first_pcr + pcr;
1895       } else {
1896         GST_WARNING ("RESET detected. diff %" GST_TIME_FORMAT,
1897             GST_TIME_ARGS (PCRTIME_TO_GSTTIME (lastpcr - pcr)));
1898         /* The previous group closed at the raw last_pcr diff (+100ms for safety) */
1899         pcr_offset += prev->values[prev->last_value].pcr + 100 * PCR_MSECOND;
1900       }
1901     } else if (lastpcr < pcr - 500 * PCR_MSECOND) {
1902       GST_WARNING ("GAP detected. diff %" GST_TIME_FORMAT,
1903           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (pcr - lastpcr)));
1904       /* The previous group closed at the raw last_pcr diff (+500ms for safety) */
1905       pcr_offset += prev->values[prev->last_value].pcr + 500 * PCR_MSECOND;
1906     } else
1907       /* Normal continuation (contiguous in time) */
1908       pcr_offset += pcr - prev->first_pcr;
1909
1910   } else if (prev != NULL)
1911     /* If we are not contiguous and it's not the first group, the pcr_offset
1912      * will be estimated */
1913     flags = PCR_GROUP_FLAG_ESTIMATED;
1914
1915   group = _new_group (pcr, offset, pcr_offset, flags);
1916   _use_group (pcrtable, group);
1917   _insert_group_after (pcrtable, group, prev);
1918   if (!contiguous)
1919     _reevaluate_group_pcr_offset (pcrtable, group);
1920 }
1921
1922 static inline void
1923 _append_group_values (PCROffsetGroup * group, PCROffset pcroffset)
1924 {
1925   /* Only append if new values */
1926   if (group->values[group->last_value].offset == pcroffset.offset &&
1927       group->values[group->last_value].pcr == pcroffset.pcr) {
1928     GST_DEBUG ("Same values, ignoring");
1929   } else {
1930     group->last_value++;
1931     /* Resize values if needed */
1932     if (G_UNLIKELY (group->nb_allocated == group->last_value)) {
1933       group->nb_allocated += DEFAULT_ALLOCATED_OFFSET;
1934       group->values =
1935           g_realloc (group->values, group->nb_allocated * sizeof (PCROffset));
1936     }
1937     group->values[group->last_value] = pcroffset;
1938   }
1939
1940   GST_DEBUG ("First PCR:%" GST_TIME_FORMAT " offset:%" G_GUINT64_FORMAT
1941       " PCR_offset:%" GST_TIME_FORMAT,
1942       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->first_pcr)),
1943       group->first_offset,
1944       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->pcr_offset)));
1945   GST_DEBUG ("Last PCR: +%" GST_TIME_FORMAT " offset: +%" G_GUINT64_FORMAT,
1946       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (pcroffset.pcr)), pcroffset.offset);
1947 }
1948
1949 /* Move last values from current (if any) to the current group
1950  * and reset current.
1951  * Note: This does not set the CLOSED flag (since we have no next
1952  * contiguous group) */
1953 static void
1954 _close_current_group (MpegTSPCR * pcrtable)
1955 {
1956   PCROffsetCurrent *current = pcrtable->current;
1957   PCROffsetGroup *group = current->group;
1958
1959   if (group == NULL)
1960     return;
1961   GST_DEBUG ("Closing group and resetting current");
1962
1963   /* Store last values */
1964   _append_group_values (group, current->pending[current->last]);
1965   memset (current, 0, sizeof (PCROffsetCurrent));
1966   /* And re-evaluate all groups */
1967 }
1968
1969 static void
1970 record_pcr (MpegTSPacketizer2 * packetizer, MpegTSPCR * pcrtable,
1971     guint64 pcr, guint64 offset)
1972 {
1973   PCROffsetCurrent *current = pcrtable->current;
1974   gint64 corpcr, coroffset;
1975
1976   packetizer->nb_seen_offsets += 1;
1977
1978   pcrtable->last_pcrtime = PCRTIME_TO_GSTTIME (pcr);
1979   /* FIXME : Invert logic later (probability is higher that we have a
1980    * current estimator) */
1981
1982   /* Check for current */
1983   if (G_UNLIKELY (current->group == NULL)) {
1984     PCROffsetGroup *prev = NULL;
1985     GList *tmp;
1986     /* No current estimator. This happens for the initial value, or after
1987      * discont and flushes. Figure out where we need to record this position.
1988      *
1989      * Possible choices:
1990      * 1) No groups at all:
1991      *    Create a new group with pcr/offset
1992      *    Initialize current to that group
1993      * 2) Entirely within an existing group
1994      *    bail out (FIXME: Make this detection faster)
1995      * 3) Not in any group
1996      *    Create a new group with pcr/offset at the right position
1997      *    Initialize current to that group
1998      */
1999     GST_DEBUG ("No current window estimator, Checking for group to use");
2000     for (tmp = pcrtable->groups; tmp; tmp = tmp->next) {
2001       PCROffsetGroup *group = (PCROffsetGroup *) tmp->data;
2002
2003       GST_DEBUG ("First PCR:%" GST_TIME_FORMAT " offset:%" G_GUINT64_FORMAT
2004           " PCR_offset:%" GST_TIME_FORMAT,
2005           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->first_pcr)),
2006           group->first_offset,
2007           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->pcr_offset)));
2008       GST_DEBUG ("Last PCR: +%" GST_TIME_FORMAT " offset: +%" G_GUINT64_FORMAT,
2009           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->values[group->last_value].
2010                   pcr)), group->values[group->last_value].offset);
2011       /* Check if before group */
2012       if (offset < group->first_offset) {
2013         GST_DEBUG ("offset is before that group");
2014         break;
2015       }
2016       /* Check if within group */
2017       if (offset <=
2018           (group->values[group->last_value].offset + group->first_offset)) {
2019         GST_DEBUG ("Already observed PCR offset %" G_GUINT64_FORMAT, offset);
2020         return;
2021       }
2022       /* Check if just after group (i.e. continuation of it) */
2023       if (!(group->flags & PCR_GROUP_FLAG_CLOSED) &&
2024           pcr - group->first_pcr - group->values[group->last_value].pcr <=
2025           100 * PCR_MSECOND) {
2026         GST_DEBUG ("Continuation of existing group");
2027         _use_group (pcrtable, group);
2028         return;
2029       }
2030       /* Else after group */
2031       prev = group;
2032     }
2033     _set_current_group (pcrtable, prev, pcr, offset, FALSE);
2034     return;
2035   }
2036
2037   corpcr = pcr - current->first_pcr;
2038   coroffset = offset - current->first_offset;
2039
2040   /* FIXME : Detect if we've gone into the next group !
2041    * FIXME : Close group when that happens */
2042   GST_DEBUG ("first:%d, last:%d, write:%d", current->first, current->last,
2043       current->write);
2044   GST_DEBUG ("First PCR:%" GST_TIME_FORMAT " offset:%" G_GUINT64_FORMAT,
2045       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (current->first_pcr)),
2046       current->first_offset);
2047   GST_DEBUG ("Last PCR: +%" GST_TIME_FORMAT " offset: +%" G_GUINT64_FORMAT,
2048       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (current->pending[current->last].pcr)),
2049       current->pending[current->last].offset);
2050   GST_DEBUG ("To add (corrected) PCR:%" GST_TIME_FORMAT " offset:%"
2051       G_GINT64_FORMAT, GST_TIME_ARGS (PCRTIME_TO_GSTTIME (corpcr)), coroffset);
2052
2053   /* Do we need to close the current group ? */
2054   /* Check for wrapover/discont */
2055   if (G_UNLIKELY (corpcr < current->pending[current->last].pcr)) {
2056     /* FIXME : ignore very small deltas (< 500ms ?) which are most likely
2057      * stray values */
2058     GST_DEBUG
2059         ("PCR smaller than previously observed one, handling discont/wrapover");
2060     /* Take values from current and put them in the current group (closing it) */
2061     /* Create new group with new pcr/offset just after the current group
2062      * and mark it as a wrapover */
2063     /* Initialize current to that group with new values */
2064     _append_group_values (current->group, current->pending[current->last]);
2065     _set_current_group (pcrtable, current->group, pcr, offset, TRUE);
2066     return;
2067   }
2068   /* If PCR diff is greater than 500ms, create new group */
2069   if (G_UNLIKELY (corpcr - current->pending[current->last].pcr >
2070           500 * PCR_MSECOND)) {
2071     GST_DEBUG ("New PCR more than 500ms away, handling discont");
2072     /* Take values from current and put them in the current group (closing it) */
2073     /* Create new group with pcr/offset just after the current group
2074      * and mark it as a discont */
2075     /* Initialize current to that group with new values */
2076     _append_group_values (current->group, current->pending[current->last]);
2077     _set_current_group (pcrtable, current->group, pcr, offset, TRUE);
2078     return;
2079   }
2080
2081   if (G_UNLIKELY (corpcr == current->last_value.pcr)) {
2082     GST_DEBUG ("Ignoring same PCR (stream is drunk)");
2083     return;
2084   }
2085
2086   /* update current window */
2087   current->pending[current->write].pcr = corpcr;
2088   current->pending[current->write].offset = coroffset;
2089   current->last_value = current->pending[current->write];
2090   current->last = (current->last + 1) % PCR_BITRATE_NEEDED;
2091   current->write = (current->write + 1) % PCR_BITRATE_NEEDED;
2092
2093   GST_DEBUG ("first:%d, last:%d, write:%d", current->first, current->last,
2094       current->write);
2095   GST_DEBUG ("First PCR:%" GST_TIME_FORMAT " offset:%" G_GUINT64_FORMAT,
2096       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (current->first_pcr)),
2097       current->first_offset);
2098   GST_DEBUG ("Last PCR: +%" GST_TIME_FORMAT " offset: +%" G_GUINT64_FORMAT,
2099       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (current->pending[current->last].pcr)),
2100       current->pending[current->last].offset);
2101
2102   /* If we haven't stored enough values, bail out */
2103   if (current->write != current->first) {
2104     GST_DEBUG
2105         ("Not enough observations to calculate bitrate (first:%d, last:%d)",
2106         current->first, current->last);
2107     return;
2108   }
2109
2110   /* If we are at least 1s away from reference value AND we have filled our
2111    * window, we can start comparing bitrates */
2112   if (current->pending[current->first].pcr - current->prev.pcr > PCR_SECOND) {
2113     /* Calculate window bitrate */
2114     current->cur_bitrate = gst_util_uint64_scale (PCR_SECOND,
2115         current->pending[current->last].offset -
2116         current->pending[current->first].offset,
2117         current->pending[current->last].pcr -
2118         current->pending[current->first].pcr);
2119     GST_DEBUG ("Current bitrate is now %" G_GUINT64_FORMAT,
2120         current->cur_bitrate);
2121
2122     /* Calculate previous bitrate */
2123     current->prev_bitrate =
2124         gst_util_uint64_scale (PCR_SECOND,
2125         current->pending[current->first].offset - current->prev.offset,
2126         current->pending[current->first].pcr - current->prev.pcr);
2127     GST_DEBUG ("Previous group bitrate now %" G_GUINT64_FORMAT,
2128         current->prev_bitrate);
2129
2130     /* FIXME : Better bitrate changes ? Currently 10% changes */
2131     if (ABSDIFF (current->cur_bitrate,
2132             current->prev_bitrate) * 10 > current->prev_bitrate) {
2133       GST_DEBUG ("Current bitrate changed by more than 10%% (old:%"
2134           G_GUINT64_FORMAT " new:%" G_GUINT64_FORMAT ")", current->prev_bitrate,
2135           current->cur_bitrate);
2136       /* If we detected a change in bitrate, this means that
2137        * d(first - prev) is a different bitrate than d(last - first).
2138        *
2139        * Two conclusions can be made:
2140        * 1) d(first - prev) is a complete bitrate "chain" (values between the
2141        *    reference value and first pending value have consistent bitrate).
2142        * 2) next values (from second pending value onwards) will no longer have
2143        *    the same bitrate.
2144        *
2145        * The question remains as to how long the new bitrate change is going to
2146        * last for (it might be short or longer term). For this we need to restart
2147        * bitrate estimation.
2148        *
2149        * * We move over first to the last value of group (a new chain ends and
2150        *   starts from there)
2151        * * We remember that last group value as our new window reference
2152        * * We restart our window filing from the last observed value
2153        *
2154        * Once our new window is filled we will end up in two different scenarios:
2155        * 1) Either the bitrate change was consistent, and therefore the bitrate
2156        *    will have remained constant over at least 2 window length
2157        * 2) The bitrate change was very short (1 window duration) and we will
2158        *    close that chain and restart again.
2159        * X) And of course if any discont/gaps/wrapover happen in the meantime they
2160        *    will also close the group.
2161        */
2162       _append_group_values (current->group, current->pending[current->first]);
2163       current->prev = current->pending[current->first];
2164       current->first = current->last;
2165       current->write = (current->first + 1) % PCR_BITRATE_NEEDED;
2166       return;
2167     }
2168   }
2169
2170   /* Update read position */
2171   current->first = (current->first + 1) % PCR_BITRATE_NEEDED;
2172 }
2173
2174
2175 /* convert specified offset into stream time */
2176 GstClockTime
2177 mpegts_packetizer_offset_to_ts (MpegTSPacketizer2 * packetizer,
2178     guint64 offset, guint16 pid)
2179 {
2180   PCROffsetGroup *last;
2181   MpegTSPCR *pcrtable;
2182   GList *tmp;
2183   GstClockTime res;
2184   guint64 lastpcr, lastoffset;
2185
2186   GST_DEBUG ("offset %" G_GUINT64_FORMAT, offset);
2187
2188   if (G_UNLIKELY (!packetizer->calculate_offset))
2189     return GST_CLOCK_TIME_NONE;
2190
2191   if (G_UNLIKELY (packetizer->refoffset == -1))
2192     return GST_CLOCK_TIME_NONE;
2193
2194   if (G_UNLIKELY (offset < packetizer->refoffset))
2195     return GST_CLOCK_TIME_NONE;
2196
2197   PACKETIZER_GROUP_LOCK (packetizer);
2198
2199   pcrtable = get_pcr_table (packetizer, pid);
2200
2201   if (g_list_length (pcrtable->groups) < 1) {
2202     PACKETIZER_GROUP_UNLOCK (packetizer);
2203     GST_WARNING ("Not enough observations to return a duration estimate");
2204     return GST_CLOCK_TIME_NONE;
2205   }
2206
2207   if (g_list_length (pcrtable->groups) > 1) {
2208     GST_LOG ("Using last group");
2209
2210     /* FIXME : Refine this later to use neighbouring groups */
2211     tmp = g_list_last (pcrtable->groups);
2212     last = tmp->data;
2213
2214     if (G_UNLIKELY (last->flags & PCR_GROUP_FLAG_ESTIMATED))
2215       _reevaluate_group_pcr_offset (pcrtable, last);
2216
2217     /* lastpcr is the full value in PCR from the first first chunk of data */
2218     lastpcr = last->values[last->last_value].pcr + last->pcr_offset;
2219     /* lastoffset is the full offset from the first chunk of data */
2220     lastoffset =
2221         last->values[last->last_value].offset + last->first_offset -
2222         packetizer->refoffset;
2223   } else {
2224     PCROffsetCurrent *current = pcrtable->current;
2225
2226     if (!current->group) {
2227       PACKETIZER_GROUP_UNLOCK (packetizer);
2228       GST_LOG ("No PCR yet");
2229       return GST_CLOCK_TIME_NONE;
2230     }
2231     /* If doing progressive read, use current */
2232     GST_LOG ("Using current group");
2233     lastpcr = current->group->pcr_offset + current->pending[current->last].pcr;
2234     lastoffset = current->first_offset + current->pending[current->last].offset;
2235   }
2236   GST_DEBUG ("lastpcr:%" GST_TIME_FORMAT " lastoffset:%" G_GUINT64_FORMAT
2237       " refoffset:%" G_GUINT64_FORMAT,
2238       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (lastpcr)), lastoffset,
2239       packetizer->refoffset);
2240
2241   /* Convert byte difference into time difference (and transformed from 27MHz to 1GHz) */
2242   res =
2243       PCRTIME_TO_GSTTIME (gst_util_uint64_scale (offset - packetizer->refoffset,
2244           lastpcr, lastoffset));
2245
2246   PACKETIZER_GROUP_UNLOCK (packetizer);
2247
2248   GST_DEBUG ("Returning timestamp %" GST_TIME_FORMAT " for offset %"
2249       G_GUINT64_FORMAT, GST_TIME_ARGS (res), offset);
2250
2251   return res;
2252 }
2253
2254 /* Input  : local PTS (in GHz units)
2255  * Return : Stream time (in GHz units) */
2256 GstClockTime
2257 mpegts_packetizer_pts_to_ts (MpegTSPacketizer2 * packetizer,
2258     GstClockTime pts, guint16 pcr_pid)
2259 {
2260   GstClockTime res = GST_CLOCK_TIME_NONE;
2261   MpegTSPCR *pcrtable;
2262
2263   PACKETIZER_GROUP_LOCK (packetizer);
2264   pcrtable = get_pcr_table (packetizer, pcr_pid);
2265
2266   if (!GST_CLOCK_TIME_IS_VALID (pcrtable->base_time) && pcr_pid == 0x1fff &&
2267       GST_CLOCK_TIME_IS_VALID (packetizer->last_in_time)) {
2268     pcrtable->base_time = packetizer->last_in_time;
2269     pcrtable->base_pcrtime = pts;
2270   }
2271
2272   /* Use clock skew if present */
2273   if (packetizer->calculate_skew
2274       && GST_CLOCK_TIME_IS_VALID (pcrtable->base_time)) {
2275     GST_DEBUG ("pts %" GST_TIME_FORMAT " base_pcrtime:%" GST_TIME_FORMAT
2276         " base_time:%" GST_TIME_FORMAT " pcroffset:%" GST_TIME_FORMAT,
2277         GST_TIME_ARGS (pts),
2278         GST_TIME_ARGS (pcrtable->base_pcrtime),
2279         GST_TIME_ARGS (pcrtable->base_time),
2280         GST_TIME_ARGS (pcrtable->pcroffset));
2281     res = pts + pcrtable->pcroffset + packetizer->extra_shift;
2282
2283     /* Don't return anything if we differ too much against last seen PCR */
2284     if (G_UNLIKELY (pcr_pid != 0x1fff &&
2285             ABSDIFF (res, pcrtable->last_pcrtime) > 15 * GST_SECOND))
2286       res = GST_CLOCK_TIME_NONE;
2287     else {
2288       GstClockTime tmp = pcrtable->base_time + pcrtable->skew;
2289       if (tmp + res >= pcrtable->base_pcrtime) {
2290         res += tmp - pcrtable->base_pcrtime;
2291       } else if (ABSDIFF (tmp + res + PCR_GST_MAX_VALUE,
2292               pcrtable->base_pcrtime) < PCR_GST_MAX_VALUE / 2) {
2293         /* Handle wrapover */
2294         res += tmp + PCR_GST_MAX_VALUE - pcrtable->base_pcrtime;
2295       } else {
2296         /* Fallback for values that differ way too much */
2297         res = GST_CLOCK_TIME_NONE;
2298       }
2299     }
2300   } else if (packetizer->calculate_offset && pcrtable->groups) {
2301     gint64 refpcr = G_MAXINT64, refpcroffset;
2302     PCROffsetGroup *group = pcrtable->current->group;
2303
2304     /* Generic calculation:
2305      * Stream Time = PTS - first group PCR + group PCR_offset
2306      *
2307      * In case of wrapover:
2308      * Stream Time = PTS + MAX_PCR - first group PCR + group PCR_offset
2309      * (which we actually do by using first group PCR -= MAX_PCR in order
2310      *  to end up with the same calculation as for non-wrapover) */
2311
2312     if (group) {
2313       /* If we have a current group the value is pretty much guaranteed */
2314       GST_DEBUG ("Using current First PCR:%" GST_TIME_FORMAT " offset:%"
2315           G_GUINT64_FORMAT " PCR_offset:%" GST_TIME_FORMAT,
2316           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->first_pcr)),
2317           group->first_offset,
2318           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->pcr_offset)));
2319       refpcr = group->first_pcr;
2320       refpcroffset = group->pcr_offset;
2321       if (pts < PCRTIME_TO_GSTTIME (refpcr)) {
2322         /* Only apply wrapover if we're certain it is, and avoid
2323          * returning bogus values if it's a PTS/DTS which is *just*
2324          * before the start of the current group
2325          */
2326         if (PCRTIME_TO_GSTTIME (refpcr) - pts > GST_SECOND) {
2327           pts += PCR_GST_MAX_VALUE;
2328         } else
2329           refpcr = G_MAXINT64;
2330       }
2331     } else {
2332       GList *tmp;
2333       /* Otherwise, find a suitable group */
2334
2335       GST_DEBUG ("Find group for current offset %" G_GUINT64_FORMAT,
2336           packetizer->offset);
2337
2338       for (tmp = pcrtable->groups; tmp; tmp = tmp->next) {
2339         PCROffsetGroup *tgroup = tmp->data;
2340         GST_DEBUG ("Trying First PCR:%" GST_TIME_FORMAT " offset:%"
2341             G_GUINT64_FORMAT " PCR_offset:%" GST_TIME_FORMAT,
2342             GST_TIME_ARGS (PCRTIME_TO_GSTTIME (tgroup->first_pcr)),
2343             tgroup->first_offset,
2344             GST_TIME_ARGS (PCRTIME_TO_GSTTIME (tgroup->pcr_offset)));
2345         /* Gone too far ? */
2346         if (tgroup->first_offset > packetizer->offset) {
2347           /* If there isn't a pending reset, use that value */
2348           if (group) {
2349             GST_DEBUG ("PTS is %" GST_TIME_FORMAT " into group",
2350                 GST_TIME_ARGS (pts - PCRTIME_TO_GSTTIME (group->first_pcr)));
2351           }
2352           break;
2353         }
2354         group = tgroup;
2355         /* In that group ? */
2356         if (group->first_offset + group->values[group->last_value].offset >
2357             packetizer->offset) {
2358           GST_DEBUG ("PTS is %" GST_TIME_FORMAT " into group",
2359               GST_TIME_ARGS (pts - PCRTIME_TO_GSTTIME (group->first_pcr)));
2360           break;
2361         }
2362       }
2363       if (group && !(group->flags & PCR_GROUP_FLAG_RESET)) {
2364         GST_DEBUG ("Using group !");
2365         refpcr = group->first_pcr;
2366         refpcroffset = group->pcr_offset;
2367         if (pts < PCRTIME_TO_GSTTIME (refpcr)) {
2368           if (PCRTIME_TO_GSTTIME (refpcr) - pts > GST_SECOND)
2369             pts += PCR_GST_MAX_VALUE;
2370           else
2371             refpcr = G_MAXINT64;
2372         }
2373       }
2374     }
2375     if (refpcr != G_MAXINT64)
2376       res =
2377           pts - PCRTIME_TO_GSTTIME (refpcr) + PCRTIME_TO_GSTTIME (refpcroffset);
2378     else
2379       GST_WARNING ("No groups, can't calculate timestamp");
2380   } else
2381     GST_WARNING ("Not enough information to calculate proper timestamp");
2382
2383   PACKETIZER_GROUP_UNLOCK (packetizer);
2384
2385   GST_DEBUG ("Returning timestamp %" GST_TIME_FORMAT " for pts %"
2386       GST_TIME_FORMAT " pcr_pid:0x%04x", GST_TIME_ARGS (res),
2387       GST_TIME_ARGS (pts), pcr_pid);
2388   return res;
2389 }
2390
2391 /* Stream time to offset */
2392 guint64
2393 mpegts_packetizer_ts_to_offset (MpegTSPacketizer2 * packetizer,
2394     GstClockTime ts, guint16 pcr_pid)
2395 {
2396   MpegTSPCR *pcrtable;
2397   guint64 res;
2398   PCROffsetGroup *nextgroup = NULL, *prevgroup = NULL;
2399   guint64 querypcr, firstpcr, lastpcr, firstoffset, lastoffset;
2400   PCROffsetCurrent *current;
2401   GList *tmp;
2402
2403   if (!packetizer->calculate_offset)
2404     return -1;
2405
2406   PACKETIZER_GROUP_LOCK (packetizer);
2407   pcrtable = get_pcr_table (packetizer, pcr_pid);
2408
2409   if (pcrtable->groups == NULL) {
2410     PACKETIZER_GROUP_UNLOCK (packetizer);
2411     return -1;
2412   }
2413
2414   querypcr = GSTTIME_TO_PCRTIME (ts);
2415
2416   GST_DEBUG ("Searching offset for ts %" GST_TIME_FORMAT, GST_TIME_ARGS (ts));
2417
2418   /* First check if we're within the current pending group */
2419   current = pcrtable->current;
2420   if (current && current->group && (querypcr >= current->group->pcr_offset) &&
2421       querypcr - current->group->pcr_offset <=
2422       current->pending[current->last].pcr) {
2423     GST_DEBUG ("pcr is in current group");
2424     nextgroup = current->group;
2425     goto calculate_points;
2426   }
2427
2428   /* Find the neighbouring groups */
2429   for (tmp = pcrtable->groups; tmp; tmp = tmp->next) {
2430     nextgroup = (PCROffsetGroup *) tmp->data;
2431
2432     GST_DEBUG ("Trying group PCR %" GST_TIME_FORMAT " (offset %"
2433         G_GUINT64_FORMAT " pcr_offset %" GST_TIME_FORMAT,
2434         GST_TIME_ARGS (PCRTIME_TO_GSTTIME (nextgroup->first_pcr)),
2435         nextgroup->first_offset,
2436         GST_TIME_ARGS (PCRTIME_TO_GSTTIME (nextgroup->pcr_offset)));
2437
2438     /* Check if we've gone too far */
2439     if (nextgroup->pcr_offset > querypcr) {
2440       GST_DEBUG ("pcr is before that group");
2441       break;
2442     }
2443
2444     if (tmp->next == NULL) {
2445       GST_DEBUG ("pcr is beyond last group");
2446       break;
2447     }
2448
2449     prevgroup = nextgroup;
2450
2451     /* Maybe it's in this group */
2452     if (nextgroup->values[nextgroup->last_value].pcr +
2453         nextgroup->pcr_offset >= querypcr) {
2454       GST_DEBUG ("pcr is in that group");
2455       break;
2456     }
2457   }
2458
2459 calculate_points:
2460
2461   GST_DEBUG ("nextgroup:%p, prevgroup:%p", nextgroup, prevgroup);
2462
2463   if (nextgroup == prevgroup || prevgroup == NULL) {
2464     /* We use the current group to calculate position:
2465      * * if the PCR is within this group
2466      * * if there is only one group to use for calculation
2467      */
2468     GST_DEBUG ("In group or after last one");
2469     lastoffset = firstoffset = nextgroup->first_offset;
2470     lastpcr = firstpcr = nextgroup->pcr_offset;
2471     if (current && nextgroup == current->group) {
2472       lastoffset += current->pending[current->last].offset;
2473       lastpcr += current->pending[current->last].pcr;
2474     } else {
2475       lastoffset += nextgroup->values[nextgroup->last_value].offset;
2476       lastpcr += nextgroup->values[nextgroup->last_value].pcr;
2477     }
2478   } else {
2479     GST_DEBUG ("Between group");
2480     lastoffset = nextgroup->first_offset;
2481     lastpcr = nextgroup->pcr_offset;
2482     firstoffset =
2483         prevgroup->values[prevgroup->last_value].offset +
2484         prevgroup->first_offset;
2485     firstpcr =
2486         prevgroup->values[prevgroup->last_value].pcr + prevgroup->pcr_offset;
2487   }
2488
2489   PACKETIZER_GROUP_UNLOCK (packetizer);
2490
2491   GST_DEBUG ("Using prev PCR %" G_GUINT64_FORMAT " offset %" G_GUINT64_FORMAT,
2492       firstpcr, firstoffset);
2493   GST_DEBUG ("Using last PCR %" G_GUINT64_FORMAT " offset %" G_GUINT64_FORMAT,
2494       lastpcr, lastoffset);
2495
2496   res = firstoffset;
2497   if (lastpcr != firstpcr)
2498     res += gst_util_uint64_scale (querypcr - firstpcr,
2499         lastoffset - firstoffset, lastpcr - firstpcr);
2500
2501   GST_DEBUG ("Returning offset %" G_GUINT64_FORMAT " for ts %"
2502       GST_TIME_FORMAT, res, GST_TIME_ARGS (ts));
2503
2504   return res;
2505 }
2506
2507 void
2508 mpegts_packetizer_set_reference_offset (MpegTSPacketizer2 * packetizer,
2509     guint64 refoffset)
2510 {
2511   GST_DEBUG ("Setting reference offset to %" G_GUINT64_FORMAT, refoffset);
2512
2513   PACKETIZER_GROUP_LOCK (packetizer);
2514   packetizer->refoffset = refoffset;
2515   PACKETIZER_GROUP_UNLOCK (packetizer);
2516 }
2517
2518 void
2519 mpegts_packetizer_set_pcr_discont_threshold (MpegTSPacketizer2 * packetizer,
2520     GstClockTime threshold)
2521 {
2522   PACKETIZER_GROUP_LOCK (packetizer);
2523   packetizer->pcr_discont_threshold = threshold;
2524   PACKETIZER_GROUP_UNLOCK (packetizer);
2525 }
2526
2527 void
2528 mpegts_packetizer_set_current_pcr_offset (MpegTSPacketizer2 * packetizer,
2529     GstClockTime offset, guint16 pcr_pid)
2530 {
2531   guint64 pcr_offset;
2532   gint64 delta;
2533   MpegTSPCR *pcrtable;
2534   PCROffsetGroup *group;
2535   GList *tmp;
2536   gboolean apply = FALSE;
2537
2538   /* fast path */
2539   PACKETIZER_GROUP_LOCK (packetizer);
2540   pcrtable = get_pcr_table (packetizer, pcr_pid);
2541
2542   if (pcrtable == NULL || pcrtable->current->group == NULL) {
2543     PACKETIZER_GROUP_UNLOCK (packetizer);
2544     return;
2545   }
2546
2547   pcr_offset = GSTTIME_TO_PCRTIME (offset);
2548
2549   /* Pick delta from *first* group */
2550   if (pcrtable->groups)
2551     group = pcrtable->groups->data;
2552   else
2553     group = pcrtable->current->group;
2554   GST_DEBUG ("Current group PCR %" GST_TIME_FORMAT " (offset %"
2555       G_GUINT64_FORMAT " pcr_offset %" GST_TIME_FORMAT,
2556       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->first_pcr)),
2557       group->first_offset,
2558       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->pcr_offset)));
2559
2560   /* Remember the difference between previous initial pcr_offset and
2561    * new initial pcr_offset */
2562   delta = pcr_offset - group->pcr_offset;
2563   if (delta == 0) {
2564     GST_DEBUG ("No shift to apply");
2565     PACKETIZER_GROUP_UNLOCK (packetizer);
2566     return;
2567   }
2568   GST_DEBUG ("Shifting groups by %" GST_TIME_FORMAT
2569       " for new initial pcr_offset %" GST_TIME_FORMAT,
2570       GST_TIME_ARGS (PCRTIME_TO_GSTTIME (delta)), GST_TIME_ARGS (offset));
2571
2572   for (tmp = pcrtable->groups; tmp; tmp = tmp->next) {
2573     PCROffsetGroup *tgroup = (tmp->data);
2574     if (tgroup == group)
2575       apply = TRUE;
2576     if (apply) {
2577       tgroup->pcr_offset += delta;
2578       GST_DEBUG ("Update group PCR %" GST_TIME_FORMAT " (offset %"
2579           G_GUINT64_FORMAT " pcr_offset %" GST_TIME_FORMAT,
2580           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (tgroup->first_pcr)),
2581           tgroup->first_offset,
2582           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (tgroup->pcr_offset)));
2583     } else
2584       GST_DEBUG ("Not modifying group PCR %" GST_TIME_FORMAT " (offset %"
2585           G_GUINT64_FORMAT " pcr_offset %" GST_TIME_FORMAT,
2586           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (tgroup->first_pcr)),
2587           tgroup->first_offset,
2588           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (tgroup->pcr_offset)));
2589   }
2590   PACKETIZER_GROUP_UNLOCK (packetizer);
2591 }