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