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

   1 /*
   2  * Copyright 2006 BBC and Fluendo S.A.
   3  *
   4  * This library is licensed under 3 different licenses and you
   5  * can choose to use it under the terms of any one of them. The
   6  * three licenses are the MPL 1.1, the LGPL, and the MIT license.
   7  *
   8  * MPL:
   9  *
  10  * The contents of this file are subject to the Mozilla Public License
  11  * Version 1.1 (the "License"); you may not use this file except in
  12  * compliance with the License. You may obtain a copy of the License at
  13  * http://www.mozilla.org/MPL/.
  14  *
  15  * Software distributed under the License is distributed on an "AS IS"
  16  * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
  17  * License for the specific language governing rights and limitations
  18  * under the License.
  19  *
  20  * LGPL:
  21  *
  22  * This library is free software; you can redistribute it and/or
  23  * modify it under the terms of the GNU Library General Public
  24  * License as published by the Free Software Foundation; either
  25  * version 2 of the License, or (at your option) any later version.
  26  *
  27  * This library is distributed in the hope that it will be useful,
  28  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  29  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  30  * Library General Public License for more details.
  31  *
  32  * You should have received a copy of the GNU Library General Public
  33  * License along with this library; if not, write to the
  34  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  35  * Boston, MA 02110-1301, USA.
  36  *
  37  * MIT:
  38  *
  39  * Unless otherwise indicated, Source Code is licensed under MIT license.
  40  * See further explanation attached in License Statement (distributed in the file
  41  * LICENSE).
  42  *
  43  * Permission is hereby granted, free of charge, to any person obtaining a copy of
  44  * this software and associated documentation files (the "Software"), to deal in
  45  * the Software without restriction, including without limitation the rights to
  46  * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
  47  * of the Software, and to permit persons to whom the Software is furnished to do
  48  * so, subject to the following conditions:
  49  *
  50  * The above copyright notice and this permission notice shall be included in all
  51  * copies or substantial portions of the Software.
  52  *
  53  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  54  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  55  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  56  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  57  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  58  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  59  * SOFTWARE.
  60  *
  61  * SPDX-License-Identifier: MPL-1.1 OR MIT OR LGPL-2.0-or-later
  62  */
  63
  64 #ifdef HAVE_CONFIG_H
  65 #include "config.h"
  66 #endif
  67
  68 #include <string.h>
  69
  70 #include <gst/mpegts/mpegts.h>
  71
  72 #include "tsmux.h"
  73 #include "tsmuxstream.h"
  74
  75 #define GST_CAT_DEFAULT gst_base_ts_mux_debug
  76
  77 /* Maximum total data length for a PAT section is 1024 bytes, minus an
  78  * 8 byte header, then the length of each program entry is 32 bits,
  79  * then finally a 32 bit CRC. Thus the maximum number of programs in this mux
  80  * is (1024 - 8 - 4) / 4 = 253 because it only supports single section PATs */
  81 #define TSMUX_MAX_PROGRAMS 253
  82
  83 #define TSMUX_SECTION_HDR_SIZE 8
  84
  85 #define TSMUX_DEFAULT_NETWORK_ID 0x0001
  86 #define TSMUX_DEFAULT_TS_ID 0x0001
  87
  88 /* The last byte of the PCR in the header defines the byte position
  89  * at which PCR should be calculated */
  90 #define PCR_BYTE_OFFSET 11
  91
  92 /* HACK: We use a fixed buffering offset for the PCR at the moment -
  93  * this is the amount 'in advance' of the stream that the PCR sits.
  94  * 1/8 second atm */
  95 #define TSMUX_PCR_OFFSET (TSMUX_CLOCK_FREQ / 8)
  96
  97 /* Base for all written PCR and DTS/PTS,
  98  * so we have some slack to go backwards */
  99 #define CLOCK_BASE (TSMUX_CLOCK_FREQ * 10 * 360)
 100
 101 static gboolean tsmux_write_pat (TsMux * mux);
 102 static gboolean tsmux_write_pmt (TsMux * mux, TsMuxProgram * program);
 103 static gboolean tsmux_write_scte_null (TsMux * mux, TsMuxProgram * program);
 104 static gint64 get_next_pcr (TsMux * mux, gint64 cur_ts);
 105 static gint64 get_current_pcr (TsMux * mux, gint64 cur_ts);
 106 static gint64 write_new_pcr (TsMux * mux, TsMuxStream * stream, gint64 cur_pcr,
 107     gint64 next_pcr);
 108 static gboolean tsmux_write_ts_header (TsMux * mux, guint8 * buf,
 109     TsMuxPacketInfo * pi, guint stream_avail, guint * payload_len_out,
 110     guint * payload_offset_out);
 111
 112 static void
 113 tsmux_section_free (TsMuxSection * section)
 114 {
 115   gst_mpegts_section_unref (section->section);
 116   g_slice_free (TsMuxSection, section);
 117 }
 118
 119 /**
 120  * tsmux_new:
 121  *
 122  * Create a new muxer session.
 123  *
 124  * Returns: A new #TsMux object.
 125  */
 126 TsMux *
 127 tsmux_new (void)
 128 {
 129   TsMux *mux;
 130
 131   mux = g_slice_new0 (TsMux);
 132
 133   mux->transport_id = TSMUX_DEFAULT_TS_ID;
 134
 135   mux->next_pgm_no = TSMUX_START_PROGRAM_ID;
 136   mux->next_pmt_pid = TSMUX_START_PMT_PID;
 137   mux->next_stream_pid = TSMUX_START_ES_PID;
 138
 139   mux->pat_changed = TRUE;
 140   mux->next_pat_pcr = -1;
 141   mux->pat_interval = TSMUX_DEFAULT_PAT_INTERVAL;
 142
 143   mux->si_changed = TRUE;
 144   mux->si_interval = TSMUX_DEFAULT_SI_INTERVAL;
 145
 146   mux->pcr_interval = TSMUX_DEFAULT_PCR_INTERVAL;
 147
 148   mux->next_si_pcr = -1;
 149
 150   mux->si_sections = g_hash_table_new_full (g_direct_hash, g_direct_equal,
 151       NULL, (GDestroyNotify) tsmux_section_free);
 152
 153   mux->new_stream_func = (TsMuxNewStreamFunc) tsmux_stream_new;
 154   mux->new_stream_data = NULL;
 155
 156   mux->first_pcr_ts = G_MININT64;
 157
 158   return mux;
 159 }
 160
 161 /**
 162  * tsmux_set_write_func:
 163  * @mux: a #TsMux
 164  * @func: a user callback function
 165  * @user_data: user data passed to @func
 166  *
 167  * Set the callback function and user data to be called when @mux has output to
 168  * produce. @user_data will be passed as user data in @func.
 169  */
 170 void
 171 tsmux_set_write_func (TsMux * mux, TsMuxWriteFunc func, void *user_data)
 172 {
 173   g_return_if_fail (mux != NULL);
 174
 175   mux->write_func = func;
 176   mux->write_func_data = user_data;
 177 }
 178
 179 /**
 180  * tsmux_set_alloc_func:
 181  * @mux: a #TsMux
 182  * @func: a user callback function
 183  * @user_data: user data passed to @func
 184  *
 185  * Set the callback function and user data to be called when @mux needs
 186  * a new buffer to write a packet into.
 187  * @user_data will be passed as user data in @func.
 188  */
 189 void
 190 tsmux_set_alloc_func (TsMux * mux, TsMuxAllocFunc func, void *user_data)
 191 {
 192   g_return_if_fail (mux != NULL);
 193
 194   mux->alloc_func = func;
 195   mux->alloc_func_data = user_data;
 196 }
 197
 198 /**
 199  * tsmux_set_new_stream_func:
 200  * @mux: a #TsMux
 201  * @func: a user callback function
 202  * @user_data: user data passed to @func
 203  *
 204  * Set the callback function and user data to be called when @mux needs
 205  * to create a new stream.
 206  * @user_data will be passed as user data in @func.
 207  */
 208 void
 209 tsmux_set_new_stream_func (TsMux * mux, TsMuxNewStreamFunc func,
 210     void *user_data)
 211 {
 212   g_return_if_fail (mux != NULL);
 213
 214   mux->new_stream_func = func;
 215   mux->new_stream_data = user_data;
 216 }
 217
 218 /**
 219  * tsmux_set_pat_interval:
 220  * @mux: a #TsMux
 221  * @freq: a new PAT interval
 222  *
 223  * Set the interval (in cycles of the 90kHz clock) for writing out the PAT table.
 224  *
 225  * Many transport stream clients might have problems if the PAT table is not
 226  * inserted in the stream at regular intervals, especially when initially trying
 227  * to figure out the contents of the stream.
 228  */
 229 void
 230 tsmux_set_pat_interval (TsMux * mux, guint freq)
 231 {
 232   g_return_if_fail (mux != NULL);
 233
 234   mux->pat_interval = freq;
 235 }
 236
 237 /**
 238  * tsmux_set_pcr_interval:
 239  * @mux: a #TsMux
 240  * @freq: a new PCR interval
 241  *
 242  * Set the interval (in cycles of the 90kHz clock) for writing the PCR.
 243  */
 244 void
 245 tsmux_set_pcr_interval (TsMux * mux, guint freq)
 246 {
 247   g_return_if_fail (mux != NULL);
 248
 249   mux->pcr_interval = freq;
 250 }
 251
 252 /**
 253  * tsmux_get_pat_interval:
 254  * @mux: a #TsMux
 255  *
 256  * Get the configured PAT interval. See also tsmux_set_pat_interval().
 257  *
 258  * Returns: the configured PAT interval
 259  */
 260 guint
 261 tsmux_get_pat_interval (TsMux * mux)
 262 {
 263   g_return_val_if_fail (mux != NULL, 0);
 264
 265   return mux->pat_interval;
 266 }
 267
 268 /**
 269  * tsmux_resend_pat:
 270  * @mux: a #TsMux
 271  *
 272  * Resends the PAT before the next stream packet.
 273  */
 274 void
 275 tsmux_resend_pat (TsMux * mux)
 276 {
 277   g_return_if_fail (mux != NULL);
 278
 279   mux->next_pat_pcr = -1;
 280 }
 281
 282 /**
 283  * tsmux_set_si_interval:
 284  * @mux: a #TsMux
 285  * @freq: a new SI table interval
 286  *
 287  * Set the interval (in cycles of the 90kHz clock) for writing out the SI tables.
 288  *
 289  */
 290 void
 291 tsmux_set_si_interval (TsMux * mux, guint freq)
 292 {
 293   g_return_if_fail (mux != NULL);
 294
 295   mux->si_interval = freq;
 296 }
 297
 298 /**
 299  * tsmux_get_si_interval:
 300  * @mux: a #TsMux
 301  *
 302  * Get the configured SI table interval. See also tsmux_set_si_interval().
 303  *
 304  * Returns: the configured SI interval
 305  */
 306 guint
 307 tsmux_get_si_interval (TsMux * mux)
 308 {
 309   g_return_val_if_fail (mux != NULL, 0);
 310
 311   return mux->si_interval;
 312 }
 313
 314 /**
 315  * tsmux_resend_si:
 316  * @mux: a #TsMux
 317  *
 318  * Resends the SI tables before the next stream packet.
 319  *
 320  */
 321 void
 322 tsmux_resend_si (TsMux * mux)
 323 {
 324   g_return_if_fail (mux != NULL);
 325
 326   mux->next_si_pcr = -1;
 327 }
 328
 329 /**
 330  * tsmux_add_mpegts_si_section:
 331  * @mux: a #TsMux
 332  * @section: (transfer full): a #GstMpegtsSection to add
 333  *
 334  * Add a Service Information #GstMpegtsSection to the stream
 335  *
 336  * Returns: %TRUE on success, %FALSE otherwise
 337  */
 338 gboolean
 339 tsmux_add_mpegts_si_section (TsMux * mux, GstMpegtsSection * section)
 340 {
 341   TsMuxSection *tsmux_section;
 342
 343   g_return_val_if_fail (mux != NULL, FALSE);
 344   g_return_val_if_fail (section != NULL, FALSE);
 345   g_return_val_if_fail (mux->si_sections != NULL, FALSE);
 346
 347   tsmux_section = g_slice_new0 (TsMuxSection);
 348
 349   GST_DEBUG ("Adding mpegts section with type %d to mux",
 350       section->section_type);
 351
 352   tsmux_section->section = section;
 353   tsmux_section->pi.pid = section->pid;
 354
 355   g_hash_table_insert (mux->si_sections,
 356       GINT_TO_POINTER (section->section_type), tsmux_section);
 357
 358   mux->si_changed = TRUE;
 359
 360   return TRUE;
 361 }
 362
 363
 364 /**
 365  * tsmux_free:
 366  * @mux: a #TsMux
 367  *
 368  * Free all resources associated with @mux. After calling this function @mux can
 369  * not be used anymore.
 370  */
 371 void
 372 tsmux_free (TsMux * mux)
 373 {
 374   GList *cur;
 375
 376   g_return_if_fail (mux != NULL);
 377
 378   /* Free PAT section */
 379   if (mux->pat.section)
 380     gst_mpegts_section_unref (mux->pat.section);
 381
 382   /* Free all programs */
 383   for (cur = mux->programs; cur; cur = cur->next) {
 384     TsMuxProgram *program = (TsMuxProgram *) cur->data;
 385
 386     tsmux_program_free (program);
 387   }
 388   g_list_free (mux->programs);
 389
 390   /* Free all streams */
 391   for (cur = mux->streams; cur; cur = cur->next) {
 392     TsMuxStream *stream = (TsMuxStream *) cur->data;
 393
 394     tsmux_stream_free (stream);
 395   }
 396   g_list_free (mux->streams);
 397
 398   /* Free SI table sections */
 399   g_hash_table_unref (mux->si_sections);
 400
 401   g_slice_free (TsMux, mux);
 402 }
 403
 404 static gint
 405 tsmux_program_compare (TsMuxProgram * program, gint * needle)
 406 {
 407   return (program->pgm_number - *needle);
 408 }
 409
 410 /**
 411  * tsmux_program_new:
 412  * @mux: a #TsMux
 413  *
 414  * Create a new program in the missing session @mux.
 415  *
 416  * Returns: a new #TsMuxProgram or %NULL when the maximum number of programs has
 417  * been reached.
 418  */
 419 TsMuxProgram *
 420 tsmux_program_new (TsMux * mux, gint prog_id)
 421 {
 422   TsMuxProgram *program;
 423
 424   g_return_val_if_fail (mux != NULL, NULL);
 425
 426   /* Ensure we have room for another program */
 427   if (mux->nb_programs == TSMUX_MAX_PROGRAMS)
 428     return NULL;
 429
 430   program = g_slice_new0 (TsMuxProgram);
 431
 432   program->pmt_changed = TRUE;
 433   program->pmt_interval = TSMUX_DEFAULT_PMT_INTERVAL;
 434
 435   program->next_pmt_pcr = -1;
 436
 437   if (prog_id == 0) {
 438     program->pgm_number = mux->next_pgm_no++;
 439     while (g_list_find_custom (mux->programs, &program->pgm_number,
 440             (GCompareFunc) tsmux_program_compare) != NULL) {
 441       program->pgm_number = mux->next_pgm_no++;
 442     }
 443   } else {
 444     program->pgm_number = prog_id;
 445     while (g_list_find_custom (mux->programs, &program->pgm_number,
 446             (GCompareFunc) tsmux_program_compare) != NULL) {
 447       program->pgm_number++;
 448     }
 449   }
 450
 451   program->pmt_pid = mux->next_pmt_pid++;
 452   program->pcr_stream = NULL;
 453
 454   /* SCTE35 is disabled by default */
 455   program->scte35_pid = 0;
 456   program->scte35_null_interval = TSMUX_DEFAULT_SCTE_35_NULL_INTERVAL;
 457   program->next_scte35_pcr = -1;
 458
 459   /* mux->streams owns the streams */
 460   program->streams = g_ptr_array_new_full (1, NULL);
 461
 462   mux->programs = g_list_prepend (mux->programs, program);
 463   mux->nb_programs++;
 464   mux->pat_changed = TRUE;
 465
 466   return program;
 467 }
 468
 469 gboolean
 470 tsmux_program_delete (TsMux * mux, TsMuxProgram * program)
 471 {
 472   g_return_val_if_fail (mux != NULL, FALSE);
 473
 474   if (mux->nb_programs == 0)
 475     return FALSE;
 476
 477   if (!program)
 478     return FALSE;
 479
 480   mux->programs = g_list_remove (mux->programs, program);
 481   mux->nb_programs--;
 482   mux->pat_changed = TRUE;
 483   tsmux_program_free ((TsMuxProgram *) program);
 484
 485   return TRUE;
 486 }
 487
 488 /**
 489  * tsmux_set_pmt_interval:
 490  * @program: a #TsMuxProgram
 491  * @freq: a new PMT interval
 492  *
 493  * Set the interval (in cycles of the 90kHz clock) for writing out the PMT table.
 494  *
 495  * Many transport stream clients might have problems if the PMT table is not
 496  * inserted in the stream at regular intervals, especially when initially trying
 497  * to figure out the contents of the stream.
 498  */
 499 void
 500 tsmux_set_pmt_interval (TsMuxProgram * program, guint freq)
 501 {
 502   g_return_if_fail (program != NULL);
 503
 504   program->pmt_interval = freq;
 505 }
 506
 507 /**
 508  * tsmux_get_pmt_interval:
 509  * @program: a #TsMuxProgram
 510  *
 511  * Get the configured PMT interval. See also tsmux_set_pmt_interval().
 512  *
 513  * Returns: the configured PMT interval
 514  */
 515 guint
 516 tsmux_get_pmt_interval (TsMuxProgram * program)
 517 {
 518   g_return_val_if_fail (program != NULL, 0);
 519
 520   return program->pmt_interval;
 521 }
 522
 523 /**
 524  * tsmux_program_set_scte35_interval:
 525  * @program: a #TsMuxProgram
 526  * @freq: a new SCTE-35 NULL interval
 527  *
 528  * Set the interval (in cycles of the 90kHz clock) for sending out the SCTE-35
 529  * NULL command. This is only effective is the SCTE-35 PID is not 0.
 530  */
 531 void
 532 tsmux_program_set_scte35_interval (TsMuxProgram * program, guint interval)
 533 {
 534   g_return_if_fail (program != NULL);
 535
 536   program->scte35_null_interval = interval;
 537 }
 538
 539 /**
 540  * tsmux_resend_pmt:
 541  * @program: a #TsMuxProgram
 542  *
 543  * Resends the PMT before the next stream packet.
 544  */
 545 void
 546 tsmux_resend_pmt (TsMuxProgram * program)
 547 {
 548   g_return_if_fail (program != NULL);
 549
 550   program->next_pmt_pcr = -1;
 551 }
 552
 553 /**
 554  * tsmux_program_set_scte35_pid:
 555  * @program: a #TsMuxProgram
 556  * @pid: The pid to use, or 0 to deactivate usage.
 557  *
 558  * Set the @pid to use for sending SCTE-35 packets on the given
 559  * @program.
 560  *
 561  * This needs to be called as early as possible if SCTE-35 sections
 562  * are even going to be used with the given @program so that the PMT
 563  * can be properly configured.
 564  */
 565 void
 566 tsmux_program_set_scte35_pid (TsMuxProgram * program, guint16 pid)
 567 {
 568   TsMuxSection *section;
 569   GstMpegtsSCTESIT *sit;
 570   g_return_if_fail (program != NULL);
 571
 572   program->scte35_pid = pid;
 573   /* Create/Update the section */
 574   if (program->scte35_null_section) {
 575     tsmux_section_free (program->scte35_null_section);
 576     program->scte35_null_section = NULL;
 577   }
 578   if (pid != 0) {
 579     program->scte35_null_section = section = g_slice_new0 (TsMuxSection);
 580     section->pi.pid = pid;
 581     sit = gst_mpegts_scte_null_new ();
 582     section->section = gst_mpegts_section_from_scte_sit (sit, pid);
 583   }
 584 }
 585
 586 /**
 587  * tsmux_program_get_scte35_pid:
 588  * @program: a #TsMuxProgram
 589  *
 590  * Get the PID configured for sending SCTE-35 packets.
 591  *
 592  * Returns: the configured SCTE-35 PID, or 0 if not active.
 593  */
 594 guint16
 595 tsmux_program_get_scte35_pid (TsMuxProgram * program)
 596 {
 597   g_return_val_if_fail (program != NULL, 0);
 598
 599   return program->scte35_pid;
 600 }
 601
 602 /**
 603  * tsmux_program_add_stream:
 604  * @program: a #TsMuxProgram
 605  * @stream: a #TsMuxStream
 606  *
 607  * Add @stream to @program.
 608  */
 609 void
 610 tsmux_program_add_stream (TsMuxProgram * program, TsMuxStream * stream)
 611 {
 612   GPtrArray *streams;
 613   guint i;
 614   gint pmt_index, array_index = -1 /* append */ ;
 615   guint16 pid;
 616
 617   g_return_if_fail (program != NULL);
 618   g_return_if_fail (stream != NULL);
 619
 620   streams = program->streams;
 621   pmt_index = stream->pmt_index;
 622   pid = tsmux_stream_get_pid (stream);
 623
 624   if (pmt_index >= 0) {
 625     /* Insert into streams with known indices */
 626     for (i = 0; i < streams->len; i++) {
 627       TsMuxStream *s = g_ptr_array_index (streams, i);
 628
 629       if (s->pmt_index < 0 || pmt_index < s->pmt_index) {
 630         array_index = i;
 631         GST_DEBUG ("PID 0x%04x: Using known-order index %d/%u",
 632             pid, array_index, streams->len);
 633         break;
 634       }
 635     }
 636   } else {
 637     /* Insert after streams with known indices, sorted by PID */
 638     for (i = 0; i < streams->len; i++) {
 639       TsMuxStream *s = g_ptr_array_index (streams, i);
 640
 641       if (s->pmt_index < 0 && pid < tsmux_stream_get_pid (s)) {
 642         array_index = i;
 643         GST_DEBUG ("PID 0x%04x: Using PID-order index %d/%u",
 644             pid, array_index, streams->len);
 645         break;
 646       }
 647     }
 648   }
 649
 650   g_ptr_array_insert (streams, array_index, stream);
 651   program->pmt_changed = TRUE;
 652 }
 653
 654 /**
 655  * tsmux_program_set_pcr_stream:
 656  * @program: a #TsMuxProgram
 657  * @stream: a #TsMuxStream
 658  *
 659  * Set @stream as the PCR stream for @program, overwriting the previously
 660  * configured PCR stream. When @stream is NULL, program will have no PCR stream
 661  * configured.
 662  */
 663 void
 664 tsmux_program_set_pcr_stream (TsMuxProgram * program, TsMuxStream * stream)
 665 {
 666   g_return_if_fail (program != NULL);
 667
 668   if (program->pcr_stream == stream)
 669     return;
 670
 671   if (program->pcr_stream != NULL)
 672     tsmux_stream_pcr_unref (program->pcr_stream);
 673   if (stream)
 674     tsmux_stream_pcr_ref (stream);
 675   program->pcr_stream = stream;
 676
 677   program->pmt_changed = TRUE;
 678 }
 679
 680 /**
 681  * tsmux_get_new_pid:
 682  * @mux: a #TsMux
 683  *
 684  * Get a new free PID.
 685  *
 686  * Returns: a new free PID.
 687  */
 688 guint16
 689 tsmux_get_new_pid (TsMux * mux)
 690 {
 691   g_return_val_if_fail (mux != NULL, -1);
 692
 693   /* make sure this PID is free
 694    * (and not taken by a specific earlier request) */
 695   do {
 696     mux->next_stream_pid++;
 697   } while (tsmux_find_stream (mux, mux->next_stream_pid));
 698
 699   return mux->next_stream_pid;
 700 }
 701
 702 /**
 703  * tsmux_create_stream:
 704  * @mux: a #TsMux
 705  * @stream_type: the stream type
 706  * @pid: the PID of the new stream.
 707  *
 708  * Create a new stream of @stream_type in the muxer session @mux.
 709  *
 710  * When @pid is set to #TSMUX_PID_AUTO, a new free PID will automatically
 711  * be allocated for the new stream.
 712  *
 713  * Returns: a new #TsMuxStream.
 714  */
 715 TsMuxStream *
 716 tsmux_create_stream (TsMux * mux, guint stream_type, guint16 pid,
 717     gchar * language)
 718 {
 719   TsMuxStream *stream;
 720   guint16 new_pid;
 721
 722   g_return_val_if_fail (mux != NULL, NULL);
 723   g_return_val_if_fail (mux->new_stream_func != NULL, NULL);
 724
 725   if (pid == TSMUX_PID_AUTO) {
 726     new_pid = tsmux_get_new_pid (mux);
 727   } else {
 728     new_pid = pid & 0x1FFF;
 729   }
 730
 731   /* Ensure we're not creating a PID collision */
 732   if (tsmux_find_stream (mux, new_pid))
 733     return NULL;
 734
 735   stream = mux->new_stream_func (new_pid, stream_type, mux->new_stream_data);
 736
 737   mux->streams = g_list_prepend (mux->streams, stream);
 738   mux->nb_streams++;
 739
 740   if (language) {
 741     strncpy (stream->language, language, 4);
 742     stream->language[3] = 0;
 743   } else {
 744     stream->language[0] = 0;
 745   }
 746
 747   return stream;
 748 }
 749
 750 /**
 751  * tsmux_find_stream:
 752  * @mux: a #TsMux
 753  * @pid: the PID to find.
 754  *
 755  * Find the stream associated with PID.
 756  *
 757  * Returns: a #TsMuxStream with @pid or NULL when the stream was not found.
 758  */
 759 TsMuxStream *
 760 tsmux_find_stream (TsMux * mux, guint16 pid)
 761 {
 762   TsMuxStream *found = NULL;
 763   GList *cur;
 764
 765   g_return_val_if_fail (mux != NULL, NULL);
 766
 767   for (cur = mux->streams; cur; cur = cur->next) {
 768     TsMuxStream *stream = (TsMuxStream *) cur->data;
 769
 770     if (tsmux_stream_get_pid (stream) == pid) {
 771       found = stream;
 772       break;
 773     }
 774   }
 775   return found;
 776 }
 777
 778 static gboolean
 779 tsmux_program_remove_stream (TsMuxProgram * program, TsMuxStream * stream)
 780 {
 781   GPtrArray *streams = program->streams;
 782
 783   if (!g_ptr_array_remove (streams, stream)) {
 784     g_warn_if_reached ();
 785     return FALSE;
 786   }
 787
 788   return streams->len == 0;
 789 }
 790
 791
 792 gboolean
 793 tsmux_remove_stream (TsMux * mux, guint16 pid, TsMuxProgram * program)
 794 {
 795   GList *cur;
 796   gboolean ret = FALSE;
 797
 798   g_return_val_if_fail (mux != NULL, FALSE);
 799
 800   for (cur = mux->streams; cur; cur = cur->next) {
 801     TsMuxStream *stream = (TsMuxStream *) cur->data;
 802
 803     if (tsmux_stream_get_pid (stream) == pid) {
 804       ret = tsmux_program_remove_stream (program, stream);
 805       mux->streams = g_list_remove (mux->streams, stream);
 806       tsmux_stream_free (stream);
 807       break;
 808     }
 809   }
 810
 811   if (ret)
 812     tsmux_program_delete (mux, program);
 813
 814   return ret;
 815 }
 816
 817 static gboolean
 818 tsmux_get_buffer (TsMux * mux, GstBuffer ** buf)
 819 {
 820   g_return_val_if_fail (buf, FALSE);
 821
 822   if (G_UNLIKELY (!mux->alloc_func))
 823     return FALSE;
 824
 825   mux->alloc_func (buf, mux->alloc_func_data);
 826
 827   if (!*buf)
 828     return FALSE;
 829
 830   g_assert (gst_buffer_get_size (*buf) == TSMUX_PACKET_LENGTH);
 831   return TRUE;
 832 }
 833
 834 static gboolean
 835 tsmux_packet_out (TsMux * mux, GstBuffer * buf, gint64 pcr)
 836 {
 837   if (G_UNLIKELY (mux->write_func == NULL)) {
 838     if (buf)
 839       gst_buffer_unref (buf);
 840     return TRUE;
 841   }
 842
 843   if (mux->bitrate) {
 844     GST_BUFFER_PTS (buf) =
 845         gst_util_uint64_scale (mux->n_bytes * 8, GST_SECOND, mux->bitrate);
 846
 847     /* Check and insert a PCR observation for each program if needed,
 848      * but only for programs that have written their SI at least once,
 849      * so the stream starts with PAT/PMT */
 850     if (mux->first_pcr_ts != G_MININT64) {
 851       GList *cur;
 852
 853       for (cur = mux->programs; cur; cur = cur->next) {
 854         TsMuxProgram *program = (TsMuxProgram *) cur->data;
 855         TsMuxStream *stream = program->pcr_stream;
 856         gint64 cur_pcr, next_pcr, new_pcr;
 857
 858         if (!program->wrote_si)
 859           continue;
 860
 861         cur_pcr = get_current_pcr (mux, 0);
 862         next_pcr = get_next_pcr (mux, 0);
 863         new_pcr = write_new_pcr (mux, stream, cur_pcr, next_pcr);
 864
 865         if (new_pcr != -1) {
 866           GstBuffer *buf = NULL;
 867           GstMapInfo map;
 868
 869           if (!tsmux_get_buffer (mux, &buf)) {
 870             goto error;
 871           }
 872
 873           gst_buffer_map (buf, &map, GST_MAP_WRITE);
 874           tsmux_write_ts_header (mux, map.data, &stream->pi, 0, NULL, NULL);
 875           gst_buffer_unmap (buf, &map);
 876
 877           stream->pi.flags &= TSMUX_PACKET_FLAG_PES_FULL_HEADER;
 878           if (!tsmux_packet_out (mux, buf, new_pcr))
 879             goto error;
 880         }
 881       }
 882     }
 883   }
 884
 885   mux->n_bytes += gst_buffer_get_size (buf);
 886
 887   return mux->write_func (buf, mux->write_func_data, pcr);
 888
 889 error:
 890   return FALSE;
 891 }
 892
 893 /*
 894  * adaptation_field() {
 895  *   adaptation_field_length                              8 uimsbf
 896  *   if(adaptation_field_length >0) {
 897  *     discontinuity_indicator                            1 bslbf
 898  *     random_access_indicator                            1 bslbf
 899  *     elementary_stream_priority_indicator               1 bslbf
 900  *     PCR_flag                                           1 bslbf
 901  *     OPCR_flag                                          1 bslbf
 902  *     splicing_point_flag                                1 bslbf
 903  *     transport_private_data_flag                        1 bslbf
 904  *     adaptation_field_extension_flag                    1 bslbf
 905  *     if(PCR_flag == '1') {
 906  *       program_clock_reference_base                    33 uimsbf
 907  *       reserved                                         6 bslbf
 908  *       program_clock_reference_extension                9 uimsbf
 909  *     }
 910  *     if(OPCR_flag == '1') {
 911  *       original_program_clock_reference_base           33 uimsbf
 912  *       reserved                                         6 bslbf
 913  *       original_program_clock_reference_extension       9 uimsbf
 914  *     }
 915  *     if (splicing_point_flag == '1') {
 916  *       splice_countdown                                 8 tcimsbf
 917  *     }
 918  *     if(transport_private_data_flag == '1') {
 919  *       transport_private_data_length                    8 uimsbf
 920  *       for (i=0; i<transport_private_data_length;i++){
 921  *         private_data_byte                              8 bslbf
 922  *       }
 923  *     }
 924  *     if (adaptation_field_extension_flag == '1' ) {
 925  *       adaptation_field_extension_length                8 uimsbf
 926  *       ltw_flag                                         1 bslbf
 927  *       piecewise_rate_flag                              1 bslbf
 928  *       seamless_splice_flag                             1 bslbf
 929  *       reserved                                         5 bslbf
 930  *       if (ltw_flag == '1') {
 931  *         ltw_valid_flag                                 1 bslbf
 932  *         ltw_offset                                    15 uimsbf
 933  *       }
 934  *       if (piecewise_rate_flag == '1') {
 935  *         reserved                                       2 bslbf
 936  *         piecewise_rate                                22 uimsbf
 937  *       }
 938  *       if (seamless_splice_flag == '1'){
 939  *         splice_type                                    4 bslbf
 940  *         DTS_next_AU[32..30]                            3 bslbf
 941  *         marker_bit                                     1 bslbf
 942  *         DTS_next_AU[29..15]                           15 bslbf
 943  *         marker_bit                                     1 bslbf
 944  *         DTS_next_AU[14..0]                            15 bslbf
 945  *         marker_bit                                     1 bslbf
 946  *       }
 947  *       for ( i=0;i<N;i++) {
 948  *         reserved                                       8 bslbf
 949  *       }
 950  *     }
 951  *     for (i=0;i<N;i++){
 952  *       stuffing_byte                                    8 bslbf
 953  *     }
 954  *   }
 955  * }
 956  */
 957 static gboolean
 958 tsmux_write_adaptation_field (guint8 * buf,
 959     TsMuxPacketInfo * pi, guint8 min_length, guint8 * written)
 960 {
 961   guint8 pos = 2;
 962   guint8 flags = 0;
 963
 964   g_assert (min_length <= TSMUX_PAYLOAD_LENGTH);
 965
 966   /* Write out all the fields from the packet info only if the
 967    * user set the flag to request the adaptation field - if the flag
 968    * isn't set, we're just supposed to write stuffing bytes */
 969   if (pi->flags & TSMUX_PACKET_FLAG_ADAPTATION) {
 970     TS_DEBUG ("writing adaptation fields");
 971     if (pi->flags & TSMUX_PACKET_FLAG_DISCONT)
 972       flags |= 0x80;
 973     if (pi->flags & TSMUX_PACKET_FLAG_RANDOM_ACCESS)
 974       flags |= 0x40;
 975     if (pi->flags & TSMUX_PACKET_FLAG_PRIORITY)
 976       flags |= 0x20;
 977     if (pi->flags & TSMUX_PACKET_FLAG_WRITE_PCR) {
 978       guint64 pcr_base;
 979       guint32 pcr_ext;
 980
 981       pcr_base = (pi->pcr / 300);
 982       pcr_ext = (pi->pcr % 300);
 983
 984       flags |= 0x10;
 985       TS_DEBUG ("Writing PCR %" G_GUINT64_FORMAT " + ext %u", pcr_base,
 986           pcr_ext);
 987       buf[pos++] = (pcr_base >> 25) & 0xff;
 988       buf[pos++] = (pcr_base >> 17) & 0xff;
 989       buf[pos++] = (pcr_base >> 9) & 0xff;
 990       buf[pos++] = (pcr_base >> 1) & 0xff;
 991       buf[pos++] = ((pcr_base << 7) & 0x80) | 0x7e | ((pcr_ext >> 8) & 0x01);   /* set 6 reserve bits to 1 */
 992       buf[pos++] = (pcr_ext) & 0xff;
 993     }
 994     if (pi->flags & TSMUX_PACKET_FLAG_WRITE_OPCR) {
 995       guint64 opcr_base;
 996       guint32 opcr_ext;
 997
 998       opcr_base = (pi->opcr / 300);
 999       opcr_ext = (pi->opcr % 300);
1000
1001       flags |= 0x08;
1002       TS_DEBUG ("Writing OPCR");
1003       buf[pos++] = (opcr_base >> 25) & 0xff;
1004       buf[pos++] = (opcr_base >> 17) & 0xff;
1005       buf[pos++] = (opcr_base >> 9) & 0xff;
1006       buf[pos++] = (opcr_base >> 1) & 0xff;
1007       buf[pos++] = ((opcr_base << 7) & 0x80) | 0x7e | ((opcr_ext >> 8) & 0x01); /* set 6 reserve bits to 1 */
1008       buf[pos++] = (opcr_ext) & 0xff;
1009     }
1010     if (pi->flags & TSMUX_PACKET_FLAG_WRITE_SPLICE) {
1011       flags |= 0x04;
1012       buf[pos++] = pi->splice_countdown;
1013     }
1014     if (pi->private_data_len > 0) {
1015       flags |= 0x02;
1016       /* Private data to write, ensure we have enough room */
1017       if ((1 + pi->private_data_len) > (TSMUX_PAYLOAD_LENGTH - pos))
1018         return FALSE;
1019       buf[pos++] = pi->private_data_len;
1020       memcpy (&(buf[pos]), pi->private_data, pi->private_data_len);
1021       pos += pi->private_data_len;
1022       TS_DEBUG ("%u bytes of private data", pi->private_data_len);
1023     }
1024     if (pi->flags & TSMUX_PACKET_FLAG_WRITE_ADAPT_EXT) {
1025       flags |= 0x01;
1026       TS_DEBUG ("FIXME: write Adaptation extension");
1027       /* Write an empty extension for now */
1028       buf[pos++] = 1;
1029       buf[pos++] = 0x1f;        /* lower 5 bits are reserved, and should be all 1 */
1030     }
1031   }
1032   /* Write the flags at the start */
1033   buf[1] = flags;
1034
1035   /* Stuffing bytes if needed */
1036   while (pos < min_length)
1037     buf[pos++] = 0xff;
1038
1039   /* Write the adaptation field length, which doesn't include its own byte */
1040   buf[0] = pos - 1;
1041
1042   if (written)
1043     *written = pos;
1044
1045   return TRUE;
1046 }
1047
1048 static gboolean
1049 tsmux_write_ts_header (TsMux * mux, guint8 * buf, TsMuxPacketInfo * pi,
1050     guint stream_avail, guint * payload_len_out, guint * payload_offset_out)
1051 {
1052   guint8 *tmp;
1053   guint8 adaptation_flag = 0;
1054   guint8 adapt_min_length = 0;
1055   guint8 adapt_len = 0;
1056   guint payload_len;
1057   gboolean write_adapt = FALSE;
1058
1059   /* Sync byte */
1060   buf[0] = TSMUX_SYNC_BYTE;
1061
1062   TS_DEBUG ("PID 0x%04x, counter = 0x%01x, %u bytes avail", pi->pid,
1063       mux->pid_packet_counts[pi->pid] & 0x0f, stream_avail);
1064
1065   /* 3 bits:
1066    *   transport_error_indicator
1067    *   payload_unit_start_indicator
1068    *   transport_priority: (00)
1069    * 13 bits: PID
1070    */
1071   tmp = buf + 1;
1072   if (pi->packet_start_unit_indicator) {
1073     tsmux_put16 (&tmp, 0x4000 | pi->pid);
1074   } else
1075     tsmux_put16 (&tmp, pi->pid);
1076
1077   /* 2 bits: scrambling_control (NOT SUPPORTED) (00)
1078    * 2 bits: adaptation field control (1x has_adaptation_field | x1 has_payload)
1079    * 4 bits: continuity counter (xxxx)
1080    */
1081
1082   if (pi->flags & TSMUX_PACKET_FLAG_ADAPTATION) {
1083     write_adapt = TRUE;
1084   }
1085
1086   if (stream_avail < TSMUX_PAYLOAD_LENGTH) {
1087     /* Need an adaptation field regardless for stuffing */
1088     adapt_min_length = TSMUX_PAYLOAD_LENGTH - stream_avail;
1089     write_adapt = TRUE;
1090   }
1091
1092   if (write_adapt) {
1093     gboolean res;
1094
1095     /* Flag the adaptation field presence */
1096     adaptation_flag |= 0x20;
1097     res = tsmux_write_adaptation_field (buf + TSMUX_HEADER_LENGTH,
1098         pi, adapt_min_length, &adapt_len);
1099     if (G_UNLIKELY (res == FALSE))
1100       return FALSE;
1101
1102     /* Should have written at least the number of bytes we requested */
1103     g_assert (adapt_len >= adapt_min_length);
1104   }
1105
1106   /* The amount of packet data we wrote is the remaining space after
1107    * the adaptation field */
1108   payload_len = TSMUX_PAYLOAD_LENGTH - adapt_len;
1109
1110   if (payload_len_out)
1111     *payload_len_out = payload_len;
1112   else
1113     g_assert (payload_len == 0);
1114
1115   if (payload_offset_out)
1116     *payload_offset_out = TSMUX_HEADER_LENGTH + adapt_len;
1117
1118   /* Now if we are going to write out some payload, flag that fact */
1119   if (payload_len > 0 && stream_avail > 0) {
1120     /* Flag the presence of a payload */
1121     adaptation_flag |= 0x10;
1122
1123     /* We must have enough data to fill the payload, or some calculation
1124      * went wrong */
1125     g_assert (payload_len <= stream_avail);
1126
1127     /* Packet with payload, increment the continuity counter */
1128     mux->pid_packet_counts[pi->pid]++;
1129   }
1130
1131   adaptation_flag |= mux->pid_packet_counts[pi->pid] & 0x0f;
1132
1133   /* Write the byte of transport_scrambling_control, adaptation_field_control
1134    * + continuity counter out */
1135   buf[3] = adaptation_flag;
1136
1137
1138   if (write_adapt) {
1139     TS_DEBUG ("Adaptation field of size >= %d + %d bytes payload",
1140         adapt_len, payload_len);
1141   } else {
1142     TS_DEBUG ("Payload of %d bytes only", payload_len);
1143   }
1144
1145   return TRUE;
1146 }
1147
1148 /* The unused_arg is needed for g_hash_table_foreach() */
1149 static gboolean
1150 tsmux_section_write_packet (gpointer unused_arg,
1151     TsMuxSection * section, TsMux * mux)
1152 {
1153   GstBuffer *section_buffer;
1154   GstBuffer *packet_buffer = NULL;
1155   GstMemory *mem;
1156   guint8 *packet;
1157   guint8 *data;
1158   gsize data_size = 0;
1159   gsize payload_written;
1160   guint len = 0, offset = 0, payload_len = 0;
1161   guint extra_alloc_bytes = 0;
1162
1163   g_return_val_if_fail (section != NULL, FALSE);
1164   g_return_val_if_fail (mux != NULL, FALSE);
1165
1166   /* Mark the start of new PES unit */
1167   section->pi.packet_start_unit_indicator = TRUE;
1168
1169   data = gst_mpegts_section_packetize (section->section, &data_size);
1170
1171   if (!data) {
1172     TS_DEBUG ("Could not packetize section");
1173     return FALSE;
1174   }
1175
1176   /* Mark payload data size */
1177   section->pi.stream_avail = data_size;
1178   payload_written = 0;
1179
1180   /* Wrap section data in a buffer without free function.
1181      The data will be freed when the GstMpegtsSection is destroyed. */
1182   section_buffer = gst_buffer_new_wrapped_full (GST_MEMORY_FLAG_READONLY,
1183       data, data_size, 0, data_size, NULL, NULL);
1184
1185   TS_DEBUG ("Section buffer with size %" G_GSIZE_FORMAT " created",
1186       gst_buffer_get_size (section_buffer));
1187
1188   while (section->pi.stream_avail > 0) {
1189
1190     packet = g_malloc (TSMUX_PACKET_LENGTH);
1191
1192     if (section->pi.packet_start_unit_indicator) {
1193       /* Wee need room for a pointer byte */
1194       section->pi.stream_avail++;
1195
1196       if (!tsmux_write_ts_header (mux, packet, &section->pi,
1197             section->pi.stream_avail, &len, &offset))
1198         goto fail;
1199
1200       /* Write the pointer byte */
1201       packet[offset++] = 0x00;
1202       payload_len = len - 1;
1203
1204     } else {
1205       if (!tsmux_write_ts_header (mux, packet, &section->pi,
1206             section->pi.stream_avail, &len, &offset))
1207         goto fail;
1208       payload_len = len;
1209     }
1210
1211     /* Wrap the TS header and adaption field in a GstMemory */
1212     mem = gst_memory_new_wrapped (GST_MEMORY_FLAG_READONLY,
1213         packet, TSMUX_PACKET_LENGTH, 0, offset, packet, g_free);
1214
1215     TS_DEBUG ("Creating packet buffer at offset "
1216         "%" G_GSIZE_FORMAT " with length %u", payload_written, payload_len);
1217
1218     /* If in M2TS mode, we will need to resize to 4 bytes after the end
1219        of the buffer. For performance reasons, we will now try to include
1220        4 extra bytes from the source buffer, then resize down, to avoid
1221        having an extra 4 byte GstMemory appended. If the source buffer
1222        does not have enough data for this, a new GstMemory will be used */
1223     if (gst_buffer_get_size (section_buffer) - (payload_written +
1224             payload_len) >= 4) {
1225       /* enough space */
1226       extra_alloc_bytes = 4;
1227     } else {
1228       extra_alloc_bytes = 0;
1229     }
1230     packet_buffer = gst_buffer_copy_region (section_buffer, GST_BUFFER_COPY_ALL,
1231         payload_written, payload_len + extra_alloc_bytes);
1232
1233     /* Prepend the header to the section data */
1234     gst_buffer_prepend_memory (packet_buffer, mem);
1235
1236     /* add an extra 4 bytes if it could not be reserved already */
1237     if (extra_alloc_bytes == 4) {
1238       /* we allocated those already, resize */
1239       gst_buffer_set_size (packet_buffer,
1240           gst_buffer_get_size (packet_buffer) - extra_alloc_bytes);
1241     } else {
1242       void *ptr = g_malloc (4);
1243       GstMemory *extra =
1244           gst_memory_new_wrapped (GST_MEMORY_FLAG_READONLY, ptr, 4, 0, 0, ptr,
1245           g_free);
1246       gst_buffer_append_memory (packet_buffer, extra);
1247     }
1248
1249     TS_DEBUG ("Writing %d bytes to section. %d bytes remaining",
1250         len, section->pi.stream_avail - len);
1251
1252     /* Push the packet without PCR */
1253     if (G_UNLIKELY (!tsmux_packet_out (mux, packet_buffer, -1))) {
1254       /* Buffer given away */
1255       packet_buffer = NULL;
1256       goto fail;
1257     }
1258
1259     packet_buffer = NULL;
1260     section->pi.stream_avail -= len;
1261     payload_written += payload_len;
1262     section->pi.packet_start_unit_indicator = FALSE;
1263   }
1264
1265   gst_buffer_unref (section_buffer);
1266
1267   return TRUE;
1268
1269 fail:
1270   g_free (packet);
1271   if (section_buffer)
1272     gst_buffer_unref (section_buffer);
1273   return FALSE;
1274 }
1275
1276 /**
1277  * tsmux_send_section:
1278  * @mux: a #TsMux
1279  * @section: (transfer full): a #GstMpegtsSection to add
1280  *
1281  * Send a @section immediately on the stream.
1282  *
1283  * Returns: %TRUE on success, %FALSE otherwise
1284  */
1285 gboolean
1286 tsmux_send_section (TsMux * mux, GstMpegtsSection * section)
1287 {
1288   gboolean ret;
1289   TsMuxSection tsmux_section;
1290
1291   g_return_val_if_fail (mux != NULL, FALSE);
1292   g_return_val_if_fail (section != NULL, FALSE);
1293
1294   memset (&tsmux_section, 0, sizeof (tsmux_section));
1295
1296   GST_DEBUG ("Sending mpegts section with type %d to mux",
1297       section->section_type);
1298
1299   tsmux_section.section = section;
1300   tsmux_section.pi.pid = section->pid;
1301
1302   ret = tsmux_section_write_packet (NULL, &tsmux_section, mux);
1303   gst_mpegts_section_unref (section);
1304
1305   return ret;
1306 }
1307
1308 static gboolean
1309 tsmux_write_si (TsMux * mux)
1310 {
1311   g_hash_table_foreach (mux->si_sections,
1312       (GHFunc) tsmux_section_write_packet, mux);
1313
1314   mux->si_changed = FALSE;
1315
1316   return TRUE;
1317
1318 }
1319
1320 static void
1321 tsmux_write_null_ts_header (guint8 * buf)
1322 {
1323   *buf++ = TSMUX_SYNC_BYTE;
1324   *buf++ = 0x1f;
1325   *buf++ = 0xff;
1326   *buf++ = 0x10;
1327 }
1328
1329 static gint64
1330 ts_to_pcr (gint64 ts)
1331 {
1332   if (ts == G_MININT64) {
1333     return 0;
1334   }
1335
1336   return (ts - TSMUX_PCR_OFFSET) * (TSMUX_SYS_CLOCK_FREQ / TSMUX_CLOCK_FREQ);
1337 }
1338
1339 /* Calculate the PCR to write into the current packet */
1340 static gint64
1341 get_current_pcr (TsMux * mux, gint64 cur_ts)
1342 {
1343   if (!mux->bitrate)
1344     return ts_to_pcr (cur_ts);
1345
1346   if (mux->first_pcr_ts == G_MININT64) {
1347     g_assert (cur_ts != G_MININT64);
1348     mux->first_pcr_ts = cur_ts;
1349     GST_DEBUG ("First PCR offset is %" G_GUINT64_FORMAT, cur_ts);
1350   }
1351
1352   return ts_to_pcr (mux->first_pcr_ts) +
1353       gst_util_uint64_scale ((mux->n_bytes + PCR_BYTE_OFFSET) * 8,
1354       TSMUX_SYS_CLOCK_FREQ, mux->bitrate);
1355 }
1356
1357 /* Predict the PCR at the next packet if possible */
1358 static gint64
1359 get_next_pcr (TsMux * mux, gint64 cur_ts)
1360 {
1361   if (!mux->bitrate)
1362     return ts_to_pcr (cur_ts);
1363
1364   if (mux->first_pcr_ts == G_MININT64) {
1365     g_assert (cur_ts != G_MININT64);
1366     mux->first_pcr_ts = cur_ts;
1367     GST_DEBUG ("First PCR offset is %" G_GUINT64_FORMAT, cur_ts);
1368   }
1369
1370   return ts_to_pcr (mux->first_pcr_ts) +
1371       gst_util_uint64_scale ((mux->n_bytes + TSMUX_PACKET_LENGTH +
1372           PCR_BYTE_OFFSET) * 8, TSMUX_SYS_CLOCK_FREQ, mux->bitrate);
1373 }
1374
1375 static gint64
1376 write_new_pcr (TsMux * mux, TsMuxStream * stream, gint64 cur_pcr,
1377     gint64 next_pcr)
1378 {
1379   if (stream->next_pcr == -1 || next_pcr > stream->next_pcr) {
1380     stream->pi.flags |=
1381         TSMUX_PACKET_FLAG_ADAPTATION | TSMUX_PACKET_FLAG_WRITE_PCR;
1382     stream->pi.pcr = cur_pcr;
1383
1384     if (mux->bitrate && stream->next_pcr != -1 && cur_pcr >= stream->next_pcr) {
1385       GST_WARNING ("Writing PCR %" G_GUINT64_FORMAT " missed the target %"
1386           G_GUINT64_FORMAT " by %f ms", cur_pcr, stream->next_pcr,
1387           (double) (cur_pcr - stream->next_pcr) / 27000.0);
1388     }
1389     /* Next PCR deadline is now plus the scheduled interval */
1390     stream->next_pcr = cur_pcr + mux->pcr_interval * 300;
1391   } else {
1392     cur_pcr = -1;
1393   }
1394
1395   return cur_pcr;
1396 }
1397
1398 static gboolean
1399 rewrite_si (TsMux * mux, gint64 cur_ts)
1400 {
1401   gboolean write_pat;
1402   gboolean write_si;
1403   GList *cur;
1404   gint64 next_pcr;
1405
1406   next_pcr = get_next_pcr (mux, cur_ts);
1407
1408   /* check if we need to rewrite pat */
1409   if (mux->next_pat_pcr == -1 || mux->pat_changed)
1410     write_pat = TRUE;
1411   else if (next_pcr > mux->next_pat_pcr)
1412     write_pat = TRUE;
1413   else
1414     write_pat = FALSE;
1415
1416   if (write_pat) {
1417     if (mux->next_pat_pcr == -1)
1418       mux->next_pat_pcr = next_pcr + mux->pat_interval * 300;
1419     else
1420       mux->next_pat_pcr += mux->pat_interval * 300;
1421
1422     if (!tsmux_write_pat (mux))
1423       return FALSE;
1424
1425     next_pcr = get_next_pcr (mux, cur_ts);
1426   }
1427
1428   /* check if we need to rewrite sit */
1429   if (mux->next_si_pcr == -1 || mux->si_changed)
1430     write_si = TRUE;
1431   else if (next_pcr > mux->next_si_pcr)
1432     write_si = TRUE;
1433   else
1434     write_si = FALSE;
1435
1436   if (write_si) {
1437     if (mux->next_si_pcr == -1)
1438       mux->next_si_pcr = next_pcr + mux->si_interval * 300;
1439     else
1440       mux->next_si_pcr += mux->si_interval * 300;
1441
1442     if (!tsmux_write_si (mux))
1443       return FALSE;
1444
1445     next_pcr = get_current_pcr (mux, cur_ts);
1446   }
1447
1448   /* check if we need to rewrite any of the current pmts */
1449   for (cur = mux->programs; cur; cur = cur->next) {
1450     TsMuxProgram *program = (TsMuxProgram *) cur->data;
1451     gboolean write_pmt;
1452
1453     if (program->next_pmt_pcr == -1 || program->pmt_changed)
1454       write_pmt = TRUE;
1455     else if (next_pcr > program->next_pmt_pcr)
1456       write_pmt = TRUE;
1457     else
1458       write_pmt = FALSE;
1459
1460     if (write_pmt) {
1461       if (program->next_pmt_pcr == -1)
1462         program->next_pmt_pcr = next_pcr + program->pmt_interval * 300;
1463       else
1464         program->next_pmt_pcr += program->pmt_interval * 300;
1465
1466       if (!tsmux_write_pmt (mux, program))
1467         return FALSE;
1468
1469       next_pcr = get_current_pcr (mux, cur_ts);
1470     }
1471
1472     if (program->scte35_pid != 0) {
1473       gboolean write_scte_null = FALSE;
1474       if (program->next_scte35_pcr == -1)
1475         write_scte_null = TRUE;
1476       else if (next_pcr > program->next_scte35_pcr)
1477         write_scte_null = TRUE;
1478
1479       if (write_scte_null) {
1480         GST_DEBUG ("next scte35 pcr %" G_GINT64_FORMAT,
1481             program->next_scte35_pcr);
1482         if (program->next_scte35_pcr == -1)
1483           program->next_scte35_pcr =
1484               next_pcr + program->scte35_null_interval * 300;
1485         else
1486           program->next_scte35_pcr += program->scte35_null_interval * 300;
1487         GST_DEBUG ("next scte35 NOW pcr %" G_GINT64_FORMAT,
1488             program->next_scte35_pcr);
1489
1490         if (!tsmux_write_scte_null (mux, program))
1491           return FALSE;
1492
1493         next_pcr = get_current_pcr (mux, cur_ts);
1494       }
1495     }
1496
1497     program->wrote_si = TRUE;
1498   }
1499
1500   return TRUE;
1501 }
1502
1503 static gboolean
1504 pad_stream (TsMux * mux, TsMuxStream * stream, gint64 cur_ts)
1505 {
1506   guint64 bitrate;
1507   GstBuffer *buf = NULL;
1508   GstMapInfo map;
1509   gboolean ret = TRUE;
1510   GstClockTimeDiff diff;
1511   guint64 start_n_bytes;
1512
1513   if (!mux->bitrate)
1514     goto done;
1515
1516   if (!GST_CLOCK_STIME_IS_VALID (cur_ts))
1517     goto done;
1518
1519   if (!GST_CLOCK_STIME_IS_VALID (stream->first_ts))
1520     stream->first_ts = cur_ts;
1521
1522   diff = GST_CLOCK_DIFF (stream->first_ts, cur_ts);
1523   if (diff == 0)
1524     goto done;
1525
1526   start_n_bytes = mux->n_bytes;
1527   do {
1528     GST_LOG ("Transport stream bitrate: %" G_GUINT64_FORMAT " over %"
1529         G_GUINT64_FORMAT " bytes, duration %" GST_TIME_FORMAT,
1530         gst_util_uint64_scale (mux->n_bytes * 8, TSMUX_CLOCK_FREQ, diff),
1531         mux->n_bytes, GST_TIME_ARGS (diff * GST_SECOND / TSMUX_CLOCK_FREQ));
1532
1533     /* calculate what the overall bitrate will be if we add 1 more packet */
1534     bitrate =
1535         gst_util_uint64_scale ((mux->n_bytes + TSMUX_PACKET_LENGTH) * 8,
1536         TSMUX_CLOCK_FREQ, diff);
1537
1538     if (bitrate <= mux->bitrate) {
1539       gint64 new_pcr;
1540
1541       if (!tsmux_get_buffer (mux, &buf)) {
1542         ret = FALSE;
1543         goto done;
1544       }
1545
1546       if (!gst_buffer_map (buf, &map, GST_MAP_WRITE)) {
1547         gst_buffer_unref (buf);
1548         ret = FALSE;
1549         goto done;
1550       }
1551
1552       new_pcr = write_new_pcr (mux, stream, get_current_pcr (mux, cur_ts),
1553           get_next_pcr (mux, cur_ts));
1554       if (new_pcr != -1) {
1555         GST_LOG ("Writing PCR-only packet on PID 0x%04x", stream->pi.pid);
1556         tsmux_write_ts_header (mux, map.data, &stream->pi, 0, NULL, NULL);
1557       } else {
1558         GST_LOG ("Writing null stuffing packet");
1559         if (!rewrite_si (mux, cur_ts)) {
1560           gst_buffer_unmap (buf, &map);
1561           gst_buffer_unref (buf);
1562           ret = FALSE;
1563           goto done;
1564         }
1565         tsmux_write_null_ts_header (map.data);
1566         memset (map.data + TSMUX_HEADER_LENGTH, 0xFF, TSMUX_PAYLOAD_LENGTH);
1567       }
1568
1569       gst_buffer_unmap (buf, &map);
1570
1571       stream->pi.flags &= TSMUX_PACKET_FLAG_PES_FULL_HEADER;
1572       ret = tsmux_packet_out (mux, buf, new_pcr);
1573       if (!ret) {
1574         gst_buffer_unref (buf);
1575         goto done;
1576       }
1577     }
1578   } while (bitrate < mux->bitrate);
1579
1580   if (mux->n_bytes != start_n_bytes) {
1581     GST_LOG ("Finished padding the mux");
1582   }
1583
1584 done:
1585   return ret;
1586 }
1587
1588 /**
1589  * tsmux_write_stream_packet:
1590  * @mux: a #TsMux
1591  * @stream: a #TsMuxStream
1592  *
1593  * Write a packet of @stream.
1594  *
1595  * Returns: TRUE if the packet could be written.
1596  */
1597 gboolean
1598 tsmux_write_stream_packet (TsMux * mux, TsMuxStream * stream)
1599 {
1600   guint payload_len, payload_offs;
1601   TsMuxPacketInfo *pi = &stream->pi;
1602   gboolean res;
1603   gint64 new_pcr = -1;
1604   GstBuffer *buf = NULL;
1605   GstMapInfo map;
1606
1607   g_return_val_if_fail (mux != NULL, FALSE);
1608   g_return_val_if_fail (stream != NULL, FALSE);
1609
1610   if (tsmux_stream_is_pcr (stream)) {
1611     gint64 cur_ts = CLOCK_BASE;
1612     if (tsmux_stream_get_dts (stream) != G_MININT64)
1613       cur_ts += tsmux_stream_get_dts (stream);
1614     else
1615       cur_ts += tsmux_stream_get_pts (stream);
1616
1617     if (!rewrite_si (mux, cur_ts))
1618       goto fail;
1619
1620     if (!pad_stream (mux, stream, cur_ts))
1621       goto fail;
1622
1623     new_pcr =
1624         write_new_pcr (mux, stream, get_current_pcr (mux, cur_ts),
1625         get_next_pcr (mux, cur_ts));
1626   }
1627
1628   pi->packet_start_unit_indicator = tsmux_stream_at_pes_start (stream);
1629   if (pi->packet_start_unit_indicator) {
1630     tsmux_stream_initialize_pes_packet (stream);
1631     if (stream->dts != G_MININT64)
1632       stream->dts += CLOCK_BASE;
1633     if (stream->pts != G_MININT64)
1634       stream->pts += CLOCK_BASE;
1635   }
1636   pi->stream_avail = tsmux_stream_bytes_avail (stream);
1637
1638   /* obtain buffer */
1639   if (!tsmux_get_buffer (mux, &buf))
1640     return FALSE;
1641
1642   gst_buffer_map (buf, &map, GST_MAP_WRITE);
1643
1644   if (!tsmux_write_ts_header (mux, map.data, pi, pi->stream_avail, &payload_len,
1645           &payload_offs))
1646     goto fail;
1647
1648
1649   if (!tsmux_stream_get_data (stream, map.data + payload_offs, payload_len))
1650     goto fail;
1651
1652   gst_buffer_unmap (buf, &map);
1653
1654   GST_DEBUG ("Writing PES of size %d", (int) gst_buffer_get_size (buf));
1655   res = tsmux_packet_out (mux, buf, new_pcr);
1656
1657   /* Reset all dynamic flags */
1658   stream->pi.flags &= TSMUX_PACKET_FLAG_PES_FULL_HEADER;
1659
1660   return res;
1661
1662   /* ERRORS */
1663 fail:
1664   {
1665     if (buf) {
1666       gst_buffer_unmap (buf, &map);
1667       gst_buffer_unref (buf);
1668     }
1669     return FALSE;
1670   }
1671 }
1672
1673 /**
1674  * tsmux_program_free:
1675  * @program: a #TsMuxProgram
1676  *
1677  * Free the resources of @program. After this call @program can not be used
1678  * anymore.
1679  */
1680 void
1681 tsmux_program_free (TsMuxProgram * program)
1682 {
1683   g_return_if_fail (program != NULL);
1684
1685   /* Free PMT section */
1686   if (program->pmt.section)
1687     gst_mpegts_section_unref (program->pmt.section);
1688   if (program->scte35_null_section)
1689     tsmux_section_free (program->scte35_null_section);
1690
1691   g_ptr_array_free (program->streams, TRUE);
1692   g_slice_free (TsMuxProgram, program);
1693 }
1694
1695 /**
1696  * tsmux_program_set_pmt_pid:
1697  * @program: A #TsmuxProgram
1698  * @pmt_pid: PID to write PMT for this program
1699  */
1700 void
1701 tsmux_program_set_pmt_pid (TsMuxProgram * program, guint16 pmt_pid)
1702 {
1703   program->pmt_pid = pmt_pid;
1704 }
1705
1706 static gint
1707 compare_program_number (gconstpointer a, gconstpointer b)
1708 {
1709   const GstMpegtsPatProgram *pgm1 = *(const GstMpegtsPatProgram * const *) a;
1710   const GstMpegtsPatProgram *pgm2 = *(const GstMpegtsPatProgram * const *) b;
1711   gint num1 = pgm1->program_number, num2 = pgm2->program_number;
1712
1713   return num1 - num2;
1714 }
1715
1716 static gboolean
1717 tsmux_write_pat (TsMux * mux)
1718 {
1719
1720   if (mux->pat_changed) {
1721     /* program_association_section ()
1722      * for (i = 0; i < N; i++) {
1723      *    program_number                         16   uimsbf
1724      *    reserved                                3   bslbf
1725      *    network_PID_or_program_map_PID         13   uimbsf
1726      * }
1727      * CRC_32                                    32   rbchof
1728      */
1729     GList *cur;
1730     GPtrArray *pat;
1731
1732     pat = gst_mpegts_pat_new ();
1733
1734     for (cur = mux->programs; cur; cur = cur->next) {
1735       GstMpegtsPatProgram *pat_pgm;
1736       TsMuxProgram *program = (TsMuxProgram *) cur->data;
1737
1738       pat_pgm = gst_mpegts_pat_program_new ();
1739       pat_pgm->program_number = program->pgm_number;
1740       pat_pgm->network_or_program_map_PID = program->pmt_pid;
1741
1742       g_ptr_array_add (pat, pat_pgm);
1743     }
1744
1745     g_ptr_array_sort (pat, compare_program_number);
1746
1747     if (mux->pat.section)
1748       gst_mpegts_section_unref (mux->pat.section);
1749
1750     mux->pat.section = gst_mpegts_section_from_pat (pat, mux->transport_id);
1751
1752     mux->pat.section->version_number = mux->pat_version++;
1753
1754     TS_DEBUG ("PAT has %d programs", mux->nb_programs);
1755     mux->pat_changed = FALSE;
1756   }
1757
1758   return tsmux_section_write_packet (NULL, &mux->pat, mux);
1759 }
1760
1761 static gboolean
1762 tsmux_write_pmt (TsMux * mux, TsMuxProgram * program)
1763 {
1764
1765   if (program->pmt_changed) {
1766     /* program_association_section ()
1767      * reserved                                   3   bslbf
1768      * PCR_PID                                   13   uimsbf
1769      * reserved                                   4   bslbf
1770      * program_info_length                       12   uimsbf
1771      * for (i = 0; i < N; i++)
1772      *   descriptor ()
1773      *
1774      * for (i = 0; i < N1; i++) {
1775      *    stream_type                             8   uimsbf
1776      *    reserved                                3   bslbf
1777      *    elementary_PID                         13   uimbsf
1778      *    reserved                                4   bslbf
1779      *    ES_info_length                         12   uimbsf
1780      *    for (i = 0; i < N1; i++) {
1781      *      descriptor ();
1782      *    }
1783      * }
1784      */
1785     GstMpegtsDescriptor *descriptor;
1786     GstMpegtsPMT *pmt;
1787 #if 0
1788     /* See note about bluray descriptors below */
1789     guint8 desc[] = { 0x0F, 0xFF, 0xFC, 0xFC };
1790 #endif
1791     guint i;
1792
1793     pmt = gst_mpegts_pmt_new ();
1794
1795     if (program->pcr_stream == NULL)
1796       pmt->pcr_pid = 0x1FFF;
1797     else
1798       pmt->pcr_pid = tsmux_stream_get_pid (program->pcr_stream);
1799
1800 #if 0
1801     /* FIXME : These two descriptors should not be added in all PMT
1802      * but only in "bluray-compatible" mpeg-ts output. I even have my
1803      * doubt whether the DTCP descriptor is even needed */
1804     descriptor = gst_mpegts_descriptor_from_registration ("HDMV", NULL, 0);
1805     g_ptr_array_add (pmt->descriptors, descriptor);
1806
1807     /* DTCP descriptor, see
1808      * http://www.dtcp.com/documents/dtcp/info-20150204-dtcp-v1-rev%201-71.pdf */
1809     descriptor = gst_mpegts_descriptor_from_custom (0x88, desc, 4);
1810     g_ptr_array_add (pmt->descriptors, descriptor);
1811 #endif
1812
1813     /* Will SCTE-35 be potentially used ? */
1814     if (program->scte35_pid != 0) {
1815       descriptor = gst_mpegts_descriptor_from_registration ("CUEI", NULL, 0);
1816       g_ptr_array_add (pmt->descriptors, descriptor);
1817     }
1818
1819     /* Write out the entries */
1820     for (i = 0; i < program->streams->len; i++) {
1821       GstMpegtsPMTStream *pmt_stream;
1822       TsMuxStream *stream = g_ptr_array_index (program->streams, i);
1823
1824       pmt_stream = gst_mpegts_pmt_stream_new ();
1825
1826       /* FIXME: Use API to retrieve this from the stream */
1827       pmt_stream->stream_type = stream->stream_type;
1828       pmt_stream->pid = tsmux_stream_get_pid (stream);
1829
1830       /* Write any ES descriptors needed */
1831       tsmux_stream_get_es_descrs (stream, pmt_stream);
1832       g_ptr_array_add (pmt->streams, pmt_stream);
1833     }
1834
1835     /* Will SCTE-35 be potentially used ? */
1836     if (program->scte35_pid != 0) {
1837       GstMpegtsPMTStream *pmt_stream = gst_mpegts_pmt_stream_new ();
1838       pmt_stream->stream_type = GST_MPEGTS_STREAM_TYPE_SCTE_SIT;
1839       pmt_stream->pid = program->scte35_pid;
1840       g_ptr_array_add (pmt->streams, pmt_stream);
1841     }
1842
1843     TS_DEBUG ("PMT for program %d has %d streams",
1844         program->pgm_number, program->streams->len);
1845
1846     pmt->program_number = program->pgm_number;
1847
1848     program->pmt.pi.pid = program->pmt_pid;
1849     program->pmt_changed = FALSE;
1850
1851     if (program->pmt.section)
1852       gst_mpegts_section_unref (program->pmt.section);
1853
1854     program->pmt.section = gst_mpegts_section_from_pmt (pmt, program->pmt_pid);
1855     program->pmt.section->version_number = program->pmt_version++;
1856   }
1857
1858   return tsmux_section_write_packet (NULL, &program->pmt, mux);
1859 }
1860
1861 static gboolean
1862 tsmux_write_scte_null (TsMux * mux, TsMuxProgram * program)
1863 {
1864   /* SCTE-35 NULL section is created when PID is set */
1865   GST_LOG ("Writing SCTE NULL packet");
1866   return tsmux_section_write_packet (NULL, program->scte35_null_section, mux);
1867 }
1868
1869 void
1870 tsmux_set_bitrate (TsMux * mux, guint64 bitrate)
1871 {
1872   mux->bitrate = bitrate;
1873 }