3 * Copyright (C) 2009 Zaheer Abbas Merali
5 * Copyright (C) 2011, Hewlett-Packard Development Company, L.P.
6 * Author: Youness Alaoui <youness.alaoui@collabora.co.uk>, Collabora Ltd.
7 * Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>, Collabora Ltd.
8 * Author: Edward Hervey <bilboed@bilboed.com>, Collabora Ltd.
11 * Zaheer Abbas Merali <zaheerabbas at merali dot org>
12 * Edward Hervey <edward.hervey@collabora.co.uk>
14 * This library is free software; you can redistribute it and/or
15 * modify it under the terms of the GNU Library General Public
16 * License as published by the Free Software Foundation; either
17 * version 2 of the License, or (at your option) any later version.
19 * This library is distributed in the hope that it will be useful,
20 * but WITHOUT ANY WARRANTY; without even the implied warranty of
21 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
22 * Library General Public License for more details.
24 * You should have received a copy of the GNU Library General Public
25 * License along with this library; if not, write to the
26 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
27 * Boston, MA 02111-1307, USA.
39 #include "mpegtsbase.h"
41 #include "gstmpegdesc.h"
42 #include "gstmpegdefs.h"
43 #include "mpegtspacketizer.h"
44 #include "payload_parsers.h"
50 * See TODO for explanations on improvements needed
53 /* latency in mseconds */
54 #define TS_LATENCY 700
56 #define TABLE_ID_UNSET 0xFF
58 /* Size of the pendingbuffers array. */
59 #define TS_MAX_PENDING_BUFFERS 256
61 #define PCR_WRAP_SIZE_128KBPS (((gint64)1490)*(1024*1024))
62 /* small PCR for wrap detection */
63 #define PCR_SMALL 17775000
64 /* maximal PCR time */
65 #define PCR_MAX_VALUE (((((guint64)1)<<33) * 300) + 298)
66 #define PTS_DTS_MAX_VALUE (((guint64)1) << 33)
68 /* seek to SEEK_TIMESTAMP_OFFSET before the desired offset and search then
69 * either accurately or for the next timestamp
71 #define SEEK_TIMESTAMP_OFFSET (1000 * GST_MSECOND)
73 GST_DEBUG_CATEGORY_STATIC (ts_demux_debug);
74 #define GST_CAT_DEFAULT ts_demux_debug
76 #define ABSDIFF(a,b) (((a) > (b)) ? ((a) - (b)) : ((b) - (a)))
78 static GQuark QUARK_TSDEMUX;
79 static GQuark QUARK_PID;
80 static GQuark QUARK_PCR;
81 static GQuark QUARK_OPCR;
82 static GQuark QUARK_PTS;
83 static GQuark QUARK_DTS;
84 static GQuark QUARK_OFFSET;
88 PENDING_PACKET_EMPTY = 0, /* No pending packet/buffer
89 * Push incoming buffers to the array */
90 PENDING_PACKET_HEADER, /* PES header needs to be parsed
91 * Push incoming buffers to the array */
92 PENDING_PACKET_BUFFER, /* Currently filling up output buffer
93 * Push incoming buffers to the bufferlist */
94 PENDING_PACKET_DISCONT /* Discontinuity in incoming packets
95 * Drop all incoming buffers */
98 typedef struct _TSDemuxStream TSDemuxStream;
100 struct _TSDemuxStream
102 MpegTSBaseStream stream;
106 /* the return of the latest push */
107 GstFlowReturn flow_return;
110 PendingPacketState state;
111 /* Pending buffers array. */
112 /* These buffers are stored in this array until the PES header (if needed)
113 * is succesfully parsed. */
114 GstBuffer *pendingbuffers[TS_MAX_PENDING_BUFFERS];
117 /* Current data to be pushed out */
118 GstBufferList *current;
119 GstBufferListIterator *currentit;
122 /* Current PTS/DTS for this stream */
125 /* Raw value of current PTS/DTS */
128 /* Number of rollover seen for PTS/DTS (default:0) */
129 guint nb_pts_rollover;
130 guint nb_dts_rollover;
136 "mpegversion = (int) { 1, 2, 4 }, " \
137 "systemstream = (boolean) FALSE; " \
138 "video/x-h264,stream-format=(string)byte-stream," \
139 "alignment=(string)nal;" \
142 "wmvversion = (int) 3, " \
143 "format = (fourcc) WVC1" \
149 "mpegversion = (int) 1;" \
151 "mpegversion = (int) 4, " \
152 "stream-format = (string) adts; " \
154 "width = (int) { 16, 20, 24 }, " \
155 "rate = (int) { 48000, 96000 }, " \
156 "channels = (int) [ 1, 8 ], " \
157 "dynamic_range = (int) [ 0, 255 ], " \
158 "emphasis = (boolean) { FALSE, TRUE }, " \
159 "mute = (boolean) { FALSE, TRUE }; " \
160 "audio/x-ac3; audio/x-eac3;" \
162 "audio/x-private-ts-lpcm" \
165 /* Can also use the subpicture pads for text subtitles? */
166 #define SUBPICTURE_CAPS \
167 GST_STATIC_CAPS ("subpicture/x-pgs; video/x-dvd-subpicture")
169 static GstStaticPadTemplate video_template =
170 GST_STATIC_PAD_TEMPLATE ("video_%04x", GST_PAD_SRC,
174 static GstStaticPadTemplate audio_template =
175 GST_STATIC_PAD_TEMPLATE ("audio_%04x",
180 static GstStaticPadTemplate subpicture_template =
181 GST_STATIC_PAD_TEMPLATE ("subpicture_%04x",
186 static GstStaticPadTemplate private_template =
187 GST_STATIC_PAD_TEMPLATE ("private_%04x",
190 GST_STATIC_CAPS_ANY);
201 static const GstQueryType *gst_ts_demux_srcpad_query_types (GstPad * pad);
202 static gboolean gst_ts_demux_srcpad_query (GstPad * pad, GstQuery * query);
205 /* mpegtsbase methods */
207 gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program);
208 static void gst_ts_demux_reset (MpegTSBase * base);
210 gst_ts_demux_push (MpegTSBase * base, MpegTSPacketizerPacket * packet,
211 MpegTSPacketizerSection * section);
212 static void gst_ts_demux_flush (MpegTSBase * base);
214 gst_ts_demux_stream_added (MpegTSBase * base, MpegTSBaseStream * stream,
215 MpegTSBaseProgram * program);
217 gst_ts_demux_stream_removed (MpegTSBase * base, MpegTSBaseStream * stream);
218 static GstFlowReturn gst_ts_demux_do_seek (MpegTSBase * base, GstEvent * event,
220 static GstFlowReturn find_pcr_packet (MpegTSBase * base, guint64 offset,
221 gint64 length, TSPcrOffset * pcroffset);
222 static GstFlowReturn find_timestamps (MpegTSBase * base, guint64 initoff,
224 static void gst_ts_demux_set_property (GObject * object, guint prop_id,
225 const GValue * value, GParamSpec * pspec);
226 static void gst_ts_demux_get_property (GObject * object, guint prop_id,
227 GValue * value, GParamSpec * pspec);
228 static void gst_ts_demux_finalize (GObject * object);
230 process_pcr (MpegTSBase * base, guint64 initoff, TSPcrOffset * pcroffset,
231 guint numpcr, gboolean isinitial);
232 static void gst_ts_demux_flush_streams (GstTSDemux * tsdemux);
234 gst_ts_demux_push_pending_data (GstTSDemux * demux, TSDemuxStream * stream);
236 static gboolean push_event (MpegTSBase * base, GstEvent * event);
237 static void _extra_init (GType type);
239 GST_BOILERPLATE_FULL (GstTSDemux, gst_ts_demux, MpegTSBase,
240 GST_TYPE_MPEGTS_BASE, _extra_init);
243 _extra_init (GType type)
245 QUARK_TSDEMUX = g_quark_from_string ("tsdemux");
246 QUARK_PID = g_quark_from_string ("pid");
247 QUARK_PCR = g_quark_from_string ("pcr");
248 QUARK_OPCR = g_quark_from_string ("opcr");
249 QUARK_PTS = g_quark_from_string ("pts");
250 QUARK_DTS = g_quark_from_string ("dts");
251 QUARK_OFFSET = g_quark_from_string ("offset");
255 gst_ts_demux_base_init (gpointer klass)
257 GstElementClass *element_class = GST_ELEMENT_CLASS (klass);
259 gst_element_class_add_static_pad_template (element_class,
261 gst_element_class_add_static_pad_template (element_class,
263 gst_element_class_add_static_pad_template (element_class,
264 &subpicture_template);
265 gst_element_class_add_static_pad_template (element_class,
268 gst_element_class_set_details_simple (element_class,
269 "MPEG transport stream demuxer",
271 "Demuxes MPEG2 transport streams",
272 "Zaheer Abbas Merali <zaheerabbas at merali dot org>\n"
273 "Edward Hervey <edward.hervey@collabora.co.uk>");
277 gst_ts_demux_class_init (GstTSDemuxClass * klass)
279 GObjectClass *gobject_class;
280 MpegTSBaseClass *ts_class;
282 gobject_class = G_OBJECT_CLASS (klass);
283 gobject_class->set_property = gst_ts_demux_set_property;
284 gobject_class->get_property = gst_ts_demux_get_property;
285 gobject_class->finalize = gst_ts_demux_finalize;
287 g_object_class_install_property (gobject_class, PROP_PROGRAM_NUMBER,
288 g_param_spec_int ("program-number", "Program number",
289 "Program Number to demux for (-1 to ignore)", -1, G_MAXINT,
290 -1, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
292 g_object_class_install_property (gobject_class, PROP_EMIT_STATS,
293 g_param_spec_boolean ("emit-stats", "Emit statistics",
294 "Emit messages for every pcr/opcr/pts/dts", FALSE,
295 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
298 ts_class = GST_MPEGTS_BASE_CLASS (klass);
299 ts_class->reset = GST_DEBUG_FUNCPTR (gst_ts_demux_reset);
300 ts_class->push = GST_DEBUG_FUNCPTR (gst_ts_demux_push);
301 ts_class->push_event = GST_DEBUG_FUNCPTR (push_event);
302 ts_class->program_started = GST_DEBUG_FUNCPTR (gst_ts_demux_program_started);
303 ts_class->stream_added = gst_ts_demux_stream_added;
304 ts_class->stream_removed = gst_ts_demux_stream_removed;
305 ts_class->find_timestamps = GST_DEBUG_FUNCPTR (find_timestamps);
306 ts_class->seek = GST_DEBUG_FUNCPTR (gst_ts_demux_do_seek);
307 ts_class->flush = GST_DEBUG_FUNCPTR (gst_ts_demux_flush);
311 gst_ts_demux_init (GstTSDemux * demux, GstTSDemuxClass * klass)
313 demux->need_newsegment = TRUE;
314 demux->program_number = -1;
315 demux->duration = GST_CLOCK_TIME_NONE;
316 GST_MPEGTS_BASE (demux)->stream_size = sizeof (TSDemuxStream);
317 gst_segment_init (&demux->segment, GST_FORMAT_TIME);
318 demux->first_pcr = (TSPcrOffset) {
319 GST_CLOCK_TIME_NONE, 0, 0};
320 demux->cur_pcr = (TSPcrOffset) {
322 demux->last_pcr = (TSPcrOffset) {
327 gst_ts_demux_reset (MpegTSBase * base)
329 GstTSDemux *demux = (GstTSDemux *) base;
332 g_array_free (demux->index, TRUE);
335 demux->index_size = 0;
336 demux->need_newsegment = TRUE;
337 demux->program_number = -1;
338 demux->duration = GST_CLOCK_TIME_NONE;
339 gst_segment_init (&demux->segment, GST_FORMAT_TIME);
340 demux->first_pcr = (TSPcrOffset) {
341 GST_CLOCK_TIME_NONE, 0, 0};
342 demux->cur_pcr = (TSPcrOffset) {
344 demux->last_pcr = (TSPcrOffset) {
349 gst_ts_demux_finalize (GObject * object)
351 if (G_OBJECT_CLASS (parent_class)->finalize)
352 G_OBJECT_CLASS (parent_class)->finalize (object);
358 gst_ts_demux_set_property (GObject * object, guint prop_id,
359 const GValue * value, GParamSpec * pspec)
361 GstTSDemux *demux = GST_TS_DEMUX (object);
364 case PROP_PROGRAM_NUMBER:
365 /* FIXME: do something if program is switched as opposed to set at
367 demux->program_number = g_value_get_int (value);
369 case PROP_EMIT_STATS:
370 demux->emit_statistics = g_value_get_boolean (value);
373 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
378 gst_ts_demux_get_property (GObject * object, guint prop_id,
379 GValue * value, GParamSpec * pspec)
381 GstTSDemux *demux = GST_TS_DEMUX (object);
384 case PROP_PROGRAM_NUMBER:
385 g_value_set_int (value, demux->program_number);
387 case PROP_EMIT_STATS:
388 g_value_set_boolean (value, demux->emit_statistics);
391 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
395 static const GstQueryType *
396 gst_ts_demux_srcpad_query_types (GstPad * pad)
398 static const GstQueryType query_types[] = {
408 gst_ts_demux_srcpad_query (GstPad * pad, GstQuery * query)
415 demux = GST_TS_DEMUX (gst_pad_get_parent (pad));
416 base = GST_MPEGTS_BASE (demux);
418 switch (GST_QUERY_TYPE (query)) {
419 case GST_QUERY_DURATION:
420 GST_DEBUG ("query duration");
421 gst_query_parse_duration (query, &format, NULL);
422 if (format == GST_FORMAT_TIME) {
423 if (!gst_pad_peer_query (base->sinkpad, query))
424 gst_query_set_duration (query, GST_FORMAT_TIME,
425 demux->segment.duration);
427 GST_DEBUG_OBJECT (demux, "only query duration on TIME is supported");
431 case GST_QUERY_SEEKING:
432 GST_DEBUG ("query seeking");
433 gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
434 if (format == GST_FORMAT_TIME) {
435 gboolean seekable = FALSE;
437 if (gst_pad_peer_query (base->sinkpad, query))
438 gst_query_parse_seeking (query, NULL, &seekable, NULL, NULL);
440 /* If upstream is not seekable in TIME format we use
441 * our own values here */
443 gst_query_set_seeking (query, GST_FORMAT_TIME,
444 demux->parent.mode != BASE_MODE_PUSHING, 0,
445 demux->segment.duration);
447 GST_DEBUG_OBJECT (demux, "only TIME is supported for query seeking");
452 res = gst_pad_query_default (pad, query);
455 gst_object_unref (demux);
460 static inline GstClockTime
461 calculate_gsttime (TSPcrOffset * start, guint64 pcr)
464 GstClockTime time = start->gsttime;
466 if (start->pcr > pcr)
467 time += PCRTIME_TO_GSTTIME (PCR_MAX_VALUE - start->pcr) +
468 PCRTIME_TO_GSTTIME (pcr);
470 time += PCRTIME_TO_GSTTIME (pcr - start->pcr);
476 gst_ts_demux_parse_pes_header_pts (GstTSDemux * demux,
477 MpegTSPacketizerPacket * packet, guint64 * time)
482 if (mpegts_parse_pes_header (packet->payload,
483 packet->data_end - packet->payload, &header, &offset))
484 return GST_FLOW_ERROR;
490 /* performs a accurate/key_unit seek */
492 gst_ts_demux_perform_auxiliary_seek (MpegTSBase * base, GstClockTime seektime,
493 TSPcrOffset * pcroffset, gint64 length, gint16 pid, GstSeekFlags flags,
494 payload_parse_keyframe auxiliary_seek_fn)
496 GstTSDemux *demux = (GstTSDemux *) base;
497 GstFlowReturn res = GST_FLOW_ERROR;
498 gboolean done = FALSE;
499 gboolean found_keyframe = FALSE, found_accurate = FALSE, need_more = TRUE;
501 MpegTSPacketizerPacket packet;
502 MpegTSPacketizerPacketReturn pret;
503 gint64 offset = pcroffset->offset;
504 gint64 scan_offset = MIN (length, 50 * MPEGTS_MAX_PACKETSIZE);
505 guint32 state = 0xffffffff;
506 TSPcrOffset key_pos = { 0 };
508 GST_DEBUG ("auxiliary seek for %" GST_TIME_FORMAT " from offset: %"
509 G_GINT64_FORMAT " in %" G_GINT64_FORMAT " bytes for PID: %d "
510 "%s %s", GST_TIME_ARGS (seektime), pcroffset->offset, length, pid,
511 (flags & GST_SEEK_FLAG_ACCURATE) ? "accurate" : "",
512 (flags & GST_SEEK_FLAG_KEY_UNIT) ? "key_unit" : "");
514 mpegts_packetizer_flush (base->packetizer);
516 if (base->packetizer->packet_size == MPEGTS_M2TS_PACKETSIZE)
519 while (!done && scan_offset <= length) {
521 gst_pad_pull_range (base->sinkpad, offset + scan_offset,
522 50 * MPEGTS_MAX_PACKETSIZE, &buf);
523 if (res != GST_FLOW_OK)
525 mpegts_packetizer_push (base->packetizer, buf);
529 mpegts_packetizer_next_packet (base->packetizer,
530 &packet)) != PACKET_NEED_MORE)) {
531 if (G_UNLIKELY (pret == PACKET_BAD))
532 /* bad header, skip the packet */
535 if (packet.payload_unit_start_indicator)
536 GST_DEBUG ("found packet for PID: %d with pcr: %" GST_TIME_FORMAT
537 " at offset: %" G_GINT64_FORMAT, packet.pid,
538 GST_TIME_ARGS (packet.pcr), packet.offset);
540 if (packet.payload != NULL && packet.pid == pid) {
542 if (packet.payload_unit_start_indicator) {
545 gst_ts_demux_parse_pes_header_pts (demux, &packet, &pts);
546 if (ok == GST_FLOW_OK) {
547 GstClockTime time = calculate_gsttime (pcroffset, pts * 300);
549 GST_DEBUG ("packet has PTS: %" GST_TIME_FORMAT,
550 GST_TIME_ARGS (time));
552 if (time <= seektime) {
553 pcroffset->gsttime = time;
554 pcroffset->pcr = packet.pcr;
555 pcroffset->offset = packet.offset;
557 found_accurate = TRUE;
560 /* reset state for new packet */
565 if (auxiliary_seek_fn) {
567 if (auxiliary_seek_fn (&state, &packet, &need_more)) {
568 found_keyframe = TRUE;
569 key_pos = *pcroffset;
570 GST_DEBUG ("found keyframe: time: %" GST_TIME_FORMAT " pcr: %"
571 GST_TIME_FORMAT " offset %" G_GINT64_FORMAT,
572 GST_TIME_ARGS (pcroffset->gsttime),
573 GST_TIME_ARGS (pcroffset->pcr), pcroffset->offset);
577 /* if we don't have a payload parsing function
578 * every frame is a keyframe */
579 found_keyframe = TRUE;
582 if (flags & GST_SEEK_FLAG_ACCURATE)
583 done = found_accurate && found_keyframe;
585 done = found_keyframe;
587 *pcroffset = key_pos;
589 mpegts_packetizer_clear_packet (base->packetizer, &packet);
591 scan_offset += 50 * MPEGTS_MAX_PACKETSIZE;
597 else if (GST_FLOW_OK == res)
598 res = GST_FLOW_CUSTOM_ERROR_1;
600 mpegts_packetizer_flush (base->packetizer);
605 TSPcrOffset_find (gconstpointer a, gconstpointer b, gpointer user_data)
608 /* GST_INFO ("a: %" GST_TIME_FORMAT " offset: %" G_GINT64_FORMAT, */
609 /* GST_TIME_ARGS (((TSPcrOffset *) a)->gsttime), ((TSPcrOffset *) a)->offset); */
610 /* GST_INFO ("b: %" GST_TIME_FORMAT " offset: %" G_GINT64_FORMAT, */
611 /* GST_TIME_ARGS (((TSPcrOffset *) b)->gsttime), ((TSPcrOffset *) b)->offset); */
613 if (((TSPcrOffset *) a)->gsttime < ((TSPcrOffset *) b)->gsttime)
615 else if (((TSPcrOffset *) a)->gsttime > ((TSPcrOffset *) b)->gsttime)
622 gst_ts_demux_perform_seek (MpegTSBase * base, GstSegment * segment, guint16 pid)
624 GstTSDemux *demux = (GstTSDemux *) base;
625 GstFlowReturn res = GST_FLOW_ERROR;
626 int max_loop_cnt, loop_cnt = 0;
629 GstClockTime seektime;
630 TSPcrOffset seekpcroffset, pcr_start, pcr_stop, *tmp;
632 max_loop_cnt = (segment->flags & GST_SEEK_FLAG_ACCURATE) ? 25 : 10;
636 segment->last_stop - SEEK_TIMESTAMP_OFFSET) + demux->first_pcr.gsttime;
637 seekpcroffset.gsttime = seektime;
639 GST_DEBUG ("seeking to %" GST_TIME_FORMAT, GST_TIME_ARGS (seektime));
641 gst_ts_demux_flush_streams (demux);
643 if (G_UNLIKELY (!demux->index)) {
644 GST_ERROR ("no index");
648 /* get the first index entry before the seek position */
649 tmp = gst_util_array_binary_search (demux->index->data, demux->index_size,
650 sizeof (*tmp), TSPcrOffset_find, GST_SEARCH_MODE_BEFORE, &seekpcroffset,
653 if (G_UNLIKELY (!tmp)) {
654 GST_ERROR ("value not found");
661 if (G_UNLIKELY (!pcr_stop.offset)) {
662 GST_ERROR ("invalid entry");
666 /* check if the last recorded pcr can be used */
667 if (pcr_start.offset < demux->cur_pcr.offset
668 && demux->cur_pcr.offset < pcr_stop.offset) {
669 demux->cur_pcr.gsttime = calculate_gsttime (&pcr_start, demux->cur_pcr.pcr);
670 if (demux->cur_pcr.gsttime < seekpcroffset.gsttime)
671 pcr_start = demux->cur_pcr;
673 pcr_stop = demux->cur_pcr;
676 GST_DEBUG ("start %" GST_TIME_FORMAT " offset: %" G_GINT64_FORMAT,
677 GST_TIME_ARGS (pcr_start.gsttime), pcr_start.offset);
678 GST_DEBUG ("stop %" GST_TIME_FORMAT " offset: %" G_GINT64_FORMAT,
679 GST_TIME_ARGS (pcr_stop.gsttime), pcr_stop.offset);
681 time_diff = seektime - pcr_start.gsttime;
682 seekpcroffset = pcr_start;
684 GST_DEBUG ("cur %" GST_TIME_FORMAT " offset: %" G_GINT64_FORMAT
685 " time diff: %" G_GINT64_FORMAT,
686 GST_TIME_ARGS (demux->cur_pcr.gsttime), demux->cur_pcr.offset, time_diff);
689 while (loop_cnt++ < max_loop_cnt && (time_diff > SEEK_TIMESTAMP_OFFSET >> 1)
690 && (pcr_stop.gsttime - pcr_start.gsttime > SEEK_TIMESTAMP_OFFSET)) {
691 gint64 duration = pcr_stop.gsttime - pcr_start.gsttime;
692 gint64 size = pcr_stop.offset - pcr_start.offset;
695 seekpos = pcr_start.offset + (size >> 1);
698 pcr_start.offset + size * ((double) (seektime -
699 pcr_start.gsttime) / duration);
701 /* look a litle bit behind */
703 MAX (pcr_start.offset + 188, seekpos - 55 * MPEGTS_MAX_PACKETSIZE);
705 GST_DEBUG ("looking for time: %" GST_TIME_FORMAT " .. %" GST_TIME_FORMAT
706 " .. %" GST_TIME_FORMAT,
707 GST_TIME_ARGS (pcr_start.gsttime),
708 GST_TIME_ARGS (seektime), GST_TIME_ARGS (pcr_stop.gsttime));
709 GST_DEBUG ("looking in bytes: %" G_GINT64_FORMAT " .. %" G_GINT64_FORMAT
710 " .. %" G_GINT64_FORMAT, pcr_start.offset, seekpos, pcr_stop.offset);
713 find_pcr_packet (&demux->parent, seekpos, 4000 * MPEGTS_MAX_PACKETSIZE,
715 if (G_UNLIKELY (res == GST_FLOW_UNEXPECTED)) {
717 MAX ((gint64) pcr_start.offset,
718 seekpos - 2000 * MPEGTS_MAX_PACKETSIZE) + 188;
720 find_pcr_packet (&demux->parent, seekpos,
721 8000 * MPEGTS_MAX_PACKETSIZE, &seekpcroffset);
723 if (G_UNLIKELY (res != GST_FLOW_OK)) {
724 GST_WARNING ("seeking failed %s", gst_flow_get_name (res));
728 seekpcroffset.gsttime = calculate_gsttime (&pcr_start, seekpcroffset.pcr);
731 if (G_UNLIKELY ((seekpcroffset.gsttime < pcr_start.gsttime) ||
732 (seekpcroffset.gsttime > pcr_stop.gsttime))) {
733 GST_ERROR ("Unexpected timestamp found, seeking failed! %"
734 GST_TIME_FORMAT, GST_TIME_ARGS (seekpcroffset.gsttime));
735 res = GST_FLOW_ERROR;
739 if (seekpcroffset.gsttime > seektime) {
740 pcr_stop = seekpcroffset;
742 pcr_start = seekpcroffset;
744 time_diff = seektime - pcr_start.gsttime;
745 GST_DEBUG ("seeking: %" GST_TIME_FORMAT " found: %" GST_TIME_FORMAT
746 " diff = %" G_GINT64_FORMAT, GST_TIME_ARGS (seektime),
747 GST_TIME_ARGS (seekpcroffset.gsttime), time_diff);
750 GST_DEBUG ("seeking finished after %d loops", loop_cnt);
752 /* use correct seek position for the auxiliary search */
753 seektime += SEEK_TIMESTAMP_OFFSET;
756 payload_parse_keyframe keyframe_seek = NULL;
757 MpegTSBaseProgram *program = demux->program;
758 guint64 avg_bitrate, length;
760 if (program->streams[pid]) {
761 switch (program->streams[pid]->stream_type) {
764 keyframe_seek = gst_tsdemux_has_mpeg2_keyframe;
767 keyframe_seek = gst_tsdemux_has_h264_keyframe;
771 GST_WARNING ("no payload parser for stream 0x%04x type: 0x%02x", pid,
772 program->streams[pid]->stream_type);
776 GST_WARNING ("no stream info for PID: 0x%04x", pid);
780 pcr_start.offset) * 1000 * GST_MSECOND / (pcr_stop.gsttime -
783 seekpcroffset = pcr_start;
784 /* search in 2500ms for a keyframe */
786 MIN (demux->last_pcr.offset - pcr_start.offset,
787 (avg_bitrate * 25) / 10);
789 gst_ts_demux_perform_auxiliary_seek (base, seektime, &seekpcroffset,
790 length, pid, segment->flags, keyframe_seek);
792 if (res == GST_FLOW_CUSTOM_ERROR_1) {
793 GST_ERROR ("no keyframe found in %" G_GUINT64_FORMAT
794 " bytes starting from %" G_GUINT64_FORMAT, length,
795 seekpcroffset.offset);
796 res = GST_FLOW_ERROR;
798 if (res != GST_FLOW_OK)
803 /* update seektime to the actual timestamp of the found keyframe */
804 if (segment->flags & GST_SEEK_FLAG_KEY_UNIT)
805 seektime = seekpcroffset.gsttime;
807 seektime -= demux->first_pcr.gsttime;
809 segment->last_stop = seektime;
810 segment->time = seektime;
812 /* we stop at the end */
813 if (segment->stop == -1)
814 segment->stop = demux->first_pcr.gsttime + segment->duration;
816 demux->need_newsegment = TRUE;
817 demux->parent.seek_offset = seekpcroffset.offset;
818 GST_DEBUG ("seeked to postion:%" GST_TIME_FORMAT, GST_TIME_ARGS (seektime));
827 gst_ts_demux_do_seek (MpegTSBase * base, GstEvent * event, guint16 pid)
829 GstTSDemux *demux = (GstTSDemux *) base;
830 GstFlowReturn res = GST_FLOW_ERROR;
834 GstSeekType start_type, stop_type;
836 GstSegment seeksegment;
839 gst_event_parse_seek (event, &rate, &format, &flags, &start_type, &start,
842 if (format != GST_FORMAT_TIME) {
846 GST_DEBUG ("seek event, rate: %f start: %" GST_TIME_FORMAT
847 " stop: %" GST_TIME_FORMAT, rate, GST_TIME_ARGS (start),
848 GST_TIME_ARGS (stop));
850 if (flags & (GST_SEEK_FLAG_SEGMENT | GST_SEEK_FLAG_SKIP)) {
851 GST_WARNING ("seek flags 0x%x are not supported", (int) flags);
855 /* copy segment, we need this because we still need the old
856 * segment when we close the current segment. */
857 memcpy (&seeksegment, &demux->segment, sizeof (GstSegment));
858 /* configure the segment with the seek variables */
859 GST_DEBUG_OBJECT (demux, "configuring seek");
860 GST_DEBUG ("seeksegment: start: %" GST_TIME_FORMAT " stop: %"
861 GST_TIME_FORMAT " time: %" GST_TIME_FORMAT " accum: %" GST_TIME_FORMAT
862 " last_stop: %" GST_TIME_FORMAT " duration: %" GST_TIME_FORMAT,
863 GST_TIME_ARGS (seeksegment.start), GST_TIME_ARGS (seeksegment.stop),
864 GST_TIME_ARGS (seeksegment.time), GST_TIME_ARGS (seeksegment.accum),
865 GST_TIME_ARGS (seeksegment.last_stop),
866 GST_TIME_ARGS (seeksegment.duration));
867 gst_segment_set_seek (&seeksegment, rate, format, flags, start_type, start,
868 stop_type, stop, &update);
869 GST_DEBUG ("seeksegment: start: %" GST_TIME_FORMAT " stop: %"
870 GST_TIME_FORMAT " time: %" GST_TIME_FORMAT " accum: %" GST_TIME_FORMAT
871 " last_stop: %" GST_TIME_FORMAT " duration: %" GST_TIME_FORMAT,
872 GST_TIME_ARGS (seeksegment.start), GST_TIME_ARGS (seeksegment.stop),
873 GST_TIME_ARGS (seeksegment.time), GST_TIME_ARGS (seeksegment.accum),
874 GST_TIME_ARGS (seeksegment.last_stop),
875 GST_TIME_ARGS (seeksegment.duration));
877 res = gst_ts_demux_perform_seek (base, &seeksegment, pid);
878 if (G_UNLIKELY (res != GST_FLOW_OK)) {
879 GST_WARNING ("seeking failed %s", gst_flow_get_name (res));
883 /* commit the new segment */
884 memcpy (&demux->segment, &seeksegment, sizeof (GstSegment));
886 if (demux->segment.flags & GST_SEEK_FLAG_SEGMENT) {
887 gst_element_post_message (GST_ELEMENT_CAST (demux),
888 gst_message_new_segment_start (GST_OBJECT_CAST (demux),
889 demux->segment.format, demux->segment.last_stop));
897 gst_ts_demux_srcpad_event (GstPad * pad, GstEvent * event)
900 GstTSDemux *demux = GST_TS_DEMUX (gst_pad_get_parent (pad));
902 GST_DEBUG_OBJECT (pad, "Got event %s",
903 gst_event_type_get_name (GST_EVENT_TYPE (event)));
905 switch (GST_EVENT_TYPE (event)) {
907 res = mpegts_base_handle_seek_event ((MpegTSBase *) demux, pad, event);
909 GST_WARNING ("seeking failed");
910 gst_event_unref (event);
913 res = gst_pad_event_default (pad, event);
916 gst_object_unref (demux);
921 push_event (MpegTSBase * base, GstEvent * event)
923 GstTSDemux *demux = (GstTSDemux *) base;
926 if (G_UNLIKELY (demux->program == NULL))
929 for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
930 TSDemuxStream *stream = (TSDemuxStream *) tmp->data;
932 gst_event_ref (event);
933 gst_pad_push_event (stream->pad, event);
941 tsdemux_combine_flows (GstTSDemux * demux, TSDemuxStream * stream,
946 /* Store the value */
947 stream->flow_return = ret;
949 /* any other error that is not-linked can be returned right away */
950 if (ret != GST_FLOW_NOT_LINKED)
953 /* Only return NOT_LINKED if all other pads returned NOT_LINKED */
954 for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
955 stream = (TSDemuxStream *) tmp->data;
957 ret = stream->flow_return;
958 /* some other return value (must be SUCCESS but we can return
959 * other values as well) */
960 if (ret != GST_FLOW_NOT_LINKED)
963 /* if we get here, all other pads were unlinked and we return
972 create_pad_for_stream (MpegTSBase * base, MpegTSBaseStream * bstream,
973 MpegTSBaseProgram * program)
975 TSDemuxStream *stream = (TSDemuxStream *) bstream;
977 GstCaps *caps = NULL;
978 GstPadTemplate *template = NULL;
983 GST_LOG ("Attempting to create pad for stream 0x%04x with stream_type %d",
984 bstream->pid, bstream->stream_type);
986 switch (bstream->stream_type) {
989 GST_LOG ("mpeg video");
990 template = gst_static_pad_template_get (&video_template);
991 name = g_strdup_printf ("video_%04x", bstream->pid);
992 caps = gst_caps_new_simple ("video/mpeg",
993 "mpegversion", G_TYPE_INT,
994 bstream->stream_type == ST_VIDEO_MPEG1 ? 1 : 2, "systemstream",
995 G_TYPE_BOOLEAN, FALSE, NULL);
1000 GST_LOG ("mpeg audio");
1001 template = gst_static_pad_template_get (&audio_template);
1002 name = g_strdup_printf ("audio_%04x", bstream->pid);
1004 gst_caps_new_simple ("audio/mpeg", "mpegversion", G_TYPE_INT, 1,
1007 case ST_PRIVATE_DATA:
1008 GST_LOG ("private data");
1010 mpegts_get_descriptor_from_stream ((MpegTSBaseStream *) stream,
1013 GST_LOG ("ac3 audio");
1014 template = gst_static_pad_template_get (&audio_template);
1015 name = g_strdup_printf ("audio_%04x", bstream->pid);
1016 caps = gst_caps_new_simple ("audio/x-ac3", NULL);
1021 mpegts_get_descriptor_from_stream ((MpegTSBaseStream *) stream,
1022 DESC_DVB_ENHANCED_AC3);
1024 GST_LOG ("ac3 audio");
1025 template = gst_static_pad_template_get (&audio_template);
1026 name = g_strdup_printf ("audio_%04x", bstream->pid);
1027 caps = gst_caps_new_simple ("audio/x-eac3", NULL);
1032 mpegts_get_descriptor_from_stream ((MpegTSBaseStream *) stream,
1035 GST_LOG ("teletext");
1036 template = gst_static_pad_template_get (&private_template);
1037 name = g_strdup_printf ("private_%04x", bstream->pid);
1038 caps = gst_caps_new_simple ("private/teletext", NULL);
1043 mpegts_get_descriptor_from_stream ((MpegTSBaseStream *) stream,
1044 DESC_DVB_SUBTITLING);
1046 GST_LOG ("subtitling");
1047 template = gst_static_pad_template_get (&private_template);
1048 name = g_strdup_printf ("private_%04x", bstream->pid);
1049 caps = gst_caps_new_simple ("subpicture/x-dvb", NULL);
1053 /* hack for itv hd (sid 10510, video pid 3401 */
1054 if (program->program_number == 10510 && bstream->pid == 3401) {
1055 template = gst_static_pad_template_get (&video_template);
1056 name = g_strdup_printf ("video_%04x", bstream->pid);
1057 caps = gst_caps_new_simple ("video/x-h264",
1058 "stream-format", G_TYPE_STRING, "byte-stream",
1059 "alignment", G_TYPE_STRING, "nal", NULL);
1063 /* We don't expose those streams since they're only helper streams */
1064 /* template = gst_static_pad_template_get (&private_template); */
1065 /* name = g_strdup_printf ("private_%04x", bstream->pid); */
1066 /* caps = gst_caps_new_simple ("hdv/aux-v", NULL); */
1069 /* We don't expose those streams since they're only helper streams */
1070 /* template = gst_static_pad_template_get (&private_template); */
1071 /* name = g_strdup_printf ("private_%04x", bstream->pid); */
1072 /* caps = gst_caps_new_simple ("hdv/aux-a", NULL); */
1074 case ST_PRIVATE_SECTIONS:
1081 MPEGTS_BIT_UNSET (base->is_pes, bstream->pid);
1083 case ST_AUDIO_AAC: /* ADTS */
1084 template = gst_static_pad_template_get (&audio_template);
1085 name = g_strdup_printf ("audio_%04x", bstream->pid);
1086 caps = gst_caps_new_simple ("audio/mpeg",
1087 "mpegversion", G_TYPE_INT, 4,
1088 "stream-format", G_TYPE_STRING, "adts", NULL);
1090 case ST_VIDEO_MPEG4:
1091 template = gst_static_pad_template_get (&video_template);
1092 name = g_strdup_printf ("video_%04x", bstream->pid);
1093 caps = gst_caps_new_simple ("video/mpeg",
1094 "mpegversion", G_TYPE_INT, 4,
1095 "systemstream", G_TYPE_BOOLEAN, FALSE, NULL);
1098 template = gst_static_pad_template_get (&video_template);
1099 name = g_strdup_printf ("video_%04x", bstream->pid);
1100 caps = gst_caps_new_simple ("video/x-h264",
1101 "stream-format", G_TYPE_STRING, "byte-stream",
1102 "alignment", G_TYPE_STRING, "nal", NULL);
1104 case ST_VIDEO_DIRAC:
1106 mpegts_get_descriptor_from_stream ((MpegTSBaseStream *) stream,
1109 if (DESC_LENGTH (desc) >= 4) {
1110 if (DESC_REGISTRATION_format_identifier (desc) == 0x64726163) {
1113 template = gst_static_pad_template_get (&video_template);
1114 name = g_strdup_printf ("video_%04x", bstream->pid);
1115 caps = gst_caps_new_simple ("video/x-dirac", NULL);
1121 case ST_PRIVATE_EA: /* Try to detect a VC1 stream */
1124 mpegts_get_descriptor_from_stream ((MpegTSBaseStream *) stream,
1127 if (DESC_LENGTH (desc) >= 4) {
1128 if (DESC_REGISTRATION_format_identifier (desc) == DRF_ID_VC1) {
1129 GST_WARNING ("0xea private stream type found but no descriptor "
1130 "for VC1. Assuming plain VC1.");
1131 template = gst_static_pad_template_get (&video_template);
1132 name = g_strdup_printf ("video_%04x", bstream->pid);
1133 caps = gst_caps_new_simple ("video/x-wmv",
1134 "wmvversion", G_TYPE_INT, 3,
1135 "format", GST_TYPE_FOURCC, GST_MAKE_FOURCC ('W', 'V', 'C', '1'),
1143 case ST_BD_AUDIO_AC3:
1145 /* REGISTRATION DRF_ID_HDMV */
1146 desc = mpegts_get_descriptor_from_program (program, DESC_REGISTRATION);
1148 if (DESC_REGISTRATION_format_identifier (desc) == DRF_ID_HDMV) {
1149 template = gst_static_pad_template_get (&audio_template);
1150 name = g_strdup_printf ("audio_%04x", bstream->pid);
1151 caps = gst_caps_new_simple ("audio/x-eac3", NULL);
1158 /* DVB_ENHANCED_AC3 */
1160 mpegts_get_descriptor_from_stream ((MpegTSBaseStream *) stream,
1161 DESC_DVB_ENHANCED_AC3);
1163 template = gst_static_pad_template_get (&audio_template);
1164 name = g_strdup_printf ("audio_%04x", bstream->pid);
1165 caps = gst_caps_new_simple ("audio/x-eac3", NULL);
1172 mpegts_get_descriptor_from_stream ((MpegTSBaseStream *) stream,
1175 GST_WARNING ("AC3 stream type found but no corresponding "
1176 "descriptor to differentiate between AC3 and EAC3. "
1177 "Assuming plain AC3.");
1180 template = gst_static_pad_template_get (&audio_template);
1181 name = g_strdup_printf ("audio_%04x", bstream->pid);
1182 caps = gst_caps_new_simple ("audio/x-ac3", NULL);
1185 case ST_BD_AUDIO_EAC3:
1186 template = gst_static_pad_template_get (&audio_template);
1187 name = g_strdup_printf ("audio_%04x", bstream->pid);
1188 caps = gst_caps_new_simple ("audio/x-eac3", NULL);
1190 case ST_PS_AUDIO_DTS:
1191 template = gst_static_pad_template_get (&audio_template);
1192 name = g_strdup_printf ("audio_%04x", bstream->pid);
1193 caps = gst_caps_new_simple ("audio/x-dts", NULL);
1195 case ST_PS_AUDIO_LPCM:
1196 template = gst_static_pad_template_get (&audio_template);
1197 name = g_strdup_printf ("audio_%04x", bstream->pid);
1198 caps = gst_caps_new_simple ("audio/x-lpcm", NULL);
1200 case ST_BD_AUDIO_LPCM:
1201 template = gst_static_pad_template_get (&audio_template);
1202 name = g_strdup_printf ("audio_%04x", bstream->pid);
1203 caps = gst_caps_new_simple ("audio/x-private-ts-lpcm", NULL);
1205 case ST_PS_DVD_SUBPICTURE:
1206 template = gst_static_pad_template_get (&subpicture_template);
1207 name = g_strdup_printf ("subpicture_%04x", bstream->pid);
1208 caps = gst_caps_new_simple ("video/x-dvd-subpicture", NULL);
1210 case ST_BD_PGS_SUBPICTURE:
1211 template = gst_static_pad_template_get (&subpicture_template);
1212 name = g_strdup_printf ("subpicture_%04x", bstream->pid);
1213 caps = gst_caps_new_simple ("subpicture/x-pgs", NULL);
1216 GST_WARNING ("Non-media stream (stream_type:0x%x). Not creating pad",
1217 bstream->stream_type);
1220 if (template && name && caps) {
1221 GST_LOG ("stream:%p creating pad with name %s and caps %s", stream, name,
1222 gst_caps_to_string (caps));
1223 pad = gst_pad_new_from_template (template, name);
1224 gst_pad_use_fixed_caps (pad);
1225 gst_pad_set_caps (pad, caps);
1226 gst_pad_set_query_type_function (pad, gst_ts_demux_srcpad_query_types);
1227 gst_pad_set_query_function (pad, gst_ts_demux_srcpad_query);
1228 gst_pad_set_event_function (pad, gst_ts_demux_srcpad_event);
1229 gst_caps_unref (caps);
1238 gst_ts_demux_stream_added (MpegTSBase * base, MpegTSBaseStream * bstream,
1239 MpegTSBaseProgram * program)
1241 TSDemuxStream *stream = (TSDemuxStream *) bstream;
1244 /* Create the pad */
1245 if (bstream->stream_type != 0xff)
1246 stream->pad = create_pad_for_stream (base, bstream, program);
1248 stream->pts = GST_CLOCK_TIME_NONE;
1249 stream->dts = GST_CLOCK_TIME_NONE;
1250 stream->raw_pts = 0;
1251 stream->raw_dts = 0;
1252 stream->nb_pts_rollover = 0;
1253 stream->nb_dts_rollover = 0;
1255 stream->flow_return = GST_FLOW_OK;
1259 gst_ts_demux_stream_removed (MpegTSBase * base, MpegTSBaseStream * bstream)
1261 GstTSDemux *demux = GST_TS_DEMUX (base);
1262 TSDemuxStream *stream = (TSDemuxStream *) bstream;
1265 if (gst_pad_is_active (stream->pad)) {
1266 gboolean need_newsegment = demux->need_newsegment;
1268 /* We must not send the newsegment when flushing the pending data
1269 on the removed stream. We should only push it when the newly added
1270 stream finishes parsing its PTS */
1271 demux->need_newsegment = FALSE;
1273 /* Flush out all data */
1274 GST_DEBUG_OBJECT (stream->pad, "Flushing out pending data");
1275 gst_ts_demux_push_pending_data ((GstTSDemux *) base, stream);
1277 demux->need_newsegment = need_newsegment;
1279 GST_DEBUG_OBJECT (stream->pad, "Pushing out EOS");
1280 gst_pad_push_event (stream->pad, gst_event_new_eos ());
1281 GST_DEBUG_OBJECT (stream->pad, "Deactivating and removing pad");
1282 gst_pad_set_active (stream->pad, FALSE);
1283 gst_element_remove_pad (GST_ELEMENT_CAST (base), stream->pad);
1287 stream->flow_return = GST_FLOW_NOT_LINKED;
1291 activate_pad_for_stream (GstTSDemux * tsdemux, TSDemuxStream * stream)
1294 GST_DEBUG_OBJECT (tsdemux, "Activating pad %s:%s for stream %p",
1295 GST_DEBUG_PAD_NAME (stream->pad), stream);
1296 gst_pad_set_active (stream->pad, TRUE);
1297 gst_element_add_pad ((GstElement *) tsdemux, stream->pad);
1298 GST_DEBUG_OBJECT (stream->pad, "done adding pad");
1300 GST_WARNING_OBJECT (tsdemux,
1301 "stream %p (pid 0x%04x, type:0x%03x) has no pad", stream,
1302 ((MpegTSBaseStream *) stream)->pid,
1303 ((MpegTSBaseStream *) stream)->stream_type);
1307 gst_ts_demux_stream_flush (TSDemuxStream * stream)
1311 stream->pts = GST_CLOCK_TIME_NONE;
1313 for (i = 0; i < stream->nbpending; i++)
1314 gst_buffer_unref (stream->pendingbuffers[i]);
1315 memset (stream->pendingbuffers, 0, TS_MAX_PENDING_BUFFERS);
1316 stream->nbpending = 0;
1318 stream->current = NULL;
1322 gst_ts_demux_flush_streams (GstTSDemux * demux)
1324 g_list_foreach (demux->program->stream_list,
1325 (GFunc) gst_ts_demux_stream_flush, NULL);
1329 gst_ts_demux_program_started (MpegTSBase * base, MpegTSBaseProgram * program)
1331 GstTSDemux *demux = GST_TS_DEMUX (base);
1333 GST_DEBUG ("Current program %d, new program %d",
1334 demux->program_number, program->program_number);
1336 if (demux->program_number == -1 ||
1337 demux->program_number == program->program_number) {
1340 GST_LOG ("program %d started", program->program_number);
1341 demux->program_number = program->program_number;
1342 demux->program = program;
1344 /* Activate all stream pads, pads will already have been created */
1345 if (base->mode != BASE_MODE_SCANNING) {
1346 for (tmp = program->stream_list; tmp; tmp = tmp->next)
1347 activate_pad_for_stream (demux, (TSDemuxStream *) tmp->data);
1348 gst_element_no_more_pads ((GstElement *) demux);
1351 /* Inform scanner we have got our program */
1352 demux->current_program_number = program->program_number;
1353 demux->need_newsegment = TRUE;
1358 process_section (MpegTSBase * base)
1360 GstTSDemux *demux = GST_TS_DEMUX (base);
1362 gboolean done = FALSE;
1363 MpegTSPacketizerPacket packet;
1364 MpegTSPacketizerPacketReturn pret;
1368 mpegts_packetizer_next_packet (base->packetizer,
1369 &packet)) != PACKET_NEED_MORE)) {
1370 if (G_UNLIKELY (pret == PACKET_BAD))
1371 /* bad header, skip the packet */
1375 if (packet.payload != NULL && mpegts_base_is_psi (base, &packet)) {
1376 MpegTSPacketizerSection section;
1379 mpegts_packetizer_push_section (base->packetizer, &packet, §ion);
1380 if (G_UNLIKELY (!based))
1381 /* bad section data */
1384 if (G_LIKELY (section.complete)) {
1385 /* section complete */
1386 GST_DEBUG ("Section Complete");
1387 based = mpegts_base_handle_psi (base, §ion);
1388 gst_buffer_unref (section.buffer);
1389 if (G_UNLIKELY (!based))
1395 if (demux->program != NULL) {
1396 GST_DEBUG ("Got Program");
1401 mpegts_packetizer_clear_packet (base->packetizer, &packet);
1407 process_pes (MpegTSBase * base, TSPcrOffset * pcroffset)
1409 gboolean based, done = FALSE;
1410 MpegTSPacketizerPacket packet;
1411 MpegTSPacketizerPacketReturn pret;
1412 GstTSDemux *demux = GST_TS_DEMUX (base);
1413 guint16 pcr_pid = 0;
1417 mpegts_packetizer_next_packet (base->packetizer,
1418 &packet)) != PACKET_NEED_MORE)) {
1419 if (G_UNLIKELY (pret == PACKET_BAD))
1420 /* bad header, skip the packet */
1423 if (demux->program != NULL) {
1424 pcr_pid = demux->program->pcr_pid;
1428 if (packet.payload != NULL && mpegts_base_is_psi (base, &packet)) {
1429 MpegTSPacketizerSection section;
1432 mpegts_packetizer_push_section (base->packetizer, &packet, §ion);
1433 if (G_UNLIKELY (!based))
1434 /* bad section data */
1437 if (G_LIKELY (section.complete)) {
1438 /* section complete */
1439 GST_DEBUG ("Section Complete");
1440 based = mpegts_base_handle_psi (base, §ion);
1441 gst_buffer_unref (section.buffer);
1442 if (G_UNLIKELY (!based))
1448 if (packet.pid == pcr_pid && (packet.adaptation_field_control & 0x02)
1449 && (packet.afc_flags & MPEGTS_AFC_PCR_FLAG)) {
1450 GST_DEBUG ("PCR[0x%x]: %" G_GINT64_FORMAT, packet.pid, packet.pcr);
1451 pcroffset->pcr = packet.pcr;
1452 pcroffset->offset = packet.offset;
1456 mpegts_packetizer_clear_packet (base->packetizer, &packet);
1461 static GstFlowReturn
1462 find_pcr_packet (MpegTSBase * base, guint64 offset, gint64 length,
1463 TSPcrOffset * pcroffset)
1465 GstFlowReturn ret = GST_FLOW_OK;
1466 GstTSDemux *demux = GST_TS_DEMUX (base);
1467 MpegTSBaseProgram *program;
1469 gboolean done = FALSE;
1470 guint64 scan_offset = 0;
1472 GST_DEBUG ("Scanning for PCR between:%" G_GINT64_FORMAT
1473 " and the end:%" G_GINT64_FORMAT, offset, offset + length);
1475 /* Get the program */
1476 program = demux->program;
1477 if (G_UNLIKELY (program == NULL))
1478 return GST_FLOW_ERROR;
1480 mpegts_packetizer_flush (base->packetizer);
1481 if (offset >= 4 && base->packetizer->packet_size == MPEGTS_M2TS_PACKETSIZE)
1484 while (!done && scan_offset < length) {
1486 gst_pad_pull_range (base->sinkpad, offset + scan_offset,
1487 50 * MPEGTS_MAX_PACKETSIZE, &buf);
1488 if (ret != GST_FLOW_OK)
1490 mpegts_packetizer_push (base->packetizer, buf);
1491 done = process_pes (base, pcroffset);
1492 scan_offset += 50 * MPEGTS_MAX_PACKETSIZE;
1495 if (!done || scan_offset >= length) {
1496 GST_WARNING ("No PCR found!");
1497 ret = GST_FLOW_ERROR;
1502 mpegts_packetizer_flush (base->packetizer);
1507 verify_timestamps (MpegTSBase * base, TSPcrOffset * first, TSPcrOffset * last)
1509 GstTSDemux *demux = GST_TS_DEMUX (base);
1510 guint64 length = 4000 * MPEGTS_MAX_PACKETSIZE;
1511 guint64 offset = PCR_WRAP_SIZE_128KBPS;
1514 g_array_sized_new (TRUE, TRUE, sizeof (*first),
1515 2 + 1 + ((last->offset - first->offset) / PCR_WRAP_SIZE_128KBPS));
1517 first->gsttime = PCRTIME_TO_GSTTIME (first->pcr);
1518 demux->index = g_array_append_val (demux->index, *first);
1519 demux->index_size++;
1520 demux->first_pcr = *first;
1521 demux->index_pcr = *first;
1522 GST_DEBUG ("first time: %" GST_TIME_FORMAT " pcr: %" GST_TIME_FORMAT
1523 " offset: %" G_GINT64_FORMAT
1524 " last pcr: %" GST_TIME_FORMAT " offset: %" G_GINT64_FORMAT,
1525 GST_TIME_ARGS (first->gsttime),
1526 GST_TIME_ARGS (PCRTIME_TO_GSTTIME (first->pcr)), first->offset,
1527 GST_TIME_ARGS (PCRTIME_TO_GSTTIME (last->pcr)), last->offset);
1529 while (offset + length < last->offset) {
1535 ret = find_pcr_packet (base, offset, length, &half);
1536 if (G_UNLIKELY (ret != GST_FLOW_OK)) {
1537 GST_WARNING ("no pcr found, retrying");
1546 half.gsttime = calculate_gsttime (first, half.pcr);
1548 GST_DEBUG ("add half time: %" GST_TIME_FORMAT " pcr: %" GST_TIME_FORMAT
1549 " offset: %" G_GINT64_FORMAT,
1550 GST_TIME_ARGS (half.gsttime),
1551 GST_TIME_ARGS (PCRTIME_TO_GSTTIME (half.pcr)), half.offset);
1552 demux->index = g_array_append_val (demux->index, half);
1553 demux->index_size++;
1555 length = 4000 * MPEGTS_MAX_PACKETSIZE;
1556 offset += PCR_WRAP_SIZE_128KBPS;
1560 last->gsttime = calculate_gsttime (first, last->pcr);
1562 GST_DEBUG ("add last time: %" GST_TIME_FORMAT " pcr: %" GST_TIME_FORMAT
1563 " offset: %" G_GINT64_FORMAT,
1564 GST_TIME_ARGS (last->gsttime),
1565 GST_TIME_ARGS (PCRTIME_TO_GSTTIME (last->pcr)), last->offset);
1567 demux->index = g_array_append_val (demux->index, *last);
1568 demux->index_size++;
1570 demux->last_pcr = *last;
1574 static GstFlowReturn
1575 find_timestamps (MpegTSBase * base, guint64 initoff, guint64 * offset)
1578 GstFlowReturn ret = GST_FLOW_OK;
1580 gboolean done = FALSE;
1581 GstFormat format = GST_FORMAT_BYTES;
1583 guint64 scan_offset;
1585 TSPcrOffset initial, final;
1586 GstTSDemux *demux = GST_TS_DEMUX (base);
1588 GST_DEBUG ("Scanning for timestamps");
1590 /* Flush what remained from before */
1591 mpegts_packetizer_clear (base->packetizer);
1593 /* Start scanning from know PAT offset */
1596 gst_pad_pull_range (base->sinkpad, i * 50 * MPEGTS_MAX_PACKETSIZE,
1597 50 * MPEGTS_MAX_PACKETSIZE, &buf);
1598 if (ret != GST_FLOW_OK)
1600 mpegts_packetizer_push (base->packetizer, buf);
1601 done = process_section (base);
1604 mpegts_packetizer_clear (base->packetizer);
1609 *offset = base->seek_offset;
1611 /* Search for the first PCRs */
1612 ret = process_pcr (base, base->first_pat_offset, &initial, 10, TRUE);
1614 if (ret != GST_FLOW_OK && ret != GST_FLOW_UNEXPECTED) {
1615 GST_WARNING ("Problem getting initial PCRs");
1619 mpegts_packetizer_clear (base->packetizer);
1620 /* Remove current program so we ensure looking for a PAT when scanning the
1621 * for the final PCR */
1622 gst_structure_free (base->pat);
1624 mpegts_base_remove_program (base, demux->current_program_number);
1625 demux->program = NULL;
1627 /* Find end position */
1628 if (G_UNLIKELY (!gst_pad_query_peer_duration (base->sinkpad, &format,
1629 &total_bytes) || format != GST_FORMAT_BYTES)) {
1630 GST_WARNING_OBJECT (base, "Couldn't get upstream size in bytes");
1631 ret = GST_FLOW_ERROR;
1632 mpegts_packetizer_clear (base->packetizer);
1635 GST_DEBUG ("Upstream is %" G_GINT64_FORMAT " bytes", total_bytes);
1638 /* Let's start scanning 4000 packets from the end */
1639 scan_offset = MAX (188, total_bytes - 4000 * MPEGTS_MAX_PACKETSIZE);
1641 GST_DEBUG ("Scanning for last sync point between:%" G_GINT64_FORMAT
1642 " and the end:%" G_GINT64_FORMAT, scan_offset, total_bytes);
1643 while ((!done) && (scan_offset < total_bytes)) {
1645 gst_pad_pull_range (base->sinkpad,
1646 scan_offset, 50 * MPEGTS_MAX_PACKETSIZE, &buf);
1647 if (ret != GST_FLOW_OK)
1650 mpegts_packetizer_push (base->packetizer, buf);
1651 done = process_section (base);
1652 scan_offset += 50 * MPEGTS_MAX_PACKETSIZE;
1655 mpegts_packetizer_clear (base->packetizer);
1657 GST_DEBUG ("Searching PCR");
1659 process_pcr (base, scan_offset - 50 * MPEGTS_MAX_PACKETSIZE, &final, 10,
1662 if (ret != GST_FLOW_OK) {
1663 GST_DEBUG ("Problem getting last PCRs");
1667 verify_timestamps (base, &initial, &final);
1669 gst_segment_set_duration (&demux->segment, GST_FORMAT_TIME,
1670 demux->last_pcr.gsttime - demux->first_pcr.gsttime);
1671 demux->duration = demux->last_pcr.gsttime - demux->first_pcr.gsttime;
1672 GST_DEBUG ("Done, duration:%" GST_TIME_FORMAT,
1673 GST_TIME_ARGS (demux->duration));
1677 mpegts_packetizer_clear (base->packetizer);
1678 /* Remove current program */
1680 gst_structure_free (base->pat);
1683 mpegts_base_remove_program (base, demux->current_program_number);
1684 demux->program = NULL;
1689 static GstFlowReturn
1690 process_pcr (MpegTSBase * base, guint64 initoff, TSPcrOffset * pcroffset,
1691 guint numpcr, gboolean isinitial)
1693 GstTSDemux *demux = GST_TS_DEMUX (base);
1694 GstFlowReturn ret = GST_FLOW_OK;
1695 MpegTSBaseProgram *program;
1698 guint32 pcrmask, pcrpattern;
1700 guint64 pcroffs[50];
1703 GST_DEBUG ("initoff:%" G_GUINT64_FORMAT ", numpcr:%d, isinitial:%d",
1704 initoff, numpcr, isinitial);
1706 /* Get the program */
1707 program = demux->program;
1708 if (G_UNLIKELY (program == NULL))
1709 return GST_FLOW_ERROR;
1711 /* First find the first X PCR */
1713 /* Mask/pattern is PID:PCR_PID, AFC&0x02 */
1714 /* sync_byte (0x47) : 8bits => 0xff
1715 * transport_error_indicator : 1bit ACTIVATE
1716 * payload_unit_start_indicator : 1bit IGNORE
1717 * transport_priority : 1bit IGNORE
1718 * PID : 13bit => 0x9f 0xff
1719 * transport_scrambling_control : 2bit
1720 * adaptation_field_control : 2bit
1721 * continuity_counter : 4bit => 0x30
1723 pcrmask = 0xff9fff20;
1724 pcrpattern = 0x47000020 | ((program->pcr_pid & 0x1fff) << 8);
1726 for (i = 0; (i < 20) && (nbpcr < numpcr); i++) {
1730 gst_pad_pull_range (base->sinkpad,
1731 initoff + i * 500 * base->packetsize, 500 * base->packetsize, &buf);
1733 if (G_UNLIKELY (ret != GST_FLOW_OK))
1736 gst_byte_reader_init_from_buffer (&br, buf);
1739 size = GST_BUFFER_SIZE (buf);
1742 offset = gst_byte_reader_masked_scan_uint32 (&br, 0xff000000, 0x47000000,
1743 0, base->packetsize);
1748 while ((nbpcr < numpcr) && (size >= base->packetsize)) {
1750 guint32 header = GST_READ_UINT32_BE (br.data + offset);
1752 if ((header >> 24) != 0x47)
1755 if ((header & pcrmask) != pcrpattern) {
1756 /* Move offset forward by 1 packet */
1757 size -= base->packetsize;
1758 offset += base->packetsize;
1763 /* GST_DEBUG ("offset %" G_GUINT64_FORMAT, GST_BUFFER_OFFSET (buf) + offset);
1764 GST_MEMDUMP ("something", GST_BUFFER_DATA (buf) + offset, 16);*/
1765 if ((*(br.data + offset + 5)) & MPEGTS_AFC_PCR_FLAG) {
1766 guint64 lpcr = mpegts_packetizer_compute_pcr (br.data + offset + 6);
1768 GST_INFO ("Found PCR %" G_GUINT64_FORMAT " %" GST_TIME_FORMAT
1769 " at offset %" G_GUINT64_FORMAT, lpcr,
1770 GST_TIME_ARGS (PCRTIME_TO_GSTTIME (lpcr)),
1771 GST_BUFFER_OFFSET (buf) + offset);
1773 pcroffs[nbpcr] = GST_BUFFER_OFFSET (buf) + offset;
1774 /* Safeguard against bogus PCR (by detecting if it's the same as the
1775 * previous one or wheter the difference with the previous one is
1776 * greater than 10mins */
1778 if (pcrs[nbpcr] == pcrs[nbpcr - 1]) {
1779 GST_WARNING ("Found same PCR at different offset");
1780 } else if (pcrs[nbpcr] < pcrs[nbpcr - 1]) {
1781 GST_WARNING ("Found PCR wraparound");
1783 } else if ((pcrs[nbpcr] - pcrs[nbpcr - 1]) >
1784 (guint64) 10 * 60 * 27000000) {
1785 GST_WARNING ("PCR differs with previous PCR by more than 10 mins");
1791 /* Move offset forward by 1 packet */
1792 size -= base->packetsize;
1793 offset += base->packetsize;
1798 GST_DEBUG ("Found %d PCR", nbpcr);
1801 pcroffset->pcr = pcrs[0];
1802 pcroffset->offset = pcroffs[0];
1804 pcroffset->pcr = pcrs[nbpcr - 1];
1805 pcroffset->offset = pcroffs[nbpcr - 1];
1808 GST_DEBUG ("pcrdiff:%" GST_TIME_FORMAT " offsetdiff %" G_GUINT64_FORMAT,
1809 GST_TIME_ARGS (PCRTIME_TO_GSTTIME (pcrs[nbpcr - 1] - pcrs[0])),
1810 pcroffs[nbpcr - 1] - pcroffs[0]);
1811 GST_DEBUG ("Estimated bitrate %" G_GUINT64_FORMAT,
1812 gst_util_uint64_scale (GST_SECOND, pcroffs[nbpcr - 1] - pcroffs[0],
1813 PCRTIME_TO_GSTTIME (pcrs[nbpcr - 1] - pcrs[0])));
1814 GST_DEBUG ("Average PCR interval %" G_GUINT64_FORMAT,
1815 (pcroffs[nbpcr - 1] - pcroffs[0]) / nbpcr);
1818 /* Swallow any errors if it happened during the end scanning */
1828 gst_ts_demux_record_pcr (GstTSDemux * demux, TSDemuxStream * stream,
1829 guint64 pcr, guint64 offset)
1831 MpegTSBaseStream *bs = (MpegTSBaseStream *) stream;
1833 GST_LOG ("pid 0x%04x pcr:%" GST_TIME_FORMAT " at offset %"
1834 G_GUINT64_FORMAT, bs->pid,
1835 GST_TIME_ARGS (PCRTIME_TO_GSTTIME (pcr)), offset);
1837 if (G_LIKELY (bs->pid == demux->program->pcr_pid)) {
1838 demux->cur_pcr.gsttime = GST_CLOCK_TIME_NONE;
1839 demux->cur_pcr.offset = offset;
1840 demux->cur_pcr.pcr = pcr;
1841 /* set first_pcr in push mode */
1842 if (G_UNLIKELY (!demux->first_pcr.gsttime == GST_CLOCK_TIME_NONE)) {
1843 demux->first_pcr.gsttime = PCRTIME_TO_GSTTIME (pcr);
1844 demux->first_pcr.offset = offset;
1845 demux->first_pcr.pcr = pcr;
1849 if (G_UNLIKELY (demux->emit_statistics)) {
1851 st = gst_structure_id_empty_new (QUARK_TSDEMUX);
1852 gst_structure_id_set (st,
1853 QUARK_PID, G_TYPE_UINT, bs->pid,
1854 QUARK_OFFSET, G_TYPE_UINT64, offset, QUARK_PCR, G_TYPE_UINT64, pcr,
1856 gst_element_post_message (GST_ELEMENT_CAST (demux),
1857 gst_message_new_element (GST_OBJECT (demux), st));
1862 gst_ts_demux_record_opcr (GstTSDemux * demux, TSDemuxStream * stream,
1863 guint64 opcr, guint64 offset)
1865 MpegTSBaseStream *bs = (MpegTSBaseStream *) stream;
1867 GST_LOG ("pid 0x%04x opcr:%" GST_TIME_FORMAT " at offset %"
1868 G_GUINT64_FORMAT, bs->pid,
1869 GST_TIME_ARGS (PCRTIME_TO_GSTTIME (opcr)), offset);
1871 if (G_UNLIKELY (demux->emit_statistics)) {
1873 st = gst_structure_id_empty_new (QUARK_TSDEMUX);
1874 gst_structure_id_set (st,
1875 QUARK_PID, G_TYPE_UINT, bs->pid,
1876 QUARK_OFFSET, G_TYPE_UINT64, offset,
1877 QUARK_OPCR, G_TYPE_UINT64, opcr, NULL);
1878 gst_element_post_message (GST_ELEMENT_CAST (demux),
1879 gst_message_new_element (GST_OBJECT (demux), st));
1884 gst_ts_demux_record_pts (GstTSDemux * demux, TSDemuxStream * stream,
1885 guint64 pts, guint64 offset)
1887 MpegTSBaseStream *bs = (MpegTSBaseStream *) stream;
1889 GST_LOG ("pid 0x%04x pts:%" G_GUINT64_FORMAT " at offset %"
1890 G_GUINT64_FORMAT, bs->pid, pts, offset);
1892 if (G_UNLIKELY (GST_CLOCK_TIME_IS_VALID (stream->pts) &&
1893 ABSDIFF (stream->raw_pts, pts) > 900000)) {
1894 /* Detect rollover if diff > 10s */
1895 GST_LOG ("Detected rollover (previous:%" G_GUINT64_FORMAT " new:%"
1896 G_GUINT64_FORMAT ")", stream->raw_pts, pts);
1897 if (pts < stream->raw_pts) {
1898 /* Forward rollover */
1899 GST_LOG ("Forward rollover, incrementing nb_pts_rollover");
1900 stream->nb_pts_rollover++;
1902 /* Reverse rollover */
1903 GST_LOG ("Reverse rollover, decrementing nb_pts_rollover");
1904 stream->nb_pts_rollover--;
1908 /* Compute PTS in GstClockTime */
1909 stream->raw_pts = pts;
1911 MPEGTIME_TO_GSTTIME (pts + stream->nb_pts_rollover * PTS_DTS_MAX_VALUE);
1913 GST_LOG ("pid 0x%04x Stored PTS %" G_GUINT64_FORMAT " (%" GST_TIME_FORMAT ")",
1914 bs->pid, stream->raw_pts, GST_TIME_ARGS (stream->pts));
1917 if (G_UNLIKELY (demux->emit_statistics)) {
1919 st = gst_structure_id_empty_new (QUARK_TSDEMUX);
1920 gst_structure_id_set (st,
1921 QUARK_PID, G_TYPE_UINT, bs->pid,
1922 QUARK_OFFSET, G_TYPE_UINT64, offset, QUARK_PTS, G_TYPE_UINT64, pts,
1924 gst_element_post_message (GST_ELEMENT_CAST (demux),
1925 gst_message_new_element (GST_OBJECT (demux), st));
1930 gst_ts_demux_record_dts (GstTSDemux * demux, TSDemuxStream * stream,
1931 guint64 dts, guint64 offset)
1933 MpegTSBaseStream *bs = (MpegTSBaseStream *) stream;
1935 GST_LOG ("pid 0x%04x dts:%" G_GUINT64_FORMAT " at offset %"
1936 G_GUINT64_FORMAT, bs->pid, dts, offset);
1938 if (G_UNLIKELY (GST_CLOCK_TIME_IS_VALID (stream->dts) &&
1939 ABSDIFF (stream->raw_dts, dts) > 900000)) {
1940 /* Detect rollover if diff > 10s */
1941 GST_LOG ("Detected rollover (previous:%" G_GUINT64_FORMAT " new:%"
1942 G_GUINT64_FORMAT ")", stream->raw_dts, dts);
1943 if (dts < stream->raw_dts) {
1944 /* Forward rollover */
1945 GST_LOG ("Forward rollover, incrementing nb_dts_rollover");
1946 stream->nb_dts_rollover++;
1948 /* Reverse rollover */
1949 GST_LOG ("Reverse rollover, decrementing nb_dts_rollover");
1950 stream->nb_dts_rollover--;
1954 /* Compute DTS in GstClockTime */
1955 stream->raw_dts = dts;
1957 MPEGTIME_TO_GSTTIME (dts + stream->nb_dts_rollover * PTS_DTS_MAX_VALUE);
1959 GST_LOG ("pid 0x%04x Stored DTS %" G_GUINT64_FORMAT " (%" GST_TIME_FORMAT ")",
1960 bs->pid, stream->raw_dts, GST_TIME_ARGS (stream->dts));
1962 if (G_UNLIKELY (demux->emit_statistics)) {
1964 st = gst_structure_id_empty_new (QUARK_TSDEMUX);
1965 gst_structure_id_set (st,
1966 QUARK_PID, G_TYPE_UINT, bs->pid,
1967 QUARK_OFFSET, G_TYPE_UINT64, offset, QUARK_DTS, G_TYPE_UINT64, dts,
1969 gst_element_post_message (GST_ELEMENT_CAST (demux),
1970 gst_message_new_element (GST_OBJECT (demux), st));
1974 static inline GstClockTime
1975 calc_gsttime_from_pts (TSPcrOffset * start, guint64 pts)
1977 GstClockTime time = start->gsttime - PCRTIME_TO_GSTTIME (start->pcr);
1979 if (start->pcr > pts * 300)
1980 time += PCRTIME_TO_GSTTIME (PCR_MAX_VALUE) + MPEGTIME_TO_GSTTIME (pts);
1982 time += MPEGTIME_TO_GSTTIME (pts);
1989 TSPcrOffset_find_offset (gconstpointer a, gconstpointer b, gpointer user_data)
1991 if (((TSPcrOffset *) a)->offset < ((TSPcrOffset *) b)->offset)
1993 else if (((TSPcrOffset *) a)->offset > ((TSPcrOffset *) b)->offset)
2000 static GstFlowReturn
2001 gst_ts_demux_parse_pes_header (GstTSDemux * demux, TSDemuxStream * stream)
2003 MpegTSBase *base = (MpegTSBase *) demux;
2005 GstFlowReturn res = GST_FLOW_OK;
2009 guint64 bufferoffset;
2010 PESParsingResult parseres;
2012 data = GST_BUFFER_DATA (stream->pendingbuffers[0]);
2013 length = GST_BUFFER_SIZE (stream->pendingbuffers[0]);
2014 bufferoffset = GST_BUFFER_OFFSET (stream->pendingbuffers[0]);
2016 GST_MEMDUMP ("Header buffer", data, MIN (length, 32));
2018 parseres = mpegts_parse_pes_header (data, length, &header, &offset);
2019 if (G_UNLIKELY (parseres == PES_PARSING_NEED_MORE))
2021 if (G_UNLIKELY (parseres == PES_PARSING_BAD)) {
2022 GST_WARNING ("Error parsing PES header. pid: 0x%x stream_type: 0x%x",
2023 stream->stream.pid, stream->stream.stream_type);
2027 if (header.DTS != -1)
2028 gst_ts_demux_record_dts (demux, stream, header.DTS, bufferoffset);
2030 if (header.PTS != -1) {
2031 gst_ts_demux_record_pts (demux, stream, header.PTS, bufferoffset);
2034 /* WTH IS THIS ??? */
2035 if (demux->index_pcr.offset + PCR_WRAP_SIZE_128KBPS + 1000 * 128 < offset
2036 || (demux->index_pcr.offset > offset)) {
2037 /* find next entry */
2039 demux->index_pcr.offset = offset;
2040 next = gst_util_array_binary_search (demux->index->data,
2041 demux->index_size, sizeof (*next), TSPcrOffset_find_offset,
2042 GST_SEARCH_MODE_BEFORE, &demux->index_pcr, NULL);
2044 GST_INFO ("new index_pcr %" GST_TIME_FORMAT " offset: %"
2045 G_GINT64_FORMAT, GST_TIME_ARGS (next->gsttime), next->offset);
2047 demux->index_pcr = *next;
2050 time = calc_gsttime_from_pts (&demux->index_pcr, pts);
2053 GST_DEBUG_OBJECT (base, "stream PTS %" GST_TIME_FORMAT,
2054 GST_TIME_ARGS (stream->pts));
2056 /* safe default if insufficient upstream info */
2057 if (G_UNLIKELY (!GST_CLOCK_TIME_IS_VALID (base->in_gap) &&
2058 GST_CLOCK_TIME_IS_VALID (base->first_buf_ts) &&
2059 base->mode == BASE_MODE_PUSHING &&
2060 base->segment.format == GST_FORMAT_TIME)) {
2061 /* Find the earliest current PTS we're going to push */
2062 GstClockTime firstpts = GST_CLOCK_TIME_NONE;
2065 for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
2066 TSDemuxStream *pstream = (TSDemuxStream *) tmp->data;
2067 if (!GST_CLOCK_TIME_IS_VALID (firstpts) || pstream->pts < firstpts)
2068 firstpts = pstream->pts;
2071 base->in_gap = base->first_buf_ts - firstpts;
2072 GST_DEBUG_OBJECT (base, "upstream segment start %" GST_TIME_FORMAT
2073 ", first buffer timestamp: %" GST_TIME_FORMAT
2074 ", first PTS: %" GST_TIME_FORMAT
2075 ", interpolation gap: %" GST_TIME_FORMAT,
2076 GST_TIME_ARGS (base->segment.start),
2077 GST_TIME_ARGS (base->first_buf_ts), GST_TIME_ARGS (firstpts),
2078 GST_TIME_ARGS (base->in_gap));
2081 if (!GST_CLOCK_TIME_IS_VALID (base->in_gap))
2084 GST_BUFFER_TIMESTAMP (stream->pendingbuffers[0]) =
2085 stream->pts + base->in_gap;
2088 /* Remove PES headers */
2089 GST_DEBUG ("Moving data forward by %d bytes", header.header_size);
2090 GST_BUFFER_DATA (stream->pendingbuffers[0]) += header.header_size;
2091 GST_BUFFER_SIZE (stream->pendingbuffers[0]) -= header.header_size;
2093 /* FIXME : responsible for switching to PENDING_PACKET_BUFFER and
2094 * creating the bufferlist */
2096 /* Append to the buffer list */
2097 if (G_UNLIKELY (stream->current == NULL)) {
2100 /* Create a new bufferlist */
2101 stream->current = gst_buffer_list_new ();
2102 stream->currentit = gst_buffer_list_iterate (stream->current);
2103 stream->currentlist = NULL;
2104 gst_buffer_list_iterator_add_group (stream->currentit);
2106 /* Push pending buffers into the list */
2107 for (i = stream->nbpending; i; i--)
2108 stream->currentlist =
2109 g_list_prepend (stream->currentlist, stream->pendingbuffers[i - 1]);
2110 memset (stream->pendingbuffers, 0, TS_MAX_PENDING_BUFFERS);
2111 stream->nbpending = 0;
2113 stream->state = PENDING_PACKET_BUFFER;
2119 stream->state = PENDING_PACKET_DISCONT;
2124 * * WITH packet->payload != NULL
2125 * * WITH pending/current flushed out if beginning of new PES packet
2128 gst_ts_demux_queue_data (GstTSDemux * demux, TSDemuxStream * stream,
2129 MpegTSPacketizerPacket * packet)
2133 GST_DEBUG ("state:%d", stream->state);
2135 buf = packet->buffer;
2136 /* HACK : Instead of creating a new buffer, we just modify the data/size
2137 * of the buffer to point to the payload */
2138 GST_BUFFER_DATA (buf) = packet->payload;
2139 GST_BUFFER_SIZE (buf) = packet->data_end - packet->payload;
2141 if (stream->state == PENDING_PACKET_EMPTY) {
2142 if (G_UNLIKELY (!packet->payload_unit_start_indicator)) {
2143 stream->state = PENDING_PACKET_DISCONT;
2144 GST_WARNING ("Didn't get the first packet of this PES");
2146 GST_LOG ("EMPTY=>HEADER");
2147 stream->state = PENDING_PACKET_HEADER;
2149 GST_DEBUG ("Setting pad caps on buffer %p", buf);
2150 gst_buffer_set_caps (buf, GST_PAD_CAPS (stream->pad));
2155 if (stream->state == PENDING_PACKET_HEADER) {
2156 GST_LOG ("HEADER: appending data to array");
2157 /* Append to the array */
2158 stream->pendingbuffers[stream->nbpending++] = buf;
2160 /* parse the header */
2161 gst_ts_demux_parse_pes_header (demux, stream);
2162 } else if (stream->state == PENDING_PACKET_BUFFER) {
2163 GST_LOG ("BUFFER: appending data to bufferlist");
2164 stream->currentlist = g_list_prepend (stream->currentlist, buf);
2172 calculate_and_push_newsegment (GstTSDemux * demux, TSDemuxStream * stream)
2174 MpegTSBase *base = (MpegTSBase *) demux;
2175 GstEvent *newsegmentevent;
2176 gint64 start = 0, stop = GST_CLOCK_TIME_NONE, position = 0;
2177 GstClockTime firstpts = GST_CLOCK_TIME_NONE;
2180 GST_DEBUG ("Creating new newsegment for stream %p", stream);
2182 /* Outgoing newsegment values
2183 * start : The first/start PTS
2184 * stop : The last PTS (or -1)
2185 * position : The stream time corresponding to start
2187 * Except for live mode with incoming GST_TIME_FORMAT newsegment where
2188 * it is the same values as that incoming newsegment (and we convert the
2189 * PTS to that remote clock).
2192 for (tmp = demux->program->stream_list; tmp; tmp = tmp->next) {
2193 TSDemuxStream *pstream = (TSDemuxStream *) tmp->data;
2195 if (!GST_CLOCK_TIME_IS_VALID (firstpts) || pstream->pts < firstpts)
2196 firstpts = pstream->pts;
2199 if (base->mode == BASE_MODE_PUSHING) {
2200 /* FIXME : We're just ignore the upstream format for the time being */
2201 /* FIXME : We should use base->segment.format and a upstream latency query
2202 * to decide if we need to use live values or not */
2203 GST_DEBUG ("push-based. base Segment start:%" GST_TIME_FORMAT " duration:%"
2204 GST_TIME_FORMAT ", stop:%" GST_TIME_FORMAT ", time:%" GST_TIME_FORMAT,
2205 GST_TIME_ARGS (base->segment.start),
2206 GST_TIME_ARGS (base->segment.duration),
2207 GST_TIME_ARGS (base->segment.stop), GST_TIME_ARGS (base->segment.time));
2208 GST_DEBUG ("push-based. demux Segment start:%" GST_TIME_FORMAT " duration:%"
2209 GST_TIME_FORMAT ", stop:%" GST_TIME_FORMAT ", time:%" GST_TIME_FORMAT,
2210 GST_TIME_ARGS (demux->segment.start),
2211 GST_TIME_ARGS (demux->segment.duration),
2212 GST_TIME_ARGS (demux->segment.stop),
2213 GST_TIME_ARGS (demux->segment.time));
2215 GST_DEBUG ("stream pts: %" GST_TIME_FORMAT " first pts: %" GST_TIME_FORMAT,
2216 GST_TIME_ARGS (stream->pts), GST_TIME_ARGS (firstpts));
2218 if (base->segment.format == GST_FORMAT_TIME) {
2219 start = base->segment.start;
2220 stop = base->segment.stop;
2222 /* Shift the start depending on our position in the stream */
2223 start += firstpts + base->in_gap - base->first_buf_ts;
2227 GST_DEBUG ("pull-based. Segment start:%" GST_TIME_FORMAT " duration:%"
2228 GST_TIME_FORMAT ", time:%" GST_TIME_FORMAT,
2229 GST_TIME_ARGS (demux->segment.start),
2230 GST_TIME_ARGS (demux->segment.duration),
2231 GST_TIME_ARGS (demux->segment.time));
2233 GST_DEBUG ("firstpcr gsttime : %" GST_TIME_FORMAT,
2234 GST_TIME_ARGS (demux->first_pcr.gsttime));
2236 /* FIXME : This is not entirely correct. We should be using the PTS time
2237 * realm and not the PCR one. Doesn't matter *too* much if PTS/PCR values
2238 * aren't too far apart, but still. */
2239 start = demux->first_pcr.gsttime + demux->segment.start;
2240 stop = demux->first_pcr.gsttime + demux->segment.duration;
2241 position = demux->segment.time;
2244 GST_DEBUG ("new segment: start: %" GST_TIME_FORMAT " stop: %"
2245 GST_TIME_FORMAT " time: %" GST_TIME_FORMAT, GST_TIME_ARGS (start),
2246 GST_TIME_ARGS (stop), GST_TIME_ARGS (position));
2248 gst_event_new_new_segment (FALSE, 1.0, GST_FORMAT_TIME, start, stop,
2251 push_event ((MpegTSBase *) demux, newsegmentevent);
2253 demux->need_newsegment = FALSE;
2256 static GstFlowReturn
2257 gst_ts_demux_push_pending_data (GstTSDemux * demux, TSDemuxStream * stream)
2259 GstFlowReturn res = GST_FLOW_OK;
2260 MpegTSBaseStream *bs = (MpegTSBaseStream *) stream;
2262 GST_DEBUG ("stream:%p, pid:0x%04x stream_type:%d state:%d pad:%s:%s",
2263 stream, bs->pid, bs->stream_type, stream->state,
2264 GST_DEBUG_PAD_NAME (stream->pad));
2266 if (G_UNLIKELY (stream->current == NULL)) {
2267 GST_LOG ("stream->current == NULL");
2271 if (G_UNLIKELY (stream->state == PENDING_PACKET_EMPTY)) {
2272 GST_LOG ("EMPTY: returning");
2276 if (G_UNLIKELY (stream->state != PENDING_PACKET_BUFFER))
2279 if (G_UNLIKELY (stream->pad == NULL)) {
2280 g_list_foreach (stream->currentlist, (GFunc) gst_buffer_unref, NULL);
2281 g_list_free (stream->currentlist);
2282 gst_buffer_list_iterator_free (stream->currentit);
2283 gst_buffer_list_unref (stream->current);
2287 if (G_UNLIKELY (demux->need_newsegment))
2288 calculate_and_push_newsegment (demux, stream);
2290 /* We have a confirmed buffer, let's push it out */
2291 GST_LOG ("Putting pending data into GstBufferList");
2292 stream->currentlist = g_list_reverse (stream->currentlist);
2293 gst_buffer_list_iterator_add_list (stream->currentit, stream->currentlist);
2294 gst_buffer_list_iterator_free (stream->currentit);
2296 GST_DEBUG_OBJECT (stream->pad,
2297 "Pushing buffer list with timestamp: %" GST_TIME_FORMAT,
2298 GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (gst_buffer_list_get
2299 (stream->current, 0, 0))));
2301 res = gst_pad_push_list (stream->pad, stream->current);
2302 GST_DEBUG_OBJECT (stream->pad, "Returned %s", gst_flow_get_name (res));
2303 res = tsdemux_combine_flows (demux, stream, res);
2304 GST_DEBUG_OBJECT (stream->pad, "combined %s", gst_flow_get_name (res));
2307 /* Reset everything */
2308 GST_LOG ("Resetting to EMPTY");
2309 stream->state = PENDING_PACKET_EMPTY;
2310 memset (stream->pendingbuffers, 0, TS_MAX_PENDING_BUFFERS);
2311 stream->nbpending = 0;
2312 stream->current = NULL;
2317 static GstFlowReturn
2318 gst_ts_demux_handle_packet (GstTSDemux * demux, TSDemuxStream * stream,
2319 MpegTSPacketizerPacket * packet, MpegTSPacketizerSection * section)
2321 GstFlowReturn res = GST_FLOW_OK;
2323 GST_DEBUG ("buffer:%p, data:%p", GST_BUFFER_DATA (packet->buffer),
2325 GST_LOG ("pid 0x%04x pusi:%d, afc:%d, cont:%d, payload:%p",
2327 packet->payload_unit_start_indicator,
2328 packet->adaptation_field_control,
2329 packet->continuity_counter, packet->payload);
2332 GST_DEBUG ("section complete:%d, buffer size %d",
2333 section->complete, GST_BUFFER_SIZE (section->buffer));
2334 gst_buffer_unref (packet->buffer);
2338 if (G_UNLIKELY (packet->payload_unit_start_indicator))
2339 /* Flush previous data */
2340 res = gst_ts_demux_push_pending_data (demux, stream);
2342 if (packet->adaptation_field_control & 0x2) {
2343 if (packet->afc_flags & MPEGTS_AFC_PCR_FLAG)
2344 gst_ts_demux_record_pcr (demux, stream, packet->pcr,
2345 GST_BUFFER_OFFSET (packet->buffer));
2346 if (packet->afc_flags & MPEGTS_AFC_OPCR_FLAG)
2347 gst_ts_demux_record_opcr (demux, stream, packet->opcr,
2348 GST_BUFFER_OFFSET (packet->buffer));
2351 if (packet->payload)
2352 gst_ts_demux_queue_data (demux, stream, packet);
2354 gst_buffer_unref (packet->buffer);
2360 gst_ts_demux_flush (MpegTSBase * base)
2362 GstTSDemux *demux = GST_TS_DEMUX_CAST (base);
2364 demux->need_newsegment = TRUE;
2365 gst_ts_demux_flush_streams (demux);
2368 static GstFlowReturn
2369 gst_ts_demux_push (MpegTSBase * base, MpegTSPacketizerPacket * packet,
2370 MpegTSPacketizerSection * section)
2372 GstTSDemux *demux = GST_TS_DEMUX_CAST (base);
2373 TSDemuxStream *stream = NULL;
2374 GstFlowReturn res = GST_FLOW_OK;
2376 if (G_LIKELY (demux->program)) {
2377 stream = (TSDemuxStream *) demux->program->streams[packet->pid];
2380 res = gst_ts_demux_handle_packet (demux, stream, packet, section);
2381 } else if (packet->buffer)
2382 gst_buffer_unref (packet->buffer);
2385 gst_buffer_unref (packet->buffer);
2391 gst_ts_demux_plugin_init (GstPlugin * plugin)
2393 GST_DEBUG_CATEGORY_INIT (ts_demux_debug, "tsdemux", 0,
2394 "MPEG transport stream demuxer");
2397 return gst_element_register (plugin, "tsdemux",
2398 GST_RANK_SECONDARY, GST_TYPE_TS_DEMUX);