subprojects/gst-plugins-bad/gst/mpegtsmux/gstbasetsmux.c

   1 /*
   2  * Copyright 2006, 2007, 2008, 2009, 2010 Fluendo S.A.
   3  *  Authors: Jan Schmidt <jan@fluendo.com>
   4  *           Kapil Agrawal <kapil@fluendo.com>
   5  *           Julien Moutte <julien@fluendo.com>
   6  *
   7  * Copyright (C) 2011 Jan Schmidt <thaytan@noraisin.net>
   8  *
   9  * This library is licensed under 3 different licenses and you
  10  * can choose to use it under the terms of any one of them. The
  11  * three licenses are the MPL 1.1, the LGPL and the MIT license.
  12  *
  13  * MPL:
  14  *
  15  * The contents of this file are subject to the Mozilla Public License
  16  * Version 1.1 (the "License"); you may not use this file except in
  17  * compliance with the License. You may obtain a copy of the License at
  18  * http://www.mozilla.org/MPL/.
  19  *
  20  * Software distributed under the License is distributed on an "AS IS"
  21  * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
  22  * License for the specific language governing rights and limitations
  23  * under the License.
  24  *
  25  * LGPL:
  26  *
  27  * This library is free software; you can redistribute it and/or
  28  * modify it under the terms of the GNU Library General Public
  29  * License as published by the Free Software Foundation; either
  30  * version 2 of the License, or (at your option) any later version.
  31  *
  32  * This library is distributed in the hope that it will be useful,
  33  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  34  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  35  * Library General Public License for more details.
  36  *
  37  * You should have received a copy of the GNU Library General Public
  38  * License along with this library; if not, write to the
  39  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  40  * Boston, MA 02110-1301, USA.
  41  *
  42  * MIT:
  43  *
  44  * Unless otherwise indicated, Source Code is licensed under MIT license.
  45  * See further explanation attached in License Statement (distributed in the file
  46  * LICENSE).
  47  *
  48  * Permission is hereby granted, free of charge, to any person obtaining a copy of
  49  * this software and associated documentation files (the "Software"), to deal in
  50  * the Software without restriction, including without limitation the rights to
  51  * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
  52  * of the Software, and to permit persons to whom the Software is furnished to do
  53  * so, subject to the following conditions:
  54  *
  55  * The above copyright notice and this permission notice shall be included in all
  56  * copies or substantial portions of the Software.
  57  *
  58  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  59  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  60  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  61  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  62  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  63  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  64  * SOFTWARE.
  65  *
  66  * SPDX-License-Identifier: MPL-1.1 OR MIT OR LGPL-2.0-or-later
  67  */
  68
  69 #include <stdio.h>
  70 #include <string.h>
  71
  72 #include <gst/tag/tag.h>
  73 #include <gst/video/video.h>
  74 #include <gst/mpegts/mpegts.h>
  75 #include <gst/pbutils/pbutils.h>
  76 #include <gst/videoparsers/gstjpeg2000parse.h>
  77 #include <gst/video/video-color.h>
  78
  79 #include "gstbasetsmux.h"
  80 #include "gstbasetsmuxaac.h"
  81 #include "gstbasetsmuxttxt.h"
  82 #include "gstbasetsmuxopus.h"
  83 #include "gstbasetsmuxjpeg2000.h"
  84
  85 GST_DEBUG_CATEGORY (gst_base_ts_mux_debug);
  86 #define GST_CAT_DEFAULT gst_base_ts_mux_debug
  87
  88 /* GstBaseTsMuxPad */
  89
  90 G_DEFINE_TYPE (GstBaseTsMuxPad, gst_base_ts_mux_pad, GST_TYPE_AGGREGATOR_PAD);
  91
  92 /* Internals */
  93
  94 static void
  95 gst_base_ts_mux_pad_reset (GstBaseTsMuxPad * pad)
  96 {
  97   pad->dts = GST_CLOCK_STIME_NONE;
  98   pad->prog_id = -1;
  99
 100   if (pad->free_func)
 101     pad->free_func (pad->prepare_data);
 102   pad->prepare_data = NULL;
 103   pad->prepare_func = NULL;
 104   pad->free_func = NULL;
 105
 106   if (pad->codec_data)
 107     gst_buffer_replace (&pad->codec_data, NULL);
 108
 109   /* reference owned elsewhere */
 110   pad->stream = NULL;
 111   pad->prog = NULL;
 112
 113   if (pad->language) {
 114     g_free (pad->language);
 115     pad->language = NULL;
 116   }
 117 }
 118
 119 /* GstAggregatorPad implementation */
 120
 121 static GstFlowReturn
 122 gst_base_ts_mux_pad_flush (GstAggregatorPad * agg_pad, GstAggregator * agg)
 123 {
 124   GList *cur;
 125   GstBaseTsMux *mux = GST_BASE_TS_MUX (agg);
 126
 127   /* Send initial segments again after a flush-stop, and also resend the
 128    * header sections */
 129   mux->first = TRUE;
 130
 131   /* output PAT, SI tables */
 132   tsmux_resend_pat (mux->tsmux);
 133   tsmux_resend_si (mux->tsmux);
 134
 135   /* output PMT for each program */
 136   for (cur = mux->tsmux->programs; cur; cur = cur->next) {
 137     TsMuxProgram *program = (TsMuxProgram *) cur->data;
 138
 139     tsmux_resend_pmt (program);
 140   }
 141
 142   return GST_FLOW_OK;
 143 }
 144
 145 /* GObject implementation */
 146
 147 static void
 148 gst_base_ts_mux_pad_dispose (GObject * obj)
 149 {
 150   GstBaseTsMuxPad *ts_pad = GST_BASE_TS_MUX_PAD (obj);
 151
 152   gst_base_ts_mux_pad_reset (ts_pad);
 153
 154   G_OBJECT_CLASS (gst_base_ts_mux_pad_parent_class)->dispose (obj);
 155 }
 156
 157 static void
 158 gst_base_ts_mux_pad_class_init (GstBaseTsMuxPadClass * klass)
 159 {
 160   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 161   GstAggregatorPadClass *gstaggpad_class = GST_AGGREGATOR_PAD_CLASS (klass);
 162
 163   gobject_class->dispose = gst_base_ts_mux_pad_dispose;
 164   gstaggpad_class->flush = gst_base_ts_mux_pad_flush;
 165
 166   gst_type_mark_as_plugin_api (GST_TYPE_BASE_TS_MUX, 0);
 167 }
 168
 169 static void
 170 gst_base_ts_mux_pad_init (GstBaseTsMuxPad * vaggpad)
 171 {
 172 }
 173
 174 /* GstBaseTsMux */
 175
 176 enum
 177 {
 178   PROP_0,
 179   PROP_PROG_MAP,
 180   PROP_PAT_INTERVAL,
 181   PROP_PMT_INTERVAL,
 182   PROP_ALIGNMENT,
 183   PROP_SI_INTERVAL,
 184   PROP_BITRATE,
 185   PROP_PCR_INTERVAL,
 186   PROP_SCTE_35_PID,
 187   PROP_SCTE_35_NULL_INTERVAL
 188 };
 189
 190 #define DEFAULT_SCTE_35_PID 0
 191
 192 #define BASETSMUX_DEFAULT_ALIGNMENT    -1
 193
 194 #define CLOCK_BASE 9LL
 195 #define CLOCK_FREQ (CLOCK_BASE * 10000) /* 90 kHz PTS clock */
 196 #define CLOCK_FREQ_SCR (CLOCK_FREQ * 300)       /* 27 MHz SCR clock */
 197 #define TS_MUX_CLOCK_BASE (TSMUX_CLOCK_FREQ * 10 * 360)
 198
 199 #define GSTTIME_TO_MPEGTIME(time) \
 200     (((time) > 0 ? (gint64) 1 : (gint64) -1) * \
 201     (gint64) gst_util_uint64_scale (ABS(time), CLOCK_BASE, GST_MSECOND/10))
 202 /* 27 MHz SCR conversions: */
 203 #define MPEG_SYS_TIME_TO_GSTTIME(time) (gst_util_uint64_scale ((time), \
 204                         GST_USECOND, CLOCK_FREQ_SCR / 1000000))
 205 #define GSTTIME_TO_MPEG_SYS_TIME(time) (gst_util_uint64_scale ((time), \
 206                         CLOCK_FREQ_SCR / 1000000, GST_USECOND))
 207
 208 #define DEFAULT_PROG_ID 0
 209
 210 static GstStaticPadTemplate gst_base_ts_mux_src_factory =
 211 GST_STATIC_PAD_TEMPLATE ("src",
 212     GST_PAD_SRC,
 213     GST_PAD_ALWAYS,
 214     GST_STATIC_CAPS ("video/mpegts, "
 215         "systemstream = (boolean) true, " "packetsize = (int) { 188, 192} ")
 216     );
 217
 218 typedef struct
 219 {
 220   GstMapInfo map_info;
 221   GstBuffer *buffer;
 222 } StreamData;
 223
 224 G_DEFINE_TYPE_WITH_CODE (GstBaseTsMux, gst_base_ts_mux, GST_TYPE_AGGREGATOR,
 225     gst_mpegts_initialize ());
 226
 227 /* Internals */
 228
 229 /* Takes over the ref on the buffer */
 230 static StreamData *
 231 stream_data_new (GstBuffer * buffer)
 232 {
 233   StreamData *res = g_new (StreamData, 1);
 234   res->buffer = buffer;
 235   gst_buffer_map (buffer, &(res->map_info), GST_MAP_READ);
 236
 237   return res;
 238 }
 239
 240 static void
 241 stream_data_free (StreamData * data)
 242 {
 243   if (data) {
 244     gst_buffer_unmap (data->buffer, &data->map_info);
 245     gst_buffer_unref (data->buffer);
 246     g_free (data);
 247   }
 248 }
 249
 250 #define parent_class gst_base_ts_mux_parent_class
 251
 252 static void
 253 gst_base_ts_mux_set_header_on_caps (GstBaseTsMux * mux)
 254 {
 255   GstBuffer *buf;
 256   GstStructure *structure;
 257   GValue array = { 0 };
 258   GValue value = { 0 };
 259   GstCaps *caps;
 260
 261   caps = gst_pad_get_current_caps (GST_AGGREGATOR_SRC_PAD (mux));
 262
 263   /* If we have no caps, we are possibly shutting down */
 264   if (!caps)
 265     return;
 266
 267   caps = gst_caps_make_writable (caps);
 268   structure = gst_caps_get_structure (caps, 0);
 269
 270   g_value_init (&array, GST_TYPE_ARRAY);
 271
 272   GST_LOG_OBJECT (mux, "setting %u packets into streamheader",
 273       g_queue_get_length (&mux->streamheader));
 274
 275   while ((buf = GST_BUFFER (g_queue_pop_head (&mux->streamheader)))) {
 276     g_value_init (&value, GST_TYPE_BUFFER);
 277     gst_value_take_buffer (&value, buf);
 278     gst_value_array_append_value (&array, &value);
 279     g_value_unset (&value);
 280   }
 281
 282   gst_structure_set_value (structure, "streamheader", &array);
 283   gst_aggregator_set_src_caps (GST_AGGREGATOR (mux), caps);
 284   g_value_unset (&array);
 285   gst_caps_unref (caps);
 286 }
 287
 288 static gboolean
 289 steal_si_section (GstMpegtsSectionType * type, TsMuxSection * section,
 290     TsMux * mux)
 291 {
 292   g_hash_table_insert (mux->si_sections, type, section);
 293
 294   return TRUE;
 295 }
 296
 297 static void
 298 gst_base_ts_mux_reset (GstBaseTsMux * mux, gboolean alloc)
 299 {
 300   GstBuffer *buf;
 301   GstBaseTsMuxClass *klass = GST_BASE_TS_MUX_GET_CLASS (mux);
 302   GHashTable *si_sections = NULL;
 303   GList *l;
 304
 305   mux->first = TRUE;
 306   mux->last_flow_ret = GST_FLOW_OK;
 307   mux->last_ts = 0;
 308   mux->is_delta = TRUE;
 309   mux->is_header = FALSE;
 310
 311   mux->streamheader_sent = FALSE;
 312   mux->pending_key_unit_ts = GST_CLOCK_TIME_NONE;
 313   gst_event_replace (&mux->force_key_unit_event, NULL);
 314
 315   if (mux->out_adapter)
 316     gst_adapter_clear (mux->out_adapter);
 317   mux->output_ts_offset = GST_CLOCK_TIME_NONE;
 318
 319   if (mux->tsmux) {
 320     if (mux->tsmux->si_sections)
 321       si_sections = g_hash_table_ref (mux->tsmux->si_sections);
 322
 323     tsmux_free (mux->tsmux);
 324     mux->tsmux = NULL;
 325   }
 326
 327   if (mux->programs) {
 328     g_hash_table_destroy (mux->programs);
 329   }
 330   mux->programs = g_hash_table_new (g_direct_hash, g_direct_equal);
 331
 332   while ((buf = GST_BUFFER (g_queue_pop_head (&mux->streamheader))))
 333     gst_buffer_unref (buf);
 334
 335   gst_event_replace (&mux->force_key_unit_event, NULL);
 336   gst_buffer_replace (&mux->out_buffer, NULL);
 337
 338   GST_OBJECT_LOCK (mux);
 339
 340   for (l = GST_ELEMENT (mux)->sinkpads; l; l = l->next) {
 341     gst_base_ts_mux_pad_reset (GST_BASE_TS_MUX_PAD (l->data));
 342   }
 343
 344   GST_OBJECT_UNLOCK (mux);
 345
 346   if (alloc) {
 347     g_assert (klass->create_ts_mux);
 348
 349     mux->tsmux = klass->create_ts_mux (mux);
 350
 351     /* Preserve user-specified sections across resets */
 352     if (si_sections)
 353       g_hash_table_foreach_steal (si_sections, (GHRFunc) steal_si_section,
 354           mux->tsmux);
 355   }
 356
 357   if (si_sections)
 358     g_hash_table_unref (si_sections);
 359
 360   mux->last_scte35_event_seqnum = GST_SEQNUM_INVALID;
 361
 362   if (klass->reset)
 363     klass->reset (mux);
 364 }
 365
 366 static void
 367 release_buffer_cb (guint8 * data, void *user_data)
 368 {
 369   stream_data_free ((StreamData *) user_data);
 370 }
 371
 372 static GstFlowReturn
 373 gst_base_ts_mux_create_or_update_stream (GstBaseTsMux * mux,
 374     GstBaseTsMuxPad * ts_pad, GstCaps * caps)
 375 {
 376   GstStructure *s;
 377   guint st = TSMUX_ST_RESERVED;
 378   const gchar *mt;
 379   const GValue *value = NULL;
 380   GstBuffer *codec_data = NULL;
 381   guint8 opus_channel_config_code = 0;
 382   guint16 profile = GST_JPEG2000_PARSE_PROFILE_NONE;
 383   guint8 main_level = 0;
 384   guint32 max_rate = 0;
 385   guint8 color_spec = 0;
 386   const gchar *stream_format = NULL;
 387   const char *interlace_mode = NULL;
 388
 389   GST_DEBUG_OBJECT (ts_pad,
 390       "%s stream with PID 0x%04x for caps %" GST_PTR_FORMAT,
 391       ts_pad->stream ? "Recreating" : "Creating", ts_pad->pid, caps);
 392
 393   s = gst_caps_get_structure (caps, 0);
 394
 395   mt = gst_structure_get_name (s);
 396   value = gst_structure_get_value (s, "codec_data");
 397   if (value != NULL)
 398     codec_data = gst_value_get_buffer (value);
 399
 400   g_clear_pointer (&ts_pad->codec_data, gst_buffer_unref);
 401   ts_pad->prepare_func = NULL;
 402
 403   stream_format = gst_structure_get_string (s, "stream-format");
 404
 405   if (strcmp (mt, "video/x-dirac") == 0) {
 406     st = TSMUX_ST_VIDEO_DIRAC;
 407   } else if (strcmp (mt, "audio/x-ac3") == 0) {
 408     st = TSMUX_ST_PS_AUDIO_AC3;
 409   } else if (strcmp (mt, "audio/x-dts") == 0) {
 410     st = TSMUX_ST_PS_AUDIO_DTS;
 411   } else if (strcmp (mt, "audio/x-lpcm") == 0) {
 412     st = TSMUX_ST_PS_AUDIO_LPCM;
 413   } else if (strcmp (mt, "video/x-h264") == 0) {
 414     st = TSMUX_ST_VIDEO_H264;
 415   } else if (strcmp (mt, "video/x-h265") == 0) {
 416     st = TSMUX_ST_VIDEO_HEVC;
 417   } else if (strcmp (mt, "audio/mpeg") == 0) {
 418     gint mpegversion;
 419
 420     if (!gst_structure_get_int (s, "mpegversion", &mpegversion)) {
 421       GST_ERROR_OBJECT (ts_pad, "caps missing mpegversion");
 422       goto not_negotiated;
 423     }
 424
 425     switch (mpegversion) {
 426       case 1:{
 427         int mpegaudioversion = 1;       /* Assume mpegaudioversion=1 for backwards compatibility */
 428         (void) gst_structure_get_int (s, "mpegaudioversion", &mpegaudioversion);
 429
 430         if (mpegaudioversion == 1)
 431           st = TSMUX_ST_AUDIO_MPEG1;
 432         else
 433           st = TSMUX_ST_AUDIO_MPEG2;
 434         break;
 435       }
 436       case 2:{
 437         /* mpegversion=2 in GStreamer refers to MPEG-2 Part 7 audio,  */
 438
 439         st = TSMUX_ST_AUDIO_AAC;
 440
 441         /* Check the stream format. If raw, make dummy internal codec data from the caps */
 442         if (g_strcmp0 (stream_format, "raw") == 0) {
 443           ts_pad->codec_data =
 444               gst_base_ts_mux_aac_mpeg2_make_codec_data (mux, caps);
 445           ts_pad->prepare_func = gst_base_ts_mux_prepare_aac_mpeg2;
 446           if (ts_pad->codec_data == NULL) {
 447             GST_ERROR_OBJECT (mux, "Invalid or incomplete caps for MPEG-2 AAC");
 448             goto not_negotiated;
 449           }
 450         }
 451         break;
 452       }
 453       case 4:
 454       {
 455         st = TSMUX_ST_AUDIO_AAC;
 456
 457         /* Check the stream format. We need codec_data with RAW streams and mpegversion=4 */
 458         if (g_strcmp0 (stream_format, "raw") == 0) {
 459           if (codec_data) {
 460             GST_DEBUG_OBJECT (ts_pad,
 461                 "we have additional codec data (%" G_GSIZE_FORMAT " bytes)",
 462                 gst_buffer_get_size (codec_data));
 463             ts_pad->codec_data = gst_buffer_ref (codec_data);
 464             ts_pad->prepare_func = gst_base_ts_mux_prepare_aac_mpeg4;
 465           } else {
 466             ts_pad->codec_data = NULL;
 467             GST_ERROR_OBJECT (mux, "Need codec_data for raw MPEG-4 AAC");
 468             goto not_negotiated;
 469           }
 470         } else if (codec_data) {
 471           ts_pad->codec_data = gst_buffer_ref (codec_data);
 472         } else {
 473           ts_pad->codec_data = NULL;
 474         }
 475         break;
 476       }
 477       default:
 478         GST_WARNING_OBJECT (ts_pad, "unsupported mpegversion %d", mpegversion);
 479         goto not_negotiated;
 480     }
 481   } else if (strcmp (mt, "video/mpeg") == 0) {
 482     gint mpegversion;
 483
 484     if (!gst_structure_get_int (s, "mpegversion", &mpegversion)) {
 485       GST_ERROR_OBJECT (ts_pad, "caps missing mpegversion");
 486       goto not_negotiated;
 487     }
 488
 489     switch (mpegversion) {
 490       case 1:
 491         st = TSMUX_ST_VIDEO_MPEG1;
 492         break;
 493       case 2:
 494         st = TSMUX_ST_VIDEO_MPEG2;
 495         break;
 496       case 4:
 497         st = TSMUX_ST_VIDEO_MPEG4;
 498         break;
 499       default:
 500         GST_WARNING_OBJECT (ts_pad, "unsupported mpegversion %d", mpegversion);
 501         goto not_negotiated;
 502     }
 503   } else if (strcmp (mt, "subpicture/x-dvb") == 0) {
 504     st = TSMUX_ST_PS_DVB_SUBPICTURE;
 505   } else if (strcmp (mt, "application/x-teletext") == 0) {
 506     st = TSMUX_ST_PS_TELETEXT;
 507     /* needs a particularly sized layout */
 508     ts_pad->prepare_func = gst_base_ts_mux_prepare_teletext;
 509   } else if (strcmp (mt, "audio/x-opus") == 0) {
 510     guint8 channels, mapping_family, stream_count, coupled_count;
 511     guint8 channel_mapping[256];
 512
 513     if (!gst_codec_utils_opus_parse_caps (caps, NULL, &channels,
 514             &mapping_family, &stream_count, &coupled_count, channel_mapping)) {
 515       GST_ERROR_OBJECT (ts_pad, "Incomplete Opus caps");
 516       goto not_negotiated;
 517     }
 518
 519     if (channels <= 2 && mapping_family == 0) {
 520       opus_channel_config_code = channels;
 521     } else if (channels == 2 && mapping_family == 255 && stream_count == 1
 522         && coupled_count == 1) {
 523       /* Dual mono */
 524       opus_channel_config_code = 0;
 525     } else if (channels >= 2 && channels <= 8 && mapping_family == 1) {
 526       static const guint8 coupled_stream_counts[9] = {
 527         1, 0, 1, 1, 2, 2, 2, 3, 3
 528       };
 529       static const guint8 channel_map_a[8][8] = {
 530         {0},
 531         {0, 1},
 532         {0, 2, 1},
 533         {0, 1, 2, 3},
 534         {0, 4, 1, 2, 3},
 535         {0, 4, 1, 2, 3, 5},
 536         {0, 4, 1, 2, 3, 5, 6},
 537         {0, 6, 1, 2, 3, 4, 5, 7},
 538       };
 539       static const guint8 channel_map_b[8][8] = {
 540         {0},
 541         {0, 1},
 542         {0, 1, 2},
 543         {0, 1, 2, 3},
 544         {0, 1, 2, 3, 4},
 545         {0, 1, 2, 3, 4, 5},
 546         {0, 1, 2, 3, 4, 5, 6},
 547         {0, 1, 2, 3, 4, 5, 6, 7},
 548       };
 549
 550       /* Vorbis mapping */
 551       if (stream_count == channels - coupled_stream_counts[channels] &&
 552           coupled_count == coupled_stream_counts[channels] &&
 553           memcmp (channel_mapping, channel_map_a[channels - 1],
 554               channels) == 0) {
 555         opus_channel_config_code = channels;
 556       } else if (stream_count == channels - coupled_stream_counts[channels] &&
 557           coupled_count == coupled_stream_counts[channels] &&
 558           memcmp (channel_mapping, channel_map_b[channels - 1],
 559               channels) == 0) {
 560         opus_channel_config_code = channels | 0x80;
 561       } else {
 562         GST_FIXME_OBJECT (ts_pad, "Opus channel mapping not handled");
 563         goto not_negotiated;
 564       }
 565     }
 566
 567     st = TSMUX_ST_PS_OPUS;
 568     ts_pad->prepare_func = gst_base_ts_mux_prepare_opus;
 569   } else if (strcmp (mt, "meta/x-klv") == 0) {
 570     st = TSMUX_ST_PS_KLV;
 571   } else if (strcmp (mt, "image/x-jpc") == 0) {
 572     /*
 573      * See this document for more details on standard:
 574      *
 575      * https://www.itu.int/rec/T-REC-H.222.0-201206-S/en
 576      *  Annex S describes J2K details
 577      *  Page 104 of this document describes J2k video descriptor
 578      */
 579
 580     const GValue *vProfile = gst_structure_get_value (s, "profile");
 581     const GValue *vMainlevel = gst_structure_get_value (s, "main-level");
 582     const GValue *vFramerate = gst_structure_get_value (s, "framerate");
 583     const GValue *vColorimetry = gst_structure_get_value (s, "colorimetry");
 584     j2k_private_data *private_data;
 585
 586     /* for now, we relax the condition that profile must exist and equal
 587      * GST_JPEG2000_PARSE_PROFILE_BC_SINGLE */
 588     if (vProfile) {
 589       profile = g_value_get_int (vProfile);
 590       if (profile != GST_JPEG2000_PARSE_PROFILE_BC_SINGLE) {
 591         GST_LOG_OBJECT (ts_pad, "Invalid JPEG 2000 profile %d", profile);
 592         /* goto not_negotiated; */
 593       }
 594     }
 595     /* for now, we will relax the condition that the main level must be present */
 596     if (vMainlevel) {
 597       main_level = g_value_get_uint (vMainlevel);
 598       if (main_level > 11) {
 599         GST_ERROR_OBJECT (ts_pad, "Invalid main level %d", main_level);
 600         goto not_negotiated;
 601       }
 602       if (main_level >= 6) {
 603         max_rate = 2 ^ (main_level - 6) * 1600 * 1000000;
 604       } else {
 605         switch (main_level) {
 606           case 0:
 607           case 1:
 608           case 2:
 609           case 3:
 610             max_rate = 200 * 1000000;
 611             break;
 612           case 4:
 613             max_rate = 400 * 1000000;
 614             break;
 615           case 5:
 616             max_rate = 800 * 1000000;
 617             break;
 618           default:
 619             break;
 620         }
 621       }
 622     } else {
 623       /* GST_ERROR_OBJECT (ts_pad, "Missing main level");
 624        * goto not_negotiated; */
 625     }
 626
 627     /* We always mux video in J2K-over-MPEG-TS non-interlaced mode */
 628     private_data = g_new0 (j2k_private_data, 1);
 629     private_data->interlace = FALSE;
 630     private_data->den = 0;
 631     private_data->num = 0;
 632     private_data->max_bitrate = max_rate;
 633     private_data->color_spec = 1;
 634     /* these two fields are not used, since we always mux as non-interlaced */
 635     private_data->Fic = 1;
 636     private_data->Fio = 0;
 637
 638     /* Get Framerate */
 639     if (vFramerate != NULL) {
 640       /* Data for ELSM header */
 641       private_data->num = gst_value_get_fraction_numerator (vFramerate);
 642       private_data->den = gst_value_get_fraction_denominator (vFramerate);
 643     }
 644     /* Get Colorimetry */
 645     if (vColorimetry) {
 646       const char *colorimetry = g_value_get_string (vColorimetry);
 647       color_spec = GST_MPEGTS_JPEG2000_COLORSPEC_SRGB;  /* RGB as default */
 648       if (g_str_equal (colorimetry, GST_VIDEO_COLORIMETRY_BT601)) {
 649         color_spec = GST_MPEGTS_JPEG2000_COLORSPEC_REC601;
 650       } else {
 651         if (g_str_equal (colorimetry, GST_VIDEO_COLORIMETRY_BT709)
 652             || g_str_equal (colorimetry, GST_VIDEO_COLORIMETRY_SMPTE240M)) {
 653           color_spec = GST_MPEGTS_JPEG2000_COLORSPEC_REC709;
 654         }
 655       }
 656       private_data->color_spec = color_spec;
 657     } else {
 658       GST_ERROR_OBJECT (ts_pad, "Colorimetry not present in caps");
 659       g_free (private_data);
 660       goto not_negotiated;
 661     }
 662     st = TSMUX_ST_VIDEO_JP2K;
 663     ts_pad->prepare_func = gst_base_ts_mux_prepare_jpeg2000;
 664     ts_pad->prepare_data = private_data;
 665     ts_pad->free_func = gst_base_ts_mux_free_jpeg2000;
 666   } else {
 667     GstBaseTsMuxClass *klass = GST_BASE_TS_MUX_GET_CLASS (mux);
 668
 669     if (klass->handle_media_type) {
 670       st = klass->handle_media_type (mux, mt, ts_pad);
 671     }
 672   }
 673
 674   if (st == TSMUX_ST_RESERVED) {
 675     GST_ERROR_OBJECT (ts_pad, "Failed to determine stream type");
 676     goto error;
 677   }
 678
 679   if (ts_pad->stream && st != ts_pad->stream->stream_type) {
 680     GST_ELEMENT_ERROR (mux, STREAM, MUX,
 681         ("Stream type change from %02x to %02x not supported",
 682             ts_pad->stream->stream_type, st), NULL);
 683     goto error;
 684   }
 685
 686   if (ts_pad->stream == NULL) {
 687     ts_pad->stream =
 688         tsmux_create_stream (mux->tsmux, st, ts_pad->pid, ts_pad->language);
 689     if (ts_pad->stream == NULL)
 690       goto error;
 691   }
 692
 693   interlace_mode = gst_structure_get_string (s, "interlace-mode");
 694   gst_structure_get_int (s, "rate", &ts_pad->stream->audio_sampling);
 695   gst_structure_get_int (s, "channels", &ts_pad->stream->audio_channels);
 696   gst_structure_get_int (s, "bitrate", &ts_pad->stream->audio_bitrate);
 697
 698   /* frame rate */
 699   gst_structure_get_fraction (s, "framerate", &ts_pad->stream->num,
 700       &ts_pad->stream->den);
 701
 702   /* Interlace mode */
 703   ts_pad->stream->interlace_mode = FALSE;
 704   if (interlace_mode) {
 705     ts_pad->stream->interlace_mode =
 706         g_str_equal (interlace_mode, "interleaved");
 707   }
 708
 709   /* Width and Height */
 710   gst_structure_get_int (s, "width", &ts_pad->stream->horizontal_size);
 711   gst_structure_get_int (s, "height", &ts_pad->stream->vertical_size);
 712
 713   ts_pad->stream->color_spec = color_spec;
 714   ts_pad->stream->max_bitrate = max_rate;
 715   ts_pad->stream->profile_and_level = profile | main_level;
 716
 717   ts_pad->stream->opus_channel_config_code = opus_channel_config_code;
 718
 719   tsmux_stream_set_buffer_release_func (ts_pad->stream, release_buffer_cb);
 720
 721   return GST_FLOW_OK;
 722
 723   /* ERRORS */
 724 not_negotiated:
 725   return GST_FLOW_NOT_NEGOTIATED;
 726
 727 error:
 728   return GST_FLOW_ERROR;
 729 }
 730
 731 static gboolean
 732 is_valid_pmt_pid (guint16 pmt_pid)
 733 {
 734   if (pmt_pid < 0x0010 || pmt_pid > 0x1ffe)
 735     return FALSE;
 736   return TRUE;
 737 }
 738
 739 static GstFlowReturn
 740 gst_base_ts_mux_create_stream (GstBaseTsMux * mux, GstBaseTsMuxPad * ts_pad)
 741 {
 742   GstCaps *caps = gst_pad_get_current_caps (GST_PAD (ts_pad));
 743   GstFlowReturn ret;
 744
 745   if (caps == NULL) {
 746     GST_DEBUG_OBJECT (ts_pad, "Sink pad caps were not set before pushing");
 747     return GST_FLOW_NOT_NEGOTIATED;
 748   }
 749
 750   ret = gst_base_ts_mux_create_or_update_stream (mux, ts_pad, caps);
 751   gst_caps_unref (caps);
 752
 753   if (ret == GST_FLOW_OK) {
 754     tsmux_program_add_stream (ts_pad->prog, ts_pad->stream);
 755   }
 756
 757   return ret;
 758 }
 759
 760 static GstFlowReturn
 761 gst_base_ts_mux_create_pad_stream (GstBaseTsMux * mux, GstPad * pad)
 762 {
 763   GstBaseTsMuxPad *ts_pad = GST_BASE_TS_MUX_PAD (pad);
 764   gchar *name = NULL;
 765   gchar *prop_name;
 766   GstFlowReturn ret = GST_FLOW_OK;
 767
 768   if (ts_pad->prog_id == -1) {
 769     name = GST_PAD_NAME (pad);
 770     if (mux->prog_map != NULL && gst_structure_has_field (mux->prog_map, name)) {
 771       gint idx;
 772       gboolean ret = gst_structure_get_int (mux->prog_map, name, &idx);
 773       if (!ret) {
 774         GST_ELEMENT_ERROR (mux, STREAM, MUX,
 775             ("Reading program map failed. Assuming default"), (NULL));
 776         idx = DEFAULT_PROG_ID;
 777       }
 778       if (idx < 0) {
 779         GST_DEBUG_OBJECT (mux, "Program number %d associate with pad %s less "
 780             "than zero; DEFAULT_PROGRAM = %d is used instead",
 781             idx, name, DEFAULT_PROG_ID);
 782         idx = DEFAULT_PROG_ID;
 783       }
 784       ts_pad->prog_id = idx;
 785     } else {
 786       ts_pad->prog_id = DEFAULT_PROG_ID;
 787     }
 788   }
 789
 790   ts_pad->prog =
 791       (TsMuxProgram *) g_hash_table_lookup (mux->programs,
 792       GINT_TO_POINTER (ts_pad->prog_id));
 793   if (ts_pad->prog == NULL) {
 794     ts_pad->prog = tsmux_program_new (mux->tsmux, ts_pad->prog_id);
 795     if (ts_pad->prog == NULL)
 796       goto no_program;
 797     tsmux_set_pmt_interval (ts_pad->prog, mux->pmt_interval);
 798     tsmux_program_set_scte35_pid (ts_pad->prog, mux->scte35_pid);
 799     tsmux_program_set_scte35_interval (ts_pad->prog, mux->scte35_null_interval);
 800     g_hash_table_insert (mux->programs, GINT_TO_POINTER (ts_pad->prog_id),
 801         ts_pad->prog);
 802
 803     /* Check for user-specified PMT PID */
 804     prop_name = g_strdup_printf ("PMT_%d", ts_pad->prog->pgm_number);
 805     if (mux->prog_map && gst_structure_has_field (mux->prog_map, prop_name)) {
 806       guint pmt_pid;
 807
 808       if (gst_structure_get_uint (mux->prog_map, prop_name, &pmt_pid)) {
 809         if (is_valid_pmt_pid (pmt_pid)) {
 810           GST_DEBUG_OBJECT (mux, "User specified pid=%u as PMT for "
 811               "program (prog_id = %d)", pmt_pid, ts_pad->prog->pgm_number);
 812           tsmux_program_set_pmt_pid (ts_pad->prog, pmt_pid);
 813         } else {
 814           GST_ELEMENT_WARNING (mux, LIBRARY, SETTINGS,
 815               ("User specified PMT pid %u for program %d is not valid.",
 816                   pmt_pid, ts_pad->prog->pgm_number), (NULL));
 817         }
 818       }
 819     }
 820     g_free (prop_name);
 821   }
 822
 823   if (ts_pad->stream == NULL) {
 824     ret = gst_base_ts_mux_create_stream (mux, ts_pad);
 825     if (ret != GST_FLOW_OK)
 826       goto no_stream;
 827   }
 828
 829   if (ts_pad->prog->pcr_stream == NULL) {
 830     /* Take the first stream of the program for the PCR */
 831     GST_DEBUG_OBJECT (ts_pad,
 832         "Use stream (pid=%d) from pad as PCR for program (prog_id = %d)",
 833         ts_pad->pid, ts_pad->prog_id);
 834
 835     tsmux_program_set_pcr_stream (ts_pad->prog, ts_pad->stream);
 836   }
 837
 838   /* Check for user-specified PCR PID */
 839   prop_name = g_strdup_printf ("PCR_%d", ts_pad->prog->pgm_number);
 840   if (mux->prog_map && gst_structure_has_field (mux->prog_map, prop_name)) {
 841     const gchar *sink_name =
 842         gst_structure_get_string (mux->prog_map, prop_name);
 843
 844     if (!g_strcmp0 (name, sink_name)) {
 845       GST_DEBUG_OBJECT (mux, "User specified stream (pid=%d) as PCR for "
 846           "program (prog_id = %d)", ts_pad->pid, ts_pad->prog->pgm_number);
 847       tsmux_program_set_pcr_stream (ts_pad->prog, ts_pad->stream);
 848     }
 849   }
 850   g_free (prop_name);
 851
 852   return ret;
 853
 854   /* ERRORS */
 855 no_program:
 856   {
 857     GST_ELEMENT_ERROR (mux, STREAM, MUX,
 858         ("Could not create new program"), (NULL));
 859     return GST_FLOW_ERROR;
 860   }
 861 no_stream:
 862   {
 863     GST_ELEMENT_ERROR (mux, STREAM, MUX,
 864         ("Could not create handler for stream"), (NULL));
 865     return ret;
 866   }
 867 }
 868
 869 static gboolean
 870 gst_base_ts_mux_create_pad_stream_func (GstElement * element, GstPad * pad,
 871     gpointer user_data)
 872 {
 873   GstFlowReturn *ret = user_data;
 874
 875   *ret = gst_base_ts_mux_create_pad_stream (GST_BASE_TS_MUX (element), pad);
 876
 877   return *ret == GST_FLOW_OK;
 878 }
 879
 880 static GstFlowReturn
 881 gst_base_ts_mux_create_streams (GstBaseTsMux * mux)
 882 {
 883   GstFlowReturn ret = GST_FLOW_OK;
 884
 885   gst_element_foreach_sink_pad (GST_ELEMENT_CAST (mux),
 886       gst_base_ts_mux_create_pad_stream_func, &ret);
 887
 888   return ret;
 889 }
 890
 891 static void
 892 new_packet_common_init (GstBaseTsMux * mux, GstBuffer * buf, guint8 * data,
 893     guint len)
 894 {
 895   /* Packets should be at least 188 bytes, but check anyway */
 896   g_assert (len >= 2 || !data);
 897
 898   if (!mux->streamheader_sent && data) {
 899     guint pid = ((data[1] & 0x1f) << 8) | data[2];
 900     /* if it's a PAT or a PMT */
 901     if (pid == 0x00 || (pid >= TSMUX_START_PMT_PID && pid < TSMUX_START_ES_PID)) {
 902       GstBuffer *hbuf;
 903
 904       if (!buf) {
 905         hbuf = gst_buffer_new_and_alloc (len);
 906         gst_buffer_fill (hbuf, 0, data, len);
 907       } else {
 908         hbuf = gst_buffer_copy (buf);
 909       }
 910       GST_LOG_OBJECT (mux,
 911           "Collecting packet with pid 0x%04x into streamheaders", pid);
 912
 913       g_queue_push_tail (&mux->streamheader, hbuf);
 914     } else if (!g_queue_is_empty (&mux->streamheader)) {
 915       gst_base_ts_mux_set_header_on_caps (mux);
 916       mux->streamheader_sent = TRUE;
 917     }
 918   }
 919
 920   if (buf) {
 921     if (mux->is_header) {
 922       GST_LOG_OBJECT (mux, "marking as header buffer");
 923       GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_HEADER);
 924     }
 925     if (mux->is_delta) {
 926       GST_LOG_OBJECT (mux, "marking as delta unit");
 927       GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT);
 928     } else {
 929       GST_DEBUG_OBJECT (mux, "marking as non-delta unit");
 930       mux->is_delta = TRUE;
 931     }
 932   }
 933 }
 934
 935 static GstFlowReturn
 936 gst_base_ts_mux_push_packets (GstBaseTsMux * mux, gboolean force)
 937 {
 938   GstBufferList *buffer_list;
 939   gint align = mux->alignment;
 940   gint av, packet_size;
 941
 942   packet_size = mux->packet_size;
 943
 944   if (align < 0)
 945     align = mux->automatic_alignment;
 946
 947   av = gst_adapter_available (mux->out_adapter);
 948   GST_LOG_OBJECT (mux, "align %d, av %d", align, av);
 949
 950   if (av == 0)
 951     return GST_FLOW_OK;
 952
 953   /* no alignment, just push all available data */
 954   if (align == 0) {
 955     buffer_list = gst_adapter_take_buffer_list (mux->out_adapter, av);
 956     return gst_aggregator_finish_buffer_list (GST_AGGREGATOR (mux),
 957         buffer_list);
 958   }
 959
 960   align *= packet_size;
 961
 962   if (!force && align > av)
 963     return GST_FLOW_OK;
 964
 965   buffer_list = gst_buffer_list_new_sized ((av / align) + 1);
 966
 967   GST_LOG_OBJECT (mux, "aligning to %d bytes", align);
 968   while (align <= av) {
 969     GstBuffer *buf;
 970     GstClockTime pts;
 971
 972     pts = gst_adapter_prev_pts (mux->out_adapter, NULL);
 973     buf = gst_adapter_take_buffer (mux->out_adapter, align);
 974
 975     GST_BUFFER_PTS (buf) = pts;
 976
 977     gst_buffer_list_add (buffer_list, buf);
 978     av -= align;
 979   }
 980
 981   if (av > 0 && force) {
 982     GstBuffer *buf;
 983     GstClockTime pts;
 984     guint8 *data;
 985     guint32 header;
 986     gint dummy;
 987     GstMapInfo map;
 988
 989     GST_LOG_OBJECT (mux, "handling %d leftover bytes", av);
 990
 991     pts = gst_adapter_prev_pts (mux->out_adapter, NULL);
 992     buf = gst_buffer_new_and_alloc (align);
 993
 994     GST_BUFFER_PTS (buf) = pts;
 995
 996     gst_buffer_map (buf, &map, GST_MAP_READ);
 997     data = map.data;
 998
 999     gst_adapter_copy (mux->out_adapter, data, 0, av);
1000     gst_adapter_clear (mux->out_adapter);
1001
1002     data += av;
1003     header = GST_READ_UINT32_BE (data - packet_size);
1004
1005     dummy = (map.size - av) / packet_size;
1006     GST_LOG_OBJECT (mux, "adding %d null packets", dummy);
1007
1008     for (; dummy > 0; dummy--) {
1009       gint offset;
1010
1011       if (packet_size > GST_BASE_TS_MUX_NORMAL_PACKET_LENGTH) {
1012         GST_WRITE_UINT32_BE (data, header);
1013         /* simply increase header a bit and never mind too much */
1014         header++;
1015         offset = 4;
1016       } else {
1017         offset = 0;
1018       }
1019       GST_WRITE_UINT8 (data + offset, TSMUX_SYNC_BYTE);
1020       /* null packet PID */
1021       GST_WRITE_UINT16_BE (data + offset + 1, 0x1FFF);
1022       /* no adaptation field exists | continuity counter undefined */
1023       GST_WRITE_UINT8 (data + offset + 3, 0x10);
1024       /* payload */
1025       memset (data + offset + 4, 0, GST_BASE_TS_MUX_NORMAL_PACKET_LENGTH - 4);
1026       data += packet_size;
1027     }
1028
1029     gst_buffer_unmap (buf, &map);
1030     gst_buffer_list_add (buffer_list, buf);
1031   }
1032
1033   return gst_aggregator_finish_buffer_list (GST_AGGREGATOR (mux), buffer_list);
1034 }
1035
1036 static GstFlowReturn
1037 gst_base_ts_mux_collect_packet (GstBaseTsMux * mux, GstBuffer * buf)
1038 {
1039   GST_LOG_OBJECT (mux, "collecting packet size %" G_GSIZE_FORMAT,
1040       gst_buffer_get_size (buf));
1041   gst_adapter_push (mux->out_adapter, buf);
1042
1043   return GST_FLOW_OK;
1044 }
1045
1046 static GstEvent *
1047 check_pending_key_unit_event (GstEvent * pending_event, GstSegment * segment,
1048     GstClockTime timestamp, guint flags, GstClockTime pending_key_unit_ts)
1049 {
1050   GstClockTime running_time, stream_time;
1051   gboolean all_headers;
1052   guint count;
1053   GstEvent *event = NULL;
1054
1055   g_assert (segment != NULL);
1056
1057   if (pending_event == NULL)
1058     goto out;
1059
1060   if (GST_CLOCK_TIME_IS_VALID (pending_key_unit_ts) &&
1061       timestamp == GST_CLOCK_TIME_NONE)
1062     goto out;
1063
1064   running_time = timestamp;
1065
1066   GST_INFO ("now %" GST_TIME_FORMAT " wanted %" GST_TIME_FORMAT,
1067       GST_TIME_ARGS (running_time), GST_TIME_ARGS (pending_key_unit_ts));
1068   if (GST_CLOCK_TIME_IS_VALID (pending_key_unit_ts) &&
1069       running_time < pending_key_unit_ts)
1070     goto out;
1071
1072   if (flags & GST_BUFFER_FLAG_DELTA_UNIT) {
1073     GST_INFO ("pending force key unit, waiting for keyframe");
1074     goto out;
1075   }
1076
1077   stream_time = gst_segment_to_stream_time (segment,
1078       GST_FORMAT_TIME, timestamp);
1079
1080   if (GST_EVENT_TYPE (pending_event) == GST_EVENT_CUSTOM_DOWNSTREAM) {
1081     gst_video_event_parse_downstream_force_key_unit (pending_event,
1082         NULL, NULL, NULL, &all_headers, &count);
1083   } else {
1084     gst_video_event_parse_upstream_force_key_unit (pending_event, NULL,
1085         &all_headers, &count);
1086   }
1087
1088   event =
1089       gst_video_event_new_downstream_force_key_unit (timestamp, stream_time,
1090       running_time, all_headers, count);
1091   gst_event_set_seqnum (event, gst_event_get_seqnum (pending_event));
1092
1093 out:
1094   return event;
1095 }
1096
1097 /* Called when the TsMux has prepared a packet for output. Return FALSE
1098  * on error */
1099 static gboolean
1100 new_packet_cb (GstBuffer * buf, void *user_data, gint64 new_pcr)
1101 {
1102   GstBaseTsMux *mux = (GstBaseTsMux *) user_data;
1103   GstAggregator *agg = GST_AGGREGATOR (mux);
1104   GstBaseTsMuxClass *klass = GST_BASE_TS_MUX_GET_CLASS (mux);
1105   GstMapInfo map;
1106   GstSegment *agg_segment = &GST_AGGREGATOR_PAD (agg->srcpad)->segment;
1107
1108   g_assert (klass->output_packet);
1109
1110   gst_buffer_map (buf, &map, GST_MAP_READWRITE);
1111
1112   if (!GST_CLOCK_TIME_IS_VALID (GST_BUFFER_PTS (buf))) {
1113     /* tsmux isn't generating timestamps. Use the input times */
1114     GST_BUFFER_PTS (buf) = mux->last_ts;
1115   }
1116
1117   if (GST_CLOCK_TIME_IS_VALID (GST_BUFFER_PTS (buf))) {
1118     if (!GST_CLOCK_TIME_IS_VALID (mux->output_ts_offset)) {
1119       GstClockTime output_start_time = agg_segment->position;
1120       if (agg_segment->position == -1
1121           || agg_segment->position < agg_segment->start) {
1122         output_start_time = agg_segment->start;
1123       }
1124
1125       mux->output_ts_offset =
1126           GST_CLOCK_DIFF (GST_BUFFER_PTS (buf), output_start_time);
1127
1128       GST_DEBUG_OBJECT (mux, "New output ts offset %" GST_STIME_FORMAT,
1129           GST_STIME_ARGS (mux->output_ts_offset));
1130     }
1131
1132     if (GST_CLOCK_TIME_IS_VALID (mux->output_ts_offset)) {
1133       GST_BUFFER_PTS (buf) += mux->output_ts_offset;
1134     }
1135   }
1136
1137   if (GST_CLOCK_TIME_IS_VALID (GST_BUFFER_PTS (buf))) {
1138     agg_segment->position = GST_BUFFER_PTS (buf);
1139   }
1140
1141   /* do common init (flags and streamheaders) */
1142   new_packet_common_init (mux, buf, map.data, map.size);
1143
1144   gst_buffer_unmap (buf, &map);
1145
1146   return klass->output_packet (mux, buf, new_pcr);
1147 }
1148
1149 /* called when TsMux needs new packet to write into */
1150 static void
1151 alloc_packet_cb (GstBuffer ** buf, void *user_data)
1152 {
1153   GstBaseTsMux *mux = (GstBaseTsMux *) user_data;
1154   GstBaseTsMuxClass *klass = GST_BASE_TS_MUX_GET_CLASS (mux);
1155
1156   g_assert (klass->allocate_packet);
1157
1158   klass->allocate_packet (mux, buf);
1159 }
1160
1161 static GstFlowReturn
1162 gst_base_ts_mux_aggregate_buffer (GstBaseTsMux * mux,
1163     GstAggregatorPad * agg_pad, GstBuffer * buf)
1164 {
1165   GstFlowReturn ret = GST_FLOW_OK;
1166   GstBaseTsMuxPad *best = GST_BASE_TS_MUX_PAD (agg_pad);
1167   TsMuxProgram *prog;
1168   gint64 pts = GST_CLOCK_STIME_NONE;
1169   gint64 dts = GST_CLOCK_STIME_NONE;
1170   gboolean delta = TRUE, header = FALSE;
1171   StreamData *stream_data;
1172   GstMpegtsSection *scte_section = NULL;
1173
1174   GST_DEBUG_OBJECT (mux, "Pads collected");
1175
1176   if (buf && gst_buffer_get_size (buf) == 0
1177       && GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_GAP)) {
1178     gst_buffer_unref (buf);
1179     return GST_FLOW_OK;
1180   }
1181
1182   if (G_UNLIKELY (mux->first)) {
1183     ret = gst_base_ts_mux_create_streams (mux);
1184     if (G_UNLIKELY (ret != GST_FLOW_OK)) {
1185       if (buf)
1186         gst_buffer_unref (buf);
1187       return ret;
1188     }
1189
1190     mux->first = FALSE;
1191   }
1192
1193   prog = best->prog;
1194   if (prog == NULL) {
1195     GList *cur;
1196
1197     gst_base_ts_mux_create_pad_stream (mux, GST_PAD (best));
1198     tsmux_resend_pat (mux->tsmux);
1199     tsmux_resend_si (mux->tsmux);
1200     prog = best->prog;
1201     g_assert_nonnull (prog);
1202
1203     /* output PMT for each program */
1204     for (cur = mux->tsmux->programs; cur; cur = cur->next) {
1205       TsMuxProgram *program = (TsMuxProgram *) cur->data;
1206
1207       tsmux_resend_pmt (program);
1208     }
1209   }
1210
1211   g_assert (buf != NULL);
1212
1213   if (best->prepare_func) {
1214     GstBuffer *tmp;
1215
1216     tmp = best->prepare_func (buf, best, mux);
1217     g_assert (tmp);
1218     gst_buffer_unref (buf);
1219     buf = tmp;
1220   }
1221
1222   if (mux->force_key_unit_event != NULL && best->stream->is_video_stream) {
1223     GstEvent *event;
1224
1225     event = check_pending_key_unit_event (mux->force_key_unit_event,
1226         &agg_pad->segment, GST_BUFFER_PTS (buf),
1227         GST_BUFFER_FLAGS (buf), mux->pending_key_unit_ts);
1228     if (event) {
1229       GstClockTime running_time;
1230       guint count;
1231       GList *cur;
1232
1233       mux->pending_key_unit_ts = GST_CLOCK_TIME_NONE;
1234       gst_event_replace (&mux->force_key_unit_event, NULL);
1235
1236       gst_video_event_parse_downstream_force_key_unit (event,
1237           NULL, NULL, &running_time, NULL, &count);
1238
1239       GST_INFO_OBJECT (mux, "pushing downstream force-key-unit event %d "
1240           "%" GST_TIME_FORMAT " count %d", gst_event_get_seqnum (event),
1241           GST_TIME_ARGS (running_time), count);
1242       gst_pad_push_event (GST_AGGREGATOR_SRC_PAD (mux), event);
1243
1244       /* output PAT, SI tables */
1245       tsmux_resend_pat (mux->tsmux);
1246       tsmux_resend_si (mux->tsmux);
1247
1248       /* output PMT for each program */
1249       for (cur = mux->tsmux->programs; cur; cur = cur->next) {
1250         TsMuxProgram *program = (TsMuxProgram *) cur->data;
1251
1252         tsmux_resend_pmt (program);
1253       }
1254     }
1255   }
1256
1257   if (G_UNLIKELY (prog->pcr_stream == NULL)) {
1258     /* Take the first data stream for the PCR */
1259     GST_DEBUG_OBJECT (best,
1260         "Use stream (pid=%d) from pad as PCR for program (prog_id = %d)",
1261         best->pid, best->prog_id);
1262
1263     /* Set the chosen PCR stream */
1264     tsmux_program_set_pcr_stream (prog, best->stream);
1265   }
1266
1267   GST_DEBUG_OBJECT (best, "Chose stream for output (PID: 0x%04x)", best->pid);
1268
1269   GST_OBJECT_LOCK (mux);
1270   scte_section = mux->pending_scte35_section;
1271   mux->pending_scte35_section = NULL;
1272   GST_OBJECT_UNLOCK (mux);
1273   if (G_UNLIKELY (scte_section)) {
1274     GST_DEBUG_OBJECT (mux, "Sending pending SCTE section");
1275     if (!tsmux_send_section (mux->tsmux, scte_section))
1276       GST_ERROR_OBJECT (mux, "Error sending SCTE section !");
1277   }
1278
1279   if (GST_CLOCK_TIME_IS_VALID (GST_BUFFER_PTS (buf))) {
1280     pts = GSTTIME_TO_MPEGTIME (GST_BUFFER_PTS (buf));
1281     GST_DEBUG_OBJECT (mux, "Buffer has PTS  %" GST_TIME_FORMAT " pts %"
1282         G_GINT64_FORMAT "%s", GST_TIME_ARGS (GST_BUFFER_PTS (buf)), pts,
1283         !GST_BUFFER_FLAG_IS_SET (buf,
1284             GST_BUFFER_FLAG_DELTA_UNIT) ? " (keyframe)" : "");
1285   }
1286
1287   if (GST_CLOCK_STIME_IS_VALID (best->dts)) {
1288     dts = GSTTIME_TO_MPEGTIME (best->dts);
1289     GST_DEBUG_OBJECT (mux, "Buffer has DTS %" GST_STIME_FORMAT " dts %"
1290         G_GINT64_FORMAT, GST_STIME_ARGS (best->dts), dts);
1291   }
1292
1293   /* should not have a DTS without PTS */
1294   if (!GST_CLOCK_STIME_IS_VALID (pts) && GST_CLOCK_STIME_IS_VALID (dts)) {
1295     GST_DEBUG_OBJECT (mux, "using DTS for unknown PTS");
1296     pts = dts;
1297   }
1298
1299   if (best->stream->is_video_stream) {
1300     delta = GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_DELTA_UNIT);
1301     header = GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_HEADER);
1302   }
1303
1304   if (best->stream->is_meta && gst_buffer_get_size (buf) > (G_MAXUINT16 - 3)) {
1305     GST_WARNING_OBJECT (mux, "KLV meta unit too big, splitting not supported");
1306
1307     gst_buffer_unref (buf);
1308     return GST_FLOW_OK;
1309   }
1310
1311   GST_DEBUG_OBJECT (mux, "delta: %d", delta);
1312
1313   stream_data = stream_data_new (buf);
1314   tsmux_stream_add_data (best->stream, stream_data->map_info.data,
1315       stream_data->map_info.size, stream_data, pts, dts, !delta);
1316
1317   /* outgoing ts follows ts of PCR program stream */
1318   if (prog->pcr_stream == best->stream) {
1319     /* prefer DTS if present for PCR as it should be monotone */
1320     mux->last_ts =
1321         GST_CLOCK_TIME_IS_VALID (GST_BUFFER_DTS (buf)) ?
1322         GST_BUFFER_DTS (buf) : GST_BUFFER_PTS (buf);
1323   }
1324
1325   mux->is_delta = delta;
1326   mux->is_header = header;
1327   while (tsmux_stream_bytes_in_buffer (best->stream) > 0) {
1328     if (!tsmux_write_stream_packet (mux->tsmux, best->stream)) {
1329       /* Failed writing data for some reason. Set appropriate error */
1330       GST_DEBUG_OBJECT (mux, "Failed to write data packet");
1331       GST_ELEMENT_ERROR (mux, STREAM, MUX,
1332           ("Failed writing output data to stream %04x", best->stream->id),
1333           (NULL));
1334       goto write_fail;
1335     }
1336   }
1337   /* flush packet cache */
1338   return gst_base_ts_mux_push_packets (mux, FALSE);
1339
1340   /* ERRORS */
1341 write_fail:
1342   {
1343     return mux->last_flow_ret;
1344   }
1345 }
1346
1347 /* GstElement implementation */
1348 static gboolean
1349 gst_base_ts_mux_has_pad_with_pid (GstBaseTsMux * mux, guint16 pid)
1350 {
1351   GList *l;
1352   gboolean res = FALSE;
1353
1354   GST_OBJECT_LOCK (mux);
1355
1356   for (l = GST_ELEMENT_CAST (mux)->sinkpads; l; l = l->next) {
1357     GstBaseTsMuxPad *tpad = GST_BASE_TS_MUX_PAD (l->data);
1358
1359     if (tpad->pid == pid) {
1360       res = TRUE;
1361       break;
1362     }
1363   }
1364
1365   GST_OBJECT_UNLOCK (mux);
1366   return res;
1367 }
1368
1369 static GstPad *
1370 gst_base_ts_mux_request_new_pad (GstElement * element, GstPadTemplate * templ,
1371     const gchar * name, const GstCaps * caps)
1372 {
1373   GstBaseTsMux *mux = GST_BASE_TS_MUX (element);
1374   gint pid = -1;
1375   GstPad *pad = NULL;
1376   gchar *free_name = NULL;
1377
1378   if (name != NULL && sscanf (name, "sink_%d", &pid) == 1) {
1379     if (tsmux_find_stream (mux->tsmux, pid))
1380       goto stream_exists;
1381     /* Make sure we don't use reserved PID.
1382      * FIXME : This should be extended to other variants (ex: ATSC) reserved PID */
1383     if (pid < TSMUX_START_ES_PID)
1384       goto invalid_stream_pid;
1385   } else {
1386     do {
1387       pid = tsmux_get_new_pid (mux->tsmux);
1388     } while (gst_base_ts_mux_has_pad_with_pid (mux, pid));
1389
1390     /* Name the pad correctly after the selected pid */
1391     name = free_name = g_strdup_printf ("sink_%d", pid);
1392   }
1393
1394   pad = (GstPad *)
1395       GST_ELEMENT_CLASS (parent_class)->request_new_pad (element,
1396       templ, name, caps);
1397
1398   gst_base_ts_mux_pad_reset (GST_BASE_TS_MUX_PAD (pad));
1399   GST_BASE_TS_MUX_PAD (pad)->pid = pid;
1400
1401   g_free (free_name);
1402
1403   return pad;
1404
1405   /* ERRORS */
1406 stream_exists:
1407   {
1408     GST_ELEMENT_ERROR (element, STREAM, MUX, ("Duplicate PID requested"),
1409         (NULL));
1410     return NULL;
1411   }
1412
1413 invalid_stream_pid:
1414   {
1415     GST_ELEMENT_ERROR (element, STREAM, MUX,
1416         ("Invalid Elementary stream PID (0x%02u < 0x40)", pid), (NULL));
1417     return NULL;
1418   }
1419 }
1420
1421 static void
1422 gst_base_ts_mux_release_pad (GstElement * element, GstPad * pad)
1423 {
1424   GstBaseTsMux *mux = GST_BASE_TS_MUX (element);
1425
1426   if (mux->tsmux) {
1427     GList *cur;
1428     GstBaseTsMuxPad *ts_pad = GST_BASE_TS_MUX_PAD (pad);
1429     gint pid = ts_pad->pid;
1430
1431     if (ts_pad->prog) {
1432       if (ts_pad->prog->pcr_stream == ts_pad->stream) {
1433         tsmux_program_set_pcr_stream (ts_pad->prog, NULL);
1434       }
1435       if (tsmux_remove_stream (mux->tsmux, pid, ts_pad->prog)) {
1436         g_hash_table_remove (mux->programs, GINT_TO_POINTER (ts_pad->prog_id));
1437       }
1438     }
1439
1440     tsmux_resend_pat (mux->tsmux);
1441     tsmux_resend_si (mux->tsmux);
1442
1443     /* output PMT for each program */
1444     for (cur = mux->tsmux->programs; cur; cur = cur->next) {
1445       TsMuxProgram *program = (TsMuxProgram *) cur->data;
1446
1447       tsmux_resend_pmt (program);
1448     }
1449   }
1450
1451   GST_ELEMENT_CLASS (parent_class)->release_pad (element, pad);
1452 }
1453
1454 /* GstAggregator implementation */
1455
1456 static void
1457 request_keyframe (GstBaseTsMux * mux, GstClockTime running_time)
1458 {
1459   GList *l;
1460   GST_OBJECT_LOCK (mux);
1461
1462   for (l = GST_ELEMENT_CAST (mux)->sinkpads; l; l = l->next) {
1463     gst_pad_push_event (GST_PAD (l->data),
1464         gst_video_event_new_upstream_force_key_unit (running_time, TRUE, 0));
1465   }
1466
1467   GST_OBJECT_UNLOCK (mux);
1468 }
1469
1470 static const guint32 crc_tab[256] = {
1471   0x00000000, 0x04c11db7, 0x09823b6e, 0x0d4326d9, 0x130476dc, 0x17c56b6b,
1472   0x1a864db2, 0x1e475005, 0x2608edb8, 0x22c9f00f, 0x2f8ad6d6, 0x2b4bcb61,
1473   0x350c9b64, 0x31cd86d3, 0x3c8ea00a, 0x384fbdbd, 0x4c11db70, 0x48d0c6c7,
1474   0x4593e01e, 0x4152fda9, 0x5f15adac, 0x5bd4b01b, 0x569796c2, 0x52568b75,
1475   0x6a1936c8, 0x6ed82b7f, 0x639b0da6, 0x675a1011, 0x791d4014, 0x7ddc5da3,
1476   0x709f7b7a, 0x745e66cd, 0x9823b6e0, 0x9ce2ab57, 0x91a18d8e, 0x95609039,
1477   0x8b27c03c, 0x8fe6dd8b, 0x82a5fb52, 0x8664e6e5, 0xbe2b5b58, 0xbaea46ef,
1478   0xb7a96036, 0xb3687d81, 0xad2f2d84, 0xa9ee3033, 0xa4ad16ea, 0xa06c0b5d,
1479   0xd4326d90, 0xd0f37027, 0xddb056fe, 0xd9714b49, 0xc7361b4c, 0xc3f706fb,
1480   0xceb42022, 0xca753d95, 0xf23a8028, 0xf6fb9d9f, 0xfbb8bb46, 0xff79a6f1,
1481   0xe13ef6f4, 0xe5ffeb43, 0xe8bccd9a, 0xec7dd02d, 0x34867077, 0x30476dc0,
1482   0x3d044b19, 0x39c556ae, 0x278206ab, 0x23431b1c, 0x2e003dc5, 0x2ac12072,
1483   0x128e9dcf, 0x164f8078, 0x1b0ca6a1, 0x1fcdbb16, 0x018aeb13, 0x054bf6a4,
1484   0x0808d07d, 0x0cc9cdca, 0x7897ab07, 0x7c56b6b0, 0x71159069, 0x75d48dde,
1485   0x6b93dddb, 0x6f52c06c, 0x6211e6b5, 0x66d0fb02, 0x5e9f46bf, 0x5a5e5b08,
1486   0x571d7dd1, 0x53dc6066, 0x4d9b3063, 0x495a2dd4, 0x44190b0d, 0x40d816ba,
1487   0xaca5c697, 0xa864db20, 0xa527fdf9, 0xa1e6e04e, 0xbfa1b04b, 0xbb60adfc,
1488   0xb6238b25, 0xb2e29692, 0x8aad2b2f, 0x8e6c3698, 0x832f1041, 0x87ee0df6,
1489   0x99a95df3, 0x9d684044, 0x902b669d, 0x94ea7b2a, 0xe0b41de7, 0xe4750050,
1490   0xe9362689, 0xedf73b3e, 0xf3b06b3b, 0xf771768c, 0xfa325055, 0xfef34de2,
1491   0xc6bcf05f, 0xc27dede8, 0xcf3ecb31, 0xcbffd686, 0xd5b88683, 0xd1799b34,
1492   0xdc3abded, 0xd8fba05a, 0x690ce0ee, 0x6dcdfd59, 0x608edb80, 0x644fc637,
1493   0x7a089632, 0x7ec98b85, 0x738aad5c, 0x774bb0eb, 0x4f040d56, 0x4bc510e1,
1494   0x46863638, 0x42472b8f, 0x5c007b8a, 0x58c1663d, 0x558240e4, 0x51435d53,
1495   0x251d3b9e, 0x21dc2629, 0x2c9f00f0, 0x285e1d47, 0x36194d42, 0x32d850f5,
1496   0x3f9b762c, 0x3b5a6b9b, 0x0315d626, 0x07d4cb91, 0x0a97ed48, 0x0e56f0ff,
1497   0x1011a0fa, 0x14d0bd4d, 0x19939b94, 0x1d528623, 0xf12f560e, 0xf5ee4bb9,
1498   0xf8ad6d60, 0xfc6c70d7, 0xe22b20d2, 0xe6ea3d65, 0xeba91bbc, 0xef68060b,
1499   0xd727bbb6, 0xd3e6a601, 0xdea580d8, 0xda649d6f, 0xc423cd6a, 0xc0e2d0dd,
1500   0xcda1f604, 0xc960ebb3, 0xbd3e8d7e, 0xb9ff90c9, 0xb4bcb610, 0xb07daba7,
1501   0xae3afba2, 0xaafbe615, 0xa7b8c0cc, 0xa379dd7b, 0x9b3660c6, 0x9ff77d71,
1502   0x92b45ba8, 0x9675461f, 0x8832161a, 0x8cf30bad, 0x81b02d74, 0x857130c3,
1503   0x5d8a9099, 0x594b8d2e, 0x5408abf7, 0x50c9b640, 0x4e8ee645, 0x4a4ffbf2,
1504   0x470cdd2b, 0x43cdc09c, 0x7b827d21, 0x7f436096, 0x7200464f, 0x76c15bf8,
1505   0x68860bfd, 0x6c47164a, 0x61043093, 0x65c52d24, 0x119b4be9, 0x155a565e,
1506   0x18197087, 0x1cd86d30, 0x029f3d35, 0x065e2082, 0x0b1d065b, 0x0fdc1bec,
1507   0x3793a651, 0x3352bbe6, 0x3e119d3f, 0x3ad08088, 0x2497d08d, 0x2056cd3a,
1508   0x2d15ebe3, 0x29d4f654, 0xc5a92679, 0xc1683bce, 0xcc2b1d17, 0xc8ea00a0,
1509   0xd6ad50a5, 0xd26c4d12, 0xdf2f6bcb, 0xdbee767c, 0xe3a1cbc1, 0xe760d676,
1510   0xea23f0af, 0xeee2ed18, 0xf0a5bd1d, 0xf464a0aa, 0xf9278673, 0xfde69bc4,
1511   0x89b8fd09, 0x8d79e0be, 0x803ac667, 0x84fbdbd0, 0x9abc8bd5, 0x9e7d9662,
1512   0x933eb0bb, 0x97ffad0c, 0xafb010b1, 0xab710d06, 0xa6322bdf, 0xa2f33668,
1513   0xbcb4666d, 0xb8757bda, 0xb5365d03, 0xb1f740b4
1514 };
1515
1516 static guint32
1517 _calc_crc32 (const guint8 * data, guint datalen)
1518 {
1519   gint i;
1520   guint32 crc = 0xffffffff;
1521
1522   for (i = 0; i < datalen; i++) {
1523     crc = (crc << 8) ^ crc_tab[((crc >> 24) ^ *data++) & 0xff];
1524   }
1525   return crc;
1526 }
1527
1528 #define MPEGTIME_TO_GSTTIME(t) ((t) * (guint64)100000 / 9)
1529
1530 static GstMpegtsSCTESpliceEvent *
1531 copy_splice (GstMpegtsSCTESpliceEvent * splice)
1532 {
1533   return g_boxed_copy (GST_TYPE_MPEGTS_SCTE_SPLICE_EVENT, splice);
1534 }
1535
1536 static void
1537 free_splice (GstMpegtsSCTESpliceEvent * splice)
1538 {
1539   g_boxed_free (GST_TYPE_MPEGTS_SCTE_SPLICE_EVENT, splice);
1540 }
1541
1542 /* FIXME: get rid of this when depending on glib >= 2.62 */
1543
1544 static GPtrArray *
1545 _g_ptr_array_copy (GPtrArray * array,
1546     GCopyFunc func, GFreeFunc free_func, gpointer user_data)
1547 {
1548   GPtrArray *new_array;
1549
1550   g_return_val_if_fail (array != NULL, NULL);
1551
1552   new_array = g_ptr_array_new_with_free_func (free_func);
1553
1554   g_ptr_array_set_size (new_array, array->len);
1555
1556   if (func != NULL) {
1557     guint i;
1558
1559     for (i = 0; i < array->len; i++)
1560       new_array->pdata[i] = func (array->pdata[i], user_data);
1561   } else if (array->len > 0) {
1562     memcpy (new_array->pdata, array->pdata,
1563         array->len * sizeof (*array->pdata));
1564   }
1565
1566   new_array->len = array->len;
1567
1568   return new_array;
1569 }
1570
1571 static GstMpegtsSCTESIT *
1572 deep_copy_sit (const GstMpegtsSCTESIT * sit)
1573 {
1574   GstMpegtsSCTESIT *sit_copy = g_boxed_copy (GST_TYPE_MPEGTS_SCTE_SIT, sit);
1575   GPtrArray *splices_copy =
1576       _g_ptr_array_copy (sit_copy->splices, (GCopyFunc) copy_splice,
1577       (GFreeFunc) free_splice, NULL);
1578
1579   g_ptr_array_unref (sit_copy->splices);
1580   sit_copy->splices = splices_copy;
1581
1582   return sit_copy;
1583 }
1584
1585 /* Takes ownership of @section.
1586  *
1587  * This function is a bit complex because the SCTE sections can
1588  * have various origins:
1589  *
1590  * * Sections created by the application with the gst_mpegts_scte_*_new()
1591  *   API. The splice times / durations contained by these are expressed
1592  *   in the GStreamer running time domain, and must be translated to
1593  *   our local PES time domain. In this case, we will packetize the section
1594  *   ourselves.
1595  *
1596  * * Sections passed through from tsdemux: this case is complicated as
1597  *   splice times in the incoming stream may be encrypted, with pts_adjustment
1598  *   being the only timing field guaranteed *not* to be encrypted. In this
1599  *   case, the original binary data (section->data) will be reinjected as is
1600  *   in the output stream, with pts_adjustment adjusted. tsdemux provides us
1601  *   with the pts_offset it introduces, the difference between the original
1602  *   PES PTSs and the running times it outputs.
1603  *
1604  * Additionally, in either of these cases when the splice times aren't encrypted
1605  * we want to make use of those to request keyframes. For the passthrough case,
1606  * as the splice times are left untouched tsdemux provides us with the running
1607  * times the section originally referred to. We cannot calculate it locally
1608  * because we would need to have access to the information that the timestamps
1609  * in the original PES domain have wrapped around, and how many times they have
1610  * done so. While we could probably make educated guesses, tsdemux (more specifically
1611  * mpegtspacketizer) already keeps track of that, and it seemed more logical to
1612  * perform the calculation there and forward it alongside the downstream events.
1613  *
1614  * Finally, while we can't request keyframes at splice points in the encrypted
1615  * case, if the input stream was compliant in that regard and no reencoding took
1616  * place the splice times will still match with valid splice points, it is up
1617  * to the application to ensure that that is the case.
1618  */
1619 static void
1620 handle_scte35_section (GstBaseTsMux * mux, GstEvent * event,
1621     GstMpegtsSection * section, guint64 mpeg_pts_offset,
1622     GstStructure * rtime_map)
1623 {
1624   GstMpegtsSCTESIT *sit;
1625   guint i;
1626   gboolean forward = TRUE;
1627   guint64 pts_adjust;
1628   guint8 *section_data;
1629   guint8 *crc;
1630   gboolean translate = FALSE;
1631
1632   sit = (GstMpegtsSCTESIT *) gst_mpegts_section_get_scte_sit (section);
1633
1634   /* When the application injects manually constructed splice events,
1635    * their time domain is the GStreamer running time, we receive them
1636    * unpacketized and translate the fields in the SIT to local PTS.
1637    *
1638    * We make a copy of the SIT in order to make sure we can rewrite it.
1639    */
1640   if (sit->is_running_time) {
1641     sit = deep_copy_sit (sit);
1642     translate = TRUE;
1643   }
1644
1645   switch (sit->splice_command_type) {
1646     case GST_MTS_SCTE_SPLICE_COMMAND_NULL:
1647       /* We implement heartbeating ourselves */
1648       forward = FALSE;
1649       break;
1650     case GST_MTS_SCTE_SPLICE_COMMAND_SCHEDULE:
1651       /* No need to request keyframes at this point, splice_insert
1652        * messages will precede the future splice points and we
1653        * can request keyframes then. Only translate if needed.
1654        */
1655       if (translate) {
1656         for (i = 0; i < sit->splices->len; i++) {
1657           GstMpegtsSCTESpliceEvent *sevent =
1658               g_ptr_array_index (sit->splices, i);
1659
1660           if (sevent->program_splice_time_specified)
1661             sevent->program_splice_time =
1662                 GSTTIME_TO_MPEGTIME (sevent->program_splice_time) +
1663                 TS_MUX_CLOCK_BASE;
1664
1665           if (sevent->duration_flag)
1666             sevent->break_duration =
1667                 GSTTIME_TO_MPEGTIME (sevent->break_duration);
1668         }
1669       }
1670       break;
1671     case GST_MTS_SCTE_SPLICE_COMMAND_INSERT:
1672       /* We want keyframes at splice points */
1673       if (sit->fully_parsed && (rtime_map || translate)) {
1674
1675         for (i = 0; i < sit->splices->len; i++) {
1676           guint64 running_time = GST_CLOCK_TIME_NONE;
1677
1678           GstMpegtsSCTESpliceEvent *sevent =
1679               g_ptr_array_index (sit->splices, i);
1680           if (sevent->program_splice_time_specified) {
1681             if (rtime_map) {
1682               gchar *field_name = g_strdup_printf ("event-%u-splice-time",
1683                   sevent->splice_event_id);
1684               if (gst_structure_get_uint64 (rtime_map, field_name,
1685                       &running_time)) {
1686                 GST_DEBUG_OBJECT (mux,
1687                     "Requesting keyframe for splice point at %" GST_TIME_FORMAT,
1688                     GST_TIME_ARGS (running_time));
1689                 request_keyframe (mux, running_time);
1690               }
1691               g_free (field_name);
1692             } else {
1693               g_assert (translate == TRUE);
1694               running_time = sevent->program_splice_time;
1695               GST_DEBUG_OBJECT (mux,
1696                   "Requesting keyframe for splice point at %" GST_TIME_FORMAT,
1697                   GST_TIME_ARGS (running_time));
1698               request_keyframe (mux, running_time);
1699               sevent->program_splice_time =
1700                   GSTTIME_TO_MPEGTIME (running_time) + TS_MUX_CLOCK_BASE;
1701             }
1702           } else {
1703             GST_DEBUG_OBJECT (mux,
1704                 "Requesting keyframe for immediate splice point");
1705             request_keyframe (mux, GST_CLOCK_TIME_NONE);
1706           }
1707
1708           if (sevent->duration_flag) {
1709             if (translate) {
1710               sevent->break_duration =
1711                   GSTTIME_TO_MPEGTIME (sevent->break_duration);
1712             }
1713
1714             /* Even if auto_return is FALSE, when a break_duration is specified it
1715              * is intended as a redundancy mechanism in case the follow-up
1716              * splice insert goes missing.
1717              *
1718              * Schedule a keyframe at that point (if we can calculate its position
1719              * accurately).
1720              */
1721             if (GST_CLOCK_TIME_IS_VALID (running_time)) {
1722               running_time += MPEGTIME_TO_GSTTIME (sevent->break_duration);
1723               GST_DEBUG_OBJECT (mux,
1724                   "Requesting keyframe for end of break at %" GST_TIME_FORMAT,
1725                   GST_TIME_ARGS (running_time));
1726               request_keyframe (mux, running_time);
1727             }
1728           }
1729         }
1730       }
1731       break;
1732     case GST_MTS_SCTE_SPLICE_COMMAND_TIME:{
1733       /* Adjust timestamps and potentially request keyframes */
1734       gboolean do_request_keyframes = FALSE;
1735
1736       /* TODO: we can probably be a little more fine-tuned about determining
1737        * whether a keyframe is actually needed, but this at least takes care
1738        * of the requirement in 10.3.4 that a keyframe should not be created
1739        * when the signal contains only a time_descriptor.
1740        */
1741       if (sit->fully_parsed && (rtime_map || translate)) {
1742         for (i = 0; i < sit->descriptors->len; i++) {
1743           GstMpegtsDescriptor *descriptor =
1744               g_ptr_array_index (sit->descriptors, i);
1745
1746           switch (descriptor->tag) {
1747             case GST_MTS_SCTE_DESC_AVAIL:
1748             case GST_MTS_SCTE_DESC_DTMF:
1749             case GST_MTS_SCTE_DESC_SEGMENTATION:
1750               do_request_keyframes = TRUE;
1751               break;
1752             case GST_MTS_SCTE_DESC_TIME:
1753             case GST_MTS_SCTE_DESC_AUDIO:
1754               break;
1755           }
1756
1757           if (do_request_keyframes)
1758             break;
1759         }
1760
1761         if (sit->splice_time_specified) {
1762           GstClockTime running_time = GST_CLOCK_TIME_NONE;
1763
1764           if (rtime_map) {
1765             if (do_request_keyframes
1766                 && gst_structure_get_uint64 (rtime_map, "splice-time",
1767                     &running_time)) {
1768               GST_DEBUG_OBJECT (mux,
1769                   "Requesting keyframe for time signal at %" GST_TIME_FORMAT,
1770                   GST_TIME_ARGS (running_time));
1771               request_keyframe (mux, running_time);
1772             }
1773           } else {
1774             g_assert (translate);
1775             running_time = sit->splice_time;
1776             sit->splice_time =
1777                 GSTTIME_TO_MPEGTIME (running_time) + TS_MUX_CLOCK_BASE;
1778             if (do_request_keyframes) {
1779               GST_DEBUG_OBJECT (mux,
1780                   "Requesting keyframe for time signal at %" GST_TIME_FORMAT,
1781                   GST_TIME_ARGS (running_time));
1782               request_keyframe (mux, running_time);
1783             }
1784           }
1785         } else if (do_request_keyframes) {
1786           GST_DEBUG_OBJECT (mux,
1787               "Requesting keyframe for immediate time signal");
1788           request_keyframe (mux, GST_CLOCK_TIME_NONE);
1789         }
1790       }
1791       break;
1792     }
1793     case GST_MTS_SCTE_SPLICE_COMMAND_BANDWIDTH:
1794     case GST_MTS_SCTE_SPLICE_COMMAND_PRIVATE:
1795       /* Just let those go through untouched, none of our business */
1796       break;
1797     default:
1798       break;
1799   }
1800
1801   if (!forward) {
1802     gst_mpegts_section_unref (section);
1803     return;
1804   }
1805
1806   if (!translate) {
1807     g_assert (section->data);
1808     /* Calculate the final adjustment, as a sum of:
1809      * - The adjustment in the original packet
1810      * - The offset introduced between the original local PTS
1811      *   and the GStreamer PTS output by tsdemux
1812      * - Our own 1-hour offset
1813      */
1814     pts_adjust = sit->pts_adjustment + mpeg_pts_offset + TS_MUX_CLOCK_BASE;
1815
1816     /* Account for offsets potentially introduced between the demuxer and us */
1817     pts_adjust +=
1818         GSTTIME_TO_MPEGTIME (gst_event_get_running_time_offset (event));
1819
1820     pts_adjust &= 0x1ffffffff;
1821     section_data = g_memdup (section->data, section->section_length);
1822     section_data[4] |= pts_adjust >> 32;
1823     section_data[5] = pts_adjust >> 24;
1824     section_data[6] = pts_adjust >> 16;
1825     section_data[7] = pts_adjust >> 8;
1826     section_data[8] = pts_adjust;
1827
1828     /* Now rewrite our checksum */
1829     crc = section_data + section->section_length - 4;
1830     GST_WRITE_UINT32_BE (crc, _calc_crc32 (section_data, crc - section_data));
1831
1832     GST_OBJECT_LOCK (mux);
1833     GST_DEBUG_OBJECT (mux, "Storing SCTE section");
1834     if (mux->pending_scte35_section)
1835       gst_mpegts_section_unref (mux->pending_scte35_section);
1836     mux->pending_scte35_section =
1837         gst_mpegts_section_new (mux->scte35_pid, section_data,
1838         section->section_length);
1839     GST_OBJECT_UNLOCK (mux);
1840
1841     gst_mpegts_section_unref (section);
1842   } else {
1843     GST_OBJECT_LOCK (mux);
1844     GST_DEBUG_OBJECT (mux, "Storing SCTE section");
1845     gst_mpegts_section_unref (section);
1846     if (mux->pending_scte35_section)
1847       gst_mpegts_section_unref (mux->pending_scte35_section);
1848     mux->pending_scte35_section =
1849         gst_mpegts_section_from_scte_sit (sit, mux->scte35_pid);;
1850     GST_OBJECT_UNLOCK (mux);
1851   }
1852 }
1853
1854 static gboolean
1855 gst_base_ts_mux_send_event (GstElement * element, GstEvent * event)
1856 {
1857   GstMpegtsSection *section;
1858   GstBaseTsMux *mux = GST_BASE_TS_MUX (element);
1859
1860   section = gst_event_parse_mpegts_section (event);
1861
1862   if (section) {
1863     GST_DEBUG ("Received event with mpegts section");
1864
1865     if (section->section_type == GST_MPEGTS_SECTION_SCTE_SIT) {
1866       handle_scte35_section (mux, event, section, 0, NULL);
1867     } else {
1868       /* TODO: Check that the section type is supported */
1869       tsmux_add_mpegts_si_section (mux->tsmux, section);
1870     }
1871
1872     gst_event_unref (event);
1873
1874     return TRUE;
1875   }
1876
1877   return GST_ELEMENT_CLASS (parent_class)->send_event (element, event);
1878 }
1879
1880 /* GstAggregator implementation */
1881
1882 static gboolean
1883 gst_base_ts_mux_sink_event (GstAggregator * agg, GstAggregatorPad * agg_pad,
1884     GstEvent * event)
1885 {
1886   GstAggregatorClass *agg_class = GST_AGGREGATOR_CLASS (parent_class);
1887   GstBaseTsMux *mux = GST_BASE_TS_MUX (agg);
1888   GstBaseTsMuxPad *ts_pad = GST_BASE_TS_MUX_PAD (agg_pad);
1889   gboolean res = FALSE;
1890   gboolean forward = TRUE;
1891
1892   switch (GST_EVENT_TYPE (event)) {
1893     case GST_EVENT_CAPS:
1894     {
1895       GstCaps *caps;
1896       GstFlowReturn ret;
1897       GList *cur;
1898
1899       if (ts_pad->stream == NULL)
1900         break;
1901
1902       forward = FALSE;
1903
1904       gst_event_parse_caps (event, &caps);
1905       if (!caps || !gst_caps_is_fixed (caps))
1906         break;
1907
1908       ret = gst_base_ts_mux_create_or_update_stream (mux, ts_pad, caps);
1909       if (ret != GST_FLOW_OK)
1910         break;
1911
1912       mux->tsmux->pat_changed = TRUE;
1913       mux->tsmux->si_changed = TRUE;
1914       tsmux_resend_pat (mux->tsmux);
1915       tsmux_resend_si (mux->tsmux);
1916
1917       /* output PMT for each program */
1918       for (cur = mux->tsmux->programs; cur; cur = cur->next) {
1919         TsMuxProgram *program = (TsMuxProgram *) cur->data;
1920
1921         program->pmt_changed = TRUE;
1922         tsmux_resend_pmt (program);
1923       }
1924
1925       res = TRUE;
1926       break;
1927     }
1928     case GST_EVENT_CUSTOM_DOWNSTREAM:
1929     {
1930       GstClockTime timestamp, stream_time, running_time;
1931       gboolean all_headers;
1932       guint count;
1933       const GstStructure *s;
1934
1935       s = gst_event_get_structure (event);
1936
1937       if (gst_structure_has_name (s, "scte-sit") && mux->scte35_pid != 0) {
1938
1939         /* When operating downstream of tsdemux, tsdemux will send out events
1940          * on all its source pads for each splice table it encounters. If we
1941          * are remuxing multiple streams it has demuxed, this means we could
1942          * unnecessarily repeat the same table multiple times, we avoid that
1943          * by deduplicating thanks to the event sequm
1944          */
1945         if (gst_event_get_seqnum (event) != mux->last_scte35_event_seqnum) {
1946           GstMpegtsSection *section;
1947
1948           gst_structure_get (s, "section", GST_TYPE_MPEGTS_SECTION, &section,
1949               NULL);
1950           if (section) {
1951             guint64 mpeg_pts_offset = 0;
1952             GstStructure *rtime_map = NULL;
1953
1954             gst_structure_get (s, "running-time-map", GST_TYPE_STRUCTURE,
1955                 &rtime_map, NULL);
1956             gst_structure_get_uint64 (s, "mpeg-pts-offset", &mpeg_pts_offset);
1957
1958             handle_scte35_section (mux, event, section, mpeg_pts_offset,
1959                 rtime_map);
1960             if (rtime_map)
1961               gst_structure_free (rtime_map);
1962             mux->last_scte35_event_seqnum = gst_event_get_seqnum (event);
1963           } else {
1964             GST_WARNING_OBJECT (ts_pad,
1965                 "Ignoring scte-sit event without a section");
1966           }
1967         } else {
1968           GST_DEBUG_OBJECT (ts_pad, "Ignoring duplicate scte-sit event");
1969         }
1970         res = TRUE;
1971         forward = FALSE;
1972         goto out;
1973       }
1974
1975       if (!gst_video_event_is_force_key_unit (event))
1976         goto out;
1977
1978       res = TRUE;
1979       forward = FALSE;
1980
1981       gst_video_event_parse_downstream_force_key_unit (event,
1982           &timestamp, &stream_time, &running_time, &all_headers, &count);
1983       GST_INFO_OBJECT (ts_pad, "have downstream force-key-unit event, "
1984           "seqnum %d, running-time %" GST_TIME_FORMAT " count %d",
1985           gst_event_get_seqnum (event), GST_TIME_ARGS (running_time), count);
1986
1987       if (mux->force_key_unit_event != NULL) {
1988         GST_INFO_OBJECT (mux, "skipping downstream force key unit event "
1989             "as an upstream force key unit is already queued");
1990         goto out;
1991       }
1992
1993       if (!all_headers)
1994         goto out;
1995
1996       mux->pending_key_unit_ts = running_time;
1997       gst_event_replace (&mux->force_key_unit_event, event);
1998       break;
1999     }
2000     case GST_EVENT_TAG:{
2001       GstTagList *list;
2002       gchar *lang = NULL;
2003
2004       GST_DEBUG_OBJECT (mux, "received tag event");
2005       gst_event_parse_tag (event, &list);
2006
2007       /* Matroska wants ISO 639-2B code, taglist most likely contains 639-1 */
2008       if (gst_tag_list_get_string (list, GST_TAG_LANGUAGE_CODE, &lang)) {
2009         const gchar *lang_code;
2010
2011         lang_code = gst_tag_get_language_code_iso_639_2B (lang);
2012         if (lang_code) {
2013           GST_DEBUG_OBJECT (ts_pad, "Setting language to '%s'", lang_code);
2014
2015           g_free (ts_pad->language);
2016           ts_pad->language = g_strdup (lang_code);
2017         } else {
2018           GST_WARNING_OBJECT (ts_pad, "Did not get language code for '%s'",
2019               lang);
2020         }
2021         g_free (lang);
2022       }
2023
2024       /* handled this, don't want collectpads to forward it downstream */
2025       res = TRUE;
2026       forward = gst_tag_list_get_scope (list) == GST_TAG_SCOPE_GLOBAL;
2027       break;
2028     }
2029     case GST_EVENT_STREAM_START:{
2030       GstStreamFlags flags;
2031
2032       gst_event_parse_stream_flags (event, &flags);
2033
2034       /* Don't wait for data on sparse inputs like metadata streams */
2035       /*
2036          if ((flags & GST_STREAM_FLAG_SPARSE)) {
2037          GST_COLLECT_PADS_STATE_UNSET (data, GST_COLLECT_PADS_STATE_LOCKED);
2038          gst_collect_pads_set_waiting (pads, data, FALSE);
2039          GST_COLLECT_PADS_STATE_SET (data, GST_COLLECT_PADS_STATE_LOCKED);
2040          }
2041        */
2042       break;
2043     }
2044     default:
2045       break;
2046   }
2047
2048 out:
2049   if (!forward)
2050     gst_event_unref (event);
2051   else
2052     res = agg_class->sink_event (agg, agg_pad, event);
2053
2054   return res;
2055 }
2056
2057 static gboolean
2058 gst_base_ts_mux_src_event (GstAggregator * agg, GstEvent * event)
2059 {
2060   GstAggregatorClass *agg_class = GST_AGGREGATOR_CLASS (parent_class);
2061   GstBaseTsMux *mux = GST_BASE_TS_MUX (agg);
2062   gboolean res = TRUE, forward = TRUE;
2063
2064   switch (GST_EVENT_TYPE (event)) {
2065     case GST_EVENT_CUSTOM_UPSTREAM:
2066     {
2067       GstIterator *iter;
2068       GValue sinkpad_value = G_VALUE_INIT;
2069       GstClockTime running_time;
2070       gboolean all_headers, done = FALSE, res = FALSE;
2071       guint count;
2072
2073       if (!gst_video_event_is_force_key_unit (event))
2074         break;
2075
2076       forward = FALSE;
2077
2078       gst_video_event_parse_upstream_force_key_unit (event,
2079           &running_time, &all_headers, &count);
2080
2081       GST_INFO_OBJECT (mux, "received upstream force-key-unit event, "
2082           "seqnum %d running_time %" GST_TIME_FORMAT " all_headers %d count %d",
2083           gst_event_get_seqnum (event), GST_TIME_ARGS (running_time),
2084           all_headers, count);
2085
2086       if (!all_headers)
2087         break;
2088
2089       mux->pending_key_unit_ts = running_time;
2090       gst_event_replace (&mux->force_key_unit_event, event);
2091
2092       iter = gst_element_iterate_sink_pads (GST_ELEMENT_CAST (mux));
2093
2094       while (!done) {
2095         switch (gst_iterator_next (iter, &sinkpad_value)) {
2096           case GST_ITERATOR_OK:{
2097             GstPad *sinkpad = g_value_get_object (&sinkpad_value);
2098             gboolean tmp;
2099
2100             GST_INFO_OBJECT (GST_AGGREGATOR_SRC_PAD (agg), "forwarding");
2101             tmp = gst_pad_push_event (sinkpad, gst_event_ref (event));
2102             GST_INFO_OBJECT (mux, "result %d", tmp);
2103             /* succeed if at least one pad succeeds */
2104             res |= tmp;
2105             break;
2106           }
2107           case GST_ITERATOR_DONE:
2108             done = TRUE;
2109             break;
2110           case GST_ITERATOR_RESYNC:
2111             gst_iterator_resync (iter);
2112             break;
2113           case GST_ITERATOR_ERROR:
2114             g_assert_not_reached ();
2115             break;
2116         }
2117         g_value_reset (&sinkpad_value);
2118       }
2119       g_value_unset (&sinkpad_value);
2120       gst_iterator_free (iter);
2121       break;
2122     }
2123     default:
2124       break;
2125   }
2126
2127   if (forward)
2128     res = agg_class->src_event (agg, event);
2129   else
2130     gst_event_unref (event);
2131
2132   return res;
2133 }
2134
2135 static GstBuffer *
2136 gst_base_ts_mux_clip (GstAggregator * agg,
2137     GstAggregatorPad * agg_pad, GstBuffer * buf)
2138 {
2139   GstBaseTsMuxPad *pad = GST_BASE_TS_MUX_PAD (agg_pad);
2140   GstClockTime time;
2141   GstBuffer *ret;
2142
2143   ret = buf;
2144
2145   /* PTS */
2146   time = GST_BUFFER_PTS (buf);
2147
2148   /* invalid left alone and passed */
2149   if (G_LIKELY (GST_CLOCK_TIME_IS_VALID (time))) {
2150     time =
2151         gst_segment_to_running_time (&agg_pad->segment, GST_FORMAT_TIME, time);
2152     if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (time))) {
2153       GST_DEBUG_OBJECT (pad, "clipping buffer on pad outside segment");
2154       gst_buffer_unref (buf);
2155       ret = NULL;
2156       goto beach;
2157     } else {
2158       GST_LOG_OBJECT (pad, "buffer pts %" GST_TIME_FORMAT " ->  %"
2159           GST_TIME_FORMAT " running time",
2160           GST_TIME_ARGS (GST_BUFFER_PTS (buf)), GST_TIME_ARGS (time));
2161       buf = ret = gst_buffer_make_writable (buf);
2162       GST_BUFFER_PTS (ret) = time;
2163     }
2164   }
2165
2166   /* DTS */
2167   time = GST_BUFFER_DTS (buf);
2168
2169   /* invalid left alone and passed */
2170   if (G_LIKELY (GST_CLOCK_TIME_IS_VALID (time))) {
2171     gint sign;
2172     gint64 dts;
2173
2174     sign = gst_segment_to_running_time_full (&agg_pad->segment, GST_FORMAT_TIME,
2175         time, &time);
2176
2177     if (sign > 0)
2178       dts = (gint64) time;
2179     else
2180       dts = -((gint64) time);
2181
2182     GST_LOG_OBJECT (pad, "buffer dts %" GST_TIME_FORMAT " -> %"
2183         GST_STIME_FORMAT " running time", GST_TIME_ARGS (GST_BUFFER_DTS (buf)),
2184         GST_STIME_ARGS (dts));
2185
2186     if (GST_CLOCK_STIME_IS_VALID (pad->dts) && dts < pad->dts) {
2187       /* Ignore DTS going backward */
2188       GST_WARNING_OBJECT (pad, "ignoring DTS going backward");
2189       dts = pad->dts;
2190     }
2191
2192     ret = gst_buffer_make_writable (buf);
2193     if (sign > 0)
2194       GST_BUFFER_DTS (ret) = time;
2195     else
2196       GST_BUFFER_DTS (ret) = GST_CLOCK_TIME_NONE;
2197
2198     pad->dts = dts;
2199   } else {
2200     pad->dts = GST_CLOCK_STIME_NONE;
2201   }
2202
2203 beach:
2204   return ret;
2205 }
2206
2207 static GstFlowReturn
2208 gst_base_ts_mux_update_src_caps (GstAggregator * agg, GstCaps * caps,
2209     GstCaps ** ret)
2210 {
2211   GstBaseTsMux *mux = GST_BASE_TS_MUX (agg);
2212   GstStructure *s;
2213
2214   *ret = gst_caps_copy (caps);
2215   s = gst_caps_get_structure (*ret, 0);
2216   gst_structure_set (s, "packetsize", G_TYPE_INT, mux->packet_size, NULL);
2217
2218   return GST_FLOW_OK;
2219 }
2220
2221 static GstBaseTsMuxPad *
2222 gst_base_ts_mux_find_best_pad (GstAggregator * aggregator)
2223 {
2224   GstBaseTsMuxPad *best = NULL;
2225   GstClockTime best_ts = GST_CLOCK_TIME_NONE;
2226   GList *l;
2227
2228   GST_OBJECT_LOCK (aggregator);
2229
2230   for (l = GST_ELEMENT_CAST (aggregator)->sinkpads; l; l = l->next) {
2231     GstBaseTsMuxPad *tpad = GST_BASE_TS_MUX_PAD (l->data);
2232     GstAggregatorPad *apad = GST_AGGREGATOR_PAD_CAST (tpad);
2233     GstBuffer *buffer;
2234
2235     buffer = gst_aggregator_pad_peek_buffer (apad);
2236     if (!buffer)
2237       continue;
2238     if (best_ts == GST_CLOCK_TIME_NONE) {
2239       best = tpad;
2240       best_ts = GST_BUFFER_DTS_OR_PTS (buffer);
2241     } else if (GST_BUFFER_DTS_OR_PTS (buffer) != GST_CLOCK_TIME_NONE) {
2242       GstClockTime t = GST_BUFFER_DTS_OR_PTS (buffer);
2243       if (t < best_ts) {
2244         best = tpad;
2245         best_ts = t;
2246       }
2247     }
2248     gst_buffer_unref (buffer);
2249   }
2250
2251   if (best)
2252     gst_object_ref (best);
2253
2254   GST_OBJECT_UNLOCK (aggregator);
2255
2256   GST_DEBUG_OBJECT (aggregator,
2257       "Best pad found with %" GST_TIME_FORMAT ": %" GST_PTR_FORMAT,
2258       GST_TIME_ARGS (best_ts), best);
2259
2260   return best;
2261 }
2262
2263 static gboolean
2264 gst_base_ts_mux_are_all_pads_eos (GstBaseTsMux * mux)
2265 {
2266   GList *l;
2267   gboolean ret = TRUE;
2268
2269   GST_OBJECT_LOCK (mux);
2270
2271   for (l = GST_ELEMENT_CAST (mux)->sinkpads; l; l = l->next) {
2272     GstBaseTsMuxPad *pad = GST_BASE_TS_MUX_PAD (l->data);
2273
2274     if (!gst_aggregator_pad_is_eos (GST_AGGREGATOR_PAD (pad))) {
2275       ret = FALSE;
2276       break;
2277     }
2278   }
2279
2280   GST_OBJECT_UNLOCK (mux);
2281
2282   return ret;
2283 }
2284
2285
2286 static GstFlowReturn
2287 gst_base_ts_mux_aggregate (GstAggregator * agg, gboolean timeout)
2288 {
2289   GstBaseTsMux *mux = GST_BASE_TS_MUX (agg);
2290   GstFlowReturn ret = GST_FLOW_OK;
2291   GstBaseTsMuxPad *best = gst_base_ts_mux_find_best_pad (agg);
2292
2293   if (best) {
2294     GstBuffer *buffer;
2295
2296     buffer = gst_aggregator_pad_pop_buffer (GST_AGGREGATOR_PAD (best));
2297     if (!buffer) {
2298       /* We might have gotten a flush event after we picked the pad */
2299       goto done;
2300     }
2301
2302     ret =
2303         gst_base_ts_mux_aggregate_buffer (GST_BASE_TS_MUX (agg),
2304         GST_AGGREGATOR_PAD (best), buffer);
2305
2306     gst_object_unref (best);
2307
2308     if (ret != GST_FLOW_OK)
2309       goto done;
2310   }
2311
2312   if (gst_base_ts_mux_are_all_pads_eos (mux)) {
2313     GstBaseTsMuxClass *klass = GST_BASE_TS_MUX_GET_CLASS (mux);
2314     /* drain some possibly cached data */
2315     if (klass->drain)
2316       klass->drain (mux);
2317     gst_base_ts_mux_push_packets (mux, TRUE);
2318
2319     ret = GST_FLOW_EOS;
2320   }
2321
2322 done:
2323   return ret;
2324 }
2325
2326 static gboolean
2327 gst_base_ts_mux_start (GstAggregator * agg)
2328 {
2329   gst_base_ts_mux_reset (GST_BASE_TS_MUX (agg), TRUE);
2330
2331   return TRUE;
2332 }
2333
2334 static gboolean
2335 gst_base_ts_mux_stop (GstAggregator * agg)
2336 {
2337   gst_base_ts_mux_reset (GST_BASE_TS_MUX (agg), TRUE);
2338
2339   return TRUE;
2340 }
2341
2342 /* GObject implementation */
2343
2344 static void
2345 gst_base_ts_mux_dispose (GObject * object)
2346 {
2347   GstBaseTsMux *mux = GST_BASE_TS_MUX (object);
2348
2349   gst_base_ts_mux_reset (mux, FALSE);
2350
2351   if (mux->out_adapter) {
2352     g_object_unref (mux->out_adapter);
2353     mux->out_adapter = NULL;
2354   }
2355   if (mux->prog_map) {
2356     gst_structure_free (mux->prog_map);
2357     mux->prog_map = NULL;
2358   }
2359   if (mux->programs) {
2360     g_hash_table_destroy (mux->programs);
2361     mux->programs = NULL;
2362   }
2363   GST_CALL_PARENT (G_OBJECT_CLASS, dispose, (object));
2364 }
2365
2366 static void
2367 gst_base_ts_mux_constructed (GObject * object)
2368 {
2369   GstBaseTsMux *mux = GST_BASE_TS_MUX (object);
2370
2371   /* initial state */
2372   gst_base_ts_mux_reset (mux, TRUE);
2373 }
2374
2375 static void
2376 gst_base_ts_mux_set_property (GObject * object, guint prop_id,
2377     const GValue * value, GParamSpec * pspec)
2378 {
2379   GstBaseTsMux *mux = GST_BASE_TS_MUX (object);
2380   GList *l;
2381
2382   switch (prop_id) {
2383     case PROP_PROG_MAP:
2384     {
2385       const GstStructure *s = gst_value_get_structure (value);
2386       if (mux->prog_map) {
2387         gst_structure_free (mux->prog_map);
2388       }
2389       if (s)
2390         mux->prog_map = gst_structure_copy (s);
2391       else
2392         mux->prog_map = NULL;
2393       break;
2394     }
2395     case PROP_PAT_INTERVAL:
2396       mux->pat_interval = g_value_get_uint (value);
2397       if (mux->tsmux)
2398         tsmux_set_pat_interval (mux->tsmux, mux->pat_interval);
2399       break;
2400     case PROP_PMT_INTERVAL:
2401       mux->pmt_interval = g_value_get_uint (value);
2402       GST_OBJECT_LOCK (mux);
2403       for (l = GST_ELEMENT_CAST (mux)->sinkpads; l; l = l->next) {
2404         GstBaseTsMuxPad *ts_pad = GST_BASE_TS_MUX_PAD (l->data);
2405
2406         tsmux_set_pmt_interval (ts_pad->prog, mux->pmt_interval);
2407       }
2408       GST_OBJECT_UNLOCK (mux);
2409       break;
2410     case PROP_ALIGNMENT:
2411       mux->alignment = g_value_get_int (value);
2412       break;
2413     case PROP_SI_INTERVAL:
2414       mux->si_interval = g_value_get_uint (value);
2415       tsmux_set_si_interval (mux->tsmux, mux->si_interval);
2416       break;
2417     case PROP_BITRATE:
2418       mux->bitrate = g_value_get_uint64 (value);
2419       if (mux->tsmux)
2420         tsmux_set_bitrate (mux->tsmux, mux->bitrate);
2421       break;
2422     case PROP_PCR_INTERVAL:
2423       mux->pcr_interval = g_value_get_uint (value);
2424       if (mux->tsmux)
2425         tsmux_set_pcr_interval (mux->tsmux, mux->pcr_interval);
2426       break;
2427     case PROP_SCTE_35_PID:
2428       mux->scte35_pid = g_value_get_uint (value);
2429       break;
2430     case PROP_SCTE_35_NULL_INTERVAL:
2431       mux->scte35_null_interval = g_value_get_uint (value);
2432       break;
2433     default:
2434       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2435       break;
2436   }
2437 }
2438
2439 static void
2440 gst_base_ts_mux_get_property (GObject * object, guint prop_id,
2441     GValue * value, GParamSpec * pspec)
2442 {
2443   GstBaseTsMux *mux = GST_BASE_TS_MUX (object);
2444
2445   switch (prop_id) {
2446     case PROP_PROG_MAP:
2447       gst_value_set_structure (value, mux->prog_map);
2448       break;
2449     case PROP_PAT_INTERVAL:
2450       g_value_set_uint (value, mux->pat_interval);
2451       break;
2452     case PROP_PMT_INTERVAL:
2453       g_value_set_uint (value, mux->pmt_interval);
2454       break;
2455     case PROP_ALIGNMENT:
2456       g_value_set_int (value, mux->alignment);
2457       break;
2458     case PROP_SI_INTERVAL:
2459       g_value_set_uint (value, mux->si_interval);
2460       break;
2461     case PROP_BITRATE:
2462       g_value_set_uint64 (value, mux->bitrate);
2463       break;
2464     case PROP_PCR_INTERVAL:
2465       g_value_set_uint (value, mux->pcr_interval);
2466       break;
2467     case PROP_SCTE_35_PID:
2468       g_value_set_uint (value, mux->scte35_pid);
2469       break;
2470     case PROP_SCTE_35_NULL_INTERVAL:
2471       g_value_set_uint (value, mux->scte35_null_interval);
2472       break;
2473     default:
2474       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2475       break;
2476   }
2477 }
2478
2479 /* Default vmethods implementation */
2480
2481 static TsMux *
2482 gst_base_ts_mux_default_create_ts_mux (GstBaseTsMux * mux)
2483 {
2484   TsMux *tsmux = tsmux_new ();
2485   tsmux_set_write_func (tsmux, new_packet_cb, mux);
2486   tsmux_set_alloc_func (tsmux, alloc_packet_cb, mux);
2487   tsmux_set_pat_interval (tsmux, mux->pat_interval);
2488   tsmux_set_si_interval (tsmux, mux->si_interval);
2489   tsmux_set_bitrate (tsmux, mux->bitrate);
2490   tsmux_set_pcr_interval (tsmux, mux->pcr_interval);
2491
2492   return tsmux;
2493 }
2494
2495 static void
2496 gst_base_ts_mux_default_allocate_packet (GstBaseTsMux * mux,
2497     GstBuffer ** buffer)
2498 {
2499   GstBuffer *buf;
2500
2501   buf = gst_buffer_new_and_alloc (mux->packet_size);
2502
2503   *buffer = buf;
2504 }
2505
2506 static gboolean
2507 gst_base_ts_mux_default_output_packet (GstBaseTsMux * mux, GstBuffer * buffer,
2508     gint64 new_pcr)
2509 {
2510   gst_base_ts_mux_collect_packet (mux, buffer);
2511
2512   return TRUE;
2513 }
2514
2515 /* Subclass API */
2516
2517 void
2518 gst_base_ts_mux_set_packet_size (GstBaseTsMux * mux, gsize size)
2519 {
2520   mux->packet_size = size;
2521 }
2522
2523 void
2524 gst_base_ts_mux_set_automatic_alignment (GstBaseTsMux * mux, gsize alignment)
2525 {
2526   mux->automatic_alignment = alignment;
2527 }
2528
2529 static void
2530 gst_base_ts_mux_class_init (GstBaseTsMuxClass * klass)
2531 {
2532   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
2533   GstAggregatorClass *gstagg_class = GST_AGGREGATOR_CLASS (klass);
2534   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
2535
2536   GST_DEBUG_CATEGORY_INIT (gst_base_ts_mux_debug, "basetsmux", 0,
2537       "MPEG Transport Stream muxer");
2538
2539   gst_element_class_set_static_metadata (gstelement_class,
2540       "MPEG Transport Stream Muxer", "Codec/Muxer",
2541       "Multiplexes media streams into an MPEG Transport Stream",
2542       "Fluendo <contact@fluendo.com>");
2543
2544   gobject_class->set_property =
2545       GST_DEBUG_FUNCPTR (gst_base_ts_mux_set_property);
2546   gobject_class->get_property =
2547       GST_DEBUG_FUNCPTR (gst_base_ts_mux_get_property);
2548   gobject_class->dispose = gst_base_ts_mux_dispose;
2549   gobject_class->constructed = gst_base_ts_mux_constructed;
2550
2551   gstelement_class->request_new_pad = gst_base_ts_mux_request_new_pad;
2552   gstelement_class->release_pad = gst_base_ts_mux_release_pad;
2553   gstelement_class->send_event = gst_base_ts_mux_send_event;
2554
2555   gstagg_class->update_src_caps = gst_base_ts_mux_update_src_caps;
2556   gstagg_class->aggregate = gst_base_ts_mux_aggregate;
2557   gstagg_class->clip = gst_base_ts_mux_clip;
2558   gstagg_class->sink_event = gst_base_ts_mux_sink_event;
2559   gstagg_class->src_event = gst_base_ts_mux_src_event;
2560   gstagg_class->start = gst_base_ts_mux_start;
2561   gstagg_class->stop = gst_base_ts_mux_stop;
2562
2563   klass->create_ts_mux = gst_base_ts_mux_default_create_ts_mux;
2564   klass->allocate_packet = gst_base_ts_mux_default_allocate_packet;
2565   klass->output_packet = gst_base_ts_mux_default_output_packet;
2566
2567   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PROG_MAP,
2568       g_param_spec_boxed ("prog-map", "Program map",
2569           "A GstStructure specifies the mapping from elementary streams to programs",
2570           GST_TYPE_STRUCTURE,
2571           (GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
2572
2573   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PAT_INTERVAL,
2574       g_param_spec_uint ("pat-interval", "PAT interval",
2575           "Set the interval (in ticks of the 90kHz clock) for writing out the PAT table",
2576           1, G_MAXUINT, TSMUX_DEFAULT_PAT_INTERVAL,
2577           (GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
2578
2579   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PMT_INTERVAL,
2580       g_param_spec_uint ("pmt-interval", "PMT interval",
2581           "Set the interval (in ticks of the 90kHz clock) for writing out the PMT table",
2582           1, G_MAXUINT, TSMUX_DEFAULT_PMT_INTERVAL,
2583           (GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
2584
2585   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_ALIGNMENT,
2586       g_param_spec_int ("alignment", "packet alignment",
2587           "Number of packets per buffer (padded with dummy packets on EOS) "
2588           "(-1 = auto, 0 = all available packets, 7 for UDP streaming)",
2589           -1, G_MAXINT, BASETSMUX_DEFAULT_ALIGNMENT,
2590           (GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
2591
2592   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SI_INTERVAL,
2593       g_param_spec_uint ("si-interval", "SI interval",
2594           "Set the interval (in ticks of the 90kHz clock) for writing out the Service"
2595           "Information tables", 1, G_MAXUINT, TSMUX_DEFAULT_SI_INTERVAL,
2596           (GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
2597
2598   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BITRATE,
2599       g_param_spec_uint64 ("bitrate", "Bitrate (in bits per second)",
2600           "Set the target bitrate, will insert null packets as padding "
2601           " to achieve multiplex-wide constant bitrate (0 means no padding)",
2602           0, G_MAXUINT64, TSMUX_DEFAULT_BITRATE,
2603           (GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
2604
2605   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PCR_INTERVAL,
2606       g_param_spec_uint ("pcr-interval", "PCR interval",
2607           "Set the interval (in ticks of the 90kHz clock) for writing PCR",
2608           1, G_MAXUINT, TSMUX_DEFAULT_PCR_INTERVAL,
2609           (GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
2610
2611   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SCTE_35_PID,
2612       g_param_spec_uint ("scte-35-pid", "SCTE-35 PID",
2613           "PID to use for inserting SCTE-35 packets (0: unused)",
2614           0, G_MAXUINT, DEFAULT_SCTE_35_PID,
2615           (GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
2616
2617   g_object_class_install_property (G_OBJECT_CLASS (klass),
2618       PROP_SCTE_35_NULL_INTERVAL, g_param_spec_uint ("scte-35-null-interval",
2619           "SCTE-35 NULL packet interval",
2620           "Set the interval (in ticks of the 90kHz clock) for writing SCTE-35 NULL (heartbeat) packets."
2621           " (only valid if scte-35-pid is different from 0)", 1, G_MAXUINT,
2622           TSMUX_DEFAULT_SCTE_35_NULL_INTERVAL,
2623           (GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
2624
2625   gst_element_class_add_static_pad_template_with_gtype (gstelement_class,
2626       &gst_base_ts_mux_src_factory, GST_TYPE_AGGREGATOR_PAD);
2627
2628   gst_type_mark_as_plugin_api (GST_TYPE_BASE_TS_MUX_PAD, 0);
2629 }
2630
2631 static void
2632 gst_base_ts_mux_init (GstBaseTsMux * mux)
2633 {
2634   mux->out_adapter = gst_adapter_new ();
2635
2636   /* properties */
2637   mux->pat_interval = TSMUX_DEFAULT_PAT_INTERVAL;
2638   mux->pmt_interval = TSMUX_DEFAULT_PMT_INTERVAL;
2639   mux->si_interval = TSMUX_DEFAULT_SI_INTERVAL;
2640   mux->pcr_interval = TSMUX_DEFAULT_PCR_INTERVAL;
2641   mux->prog_map = NULL;
2642   mux->alignment = BASETSMUX_DEFAULT_ALIGNMENT;
2643   mux->bitrate = TSMUX_DEFAULT_BITRATE;
2644   mux->scte35_pid = DEFAULT_SCTE_35_PID;
2645   mux->scte35_null_interval = TSMUX_DEFAULT_SCTE_35_NULL_INTERVAL;
2646
2647   mux->packet_size = GST_BASE_TS_MUX_NORMAL_PACKET_LENGTH;
2648   mux->automatic_alignment = 0;
2649 }