aml-tsdemux/amltsparse.c

/*
 * mpegtsparse.c -
 * Copyright (C) 2007 Alessandro Decina
 *
 * Authors:
 *   Alessandro Decina <alessandro@nnva.org>
 *   Zaheer Abbas Merali <zaheerabbas at merali dot org>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
 * Boston, MA 02110-1301, USA.
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "amltsbase.h"
#include "amltsparse.h"
#include "gstamldesc.h"
/* latency in mseconds is maximum 100 ms between PCR */
#define TS_LATENCY 100

#define TABLE_ID_UNSET 0xFF
#define RUNNING_STATUS_RUNNING 4
#define SYNC_BYTE 0x47

GST_DEBUG_CATEGORY_STATIC (mpegts_parse_debug);
#define GST_CAT_DEFAULT mpegts_parse_debug

typedef struct _AmlTSParsePad AmlTSParsePad;

typedef struct
{
  AmlTSBaseProgram program;
  AmlTSParsePad *tspad;
} AmlTSParseProgram;

struct _AmlTSParsePad
{
  GstPad *pad;

  /* the program number that the peer wants on this pad */
  gint program_number;
  AmlTSParseProgram *program;

  /* set to FALSE before a push and TRUE after */
  gboolean pushed;

  /* the return of the latest push */
  GstFlowReturn flow_return;

  AmlTSParse2Adapter ts_adapter;
};

static GstStaticPadTemplate src_template =
GST_STATIC_PAD_TEMPLATE ("src", GST_PAD_SRC,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS ("video/mpegts, " "systemstream = (boolean) true ")
    );

static GstStaticPadTemplate program_template =
GST_STATIC_PAD_TEMPLATE ("program_%u", GST_PAD_SRC,
    GST_PAD_REQUEST,
    GST_STATIC_CAPS ("video/mpegts, " "systemstream = (boolean) true ")
    );

enum
{
  PROP_0,
  PROP_SET_TIMESTAMPS,
  PROP_SMOOTHING_LATENCY,
  PROP_PCR_PID,
  PROP_ALIGNMENT,
  PROP_SPLIT_ON_RAI,
  /* FILL ME */
};

static void amlts_parse_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec);
static void amlts_parse_get_property (GObject * object, guint prop_id,
    GValue * value, GParamSpec * pspec);

static void
amlts_parse_program_started (AmlTSBase * base, AmlTSBaseProgram * program);
static void
amlts_parse_program_stopped (AmlTSBase * base, AmlTSBaseProgram * program);

static GstFlowReturn
amlts_parse_push (AmlTSBase * base, AmlTSPacketizerPacket * packet,
    GstMpegtsSection * section);
static void amlts_parse_inspect_packet (AmlTSBase * base,
    AmlTSPacketizerPacket * packet);
static GstFlowReturn amlts_parse_have_buffer (AmlTSBase * base,
    GstBuffer * buffer);

static AmlTSParsePad *amlts_parse_create_tspad (AmlTSParse2 * parse,
    const gchar * name);
static void amlts_parse_destroy_tspad (AmlTSParse2 * parse,
    AmlTSParsePad * tspad);

static void amlts_parse_pad_removed (GstElement * element, GstPad * pad);
static GstPad *amlts_parse_request_new_pad (GstElement * element,
    GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
static void amlts_parse_release_pad (GstElement * element, GstPad * pad);
static gboolean amlts_parse_src_pad_query (GstPad * pad, GstObject * parent,
    GstQuery * query);
static gboolean push_event (AmlTSBase * base, GstEvent * event);

#define amlts_parse_parent_class parent_class
G_DEFINE_TYPE (AmlTSParse2, amlts_parse, GST_TYPE_AMLTS_BASE);
/* aml mark:new method at new version
GST_ELEMENT_REGISTER_DEFINE_WITH_CODE (tsparse, "tsparse",
    GST_RANK_NONE, GST_TYPE_MPEGTS_PARSE,
    GST_DEBUG_CATEGORY_INIT (mpegts_parse_debug, "tsparse", 0,
        "MPEG transport stream parser"));
*/
static void amlts_parse_reset (AmlTSBase * base);
static GstFlowReturn amlts_parse_input_done (AmlTSBase * base);
static GstFlowReturn
aml_drain_pending_buffers (AmlTSParse2 * parse, gboolean drain_all);

static void
amlts_parse_finalize (GObject * object)
{
  AmlTSParse2 *parse = (AmlTSParse2 *) object;

  gst_flow_combiner_free (parse->flowcombiner);

  gst_adapter_clear (parse->ts_adapter.adapter);
  g_object_unref (parse->ts_adapter.adapter);

  GST_CALL_PARENT (G_OBJECT_CLASS, finalize, (object));
}

static void
amlts_parse_class_init (AmlTSParse2Class * klass)
{
  GObjectClass *gobject_class = (GObjectClass *) (klass);
  GstElementClass *element_class;
  AmlTSBaseClass *ts_class;

  gobject_class->set_property = amlts_parse_set_property;
  gobject_class->get_property = amlts_parse_get_property;
  gobject_class->finalize = amlts_parse_finalize;

  g_object_class_install_property (gobject_class, PROP_SET_TIMESTAMPS,
      g_param_spec_boolean ("set-timestamps",
          "Timestamp (or re-timestamp) the output stream",
          "If set, timestamps will be set on the output buffers using "
          "PCRs and smoothed over the smoothing-latency period", FALSE,
          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  g_object_class_install_property (gobject_class, PROP_SMOOTHING_LATENCY,
      g_param_spec_uint ("smoothing-latency", "Smoothing Latency",
          "Additional latency in microseconds for smoothing jitter in input timestamps on live capture",
          0, G_MAXUINT, 0, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  g_object_class_install_property (gobject_class, PROP_PCR_PID,
      g_param_spec_int ("pcr-pid", "PID containing PCR",
          "Set the PID to use for PCR values (-1 for auto)",
          -1, G_MAXINT, -1, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  g_object_class_install_property (gobject_class, PROP_ALIGNMENT,
      g_param_spec_uint ("alignment", "Alignment",
          "Number of packets per buffer (padded with dummy packets on EOS) (0 = auto)",
          0, G_MAXUINT, 0, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  g_object_class_install_property (gobject_class, PROP_SPLIT_ON_RAI,
      g_param_spec_boolean ("split-on-rai", "Split on RAI",
          "If set, buffers sized smaller than the alignment will be sent "
          "so that RAI packets are at the start of a new buffer", FALSE,
          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

  element_class = GST_ELEMENT_CLASS (klass);
  element_class->pad_removed = amlts_parse_pad_removed;
  element_class->request_new_pad = amlts_parse_request_new_pad;
  element_class->release_pad = amlts_parse_release_pad;

  gst_element_class_add_static_pad_template (element_class, &src_template);
  gst_element_class_add_static_pad_template (element_class, &program_template);

  gst_element_class_set_static_metadata (element_class,
      "MPEG transport stream parser", "Codec/Parser",
      "Parses MPEG2 transport streams",
      "Alessandro Decina <alessandro@nnva.org>, "
      "Zaheer Abbas Merali <zaheerabbas at merali dot org>");

  ts_class = GST_AMLTS_BASE_CLASS (klass);
  ts_class->push = GST_DEBUG_FUNCPTR (amlts_parse_push);
  ts_class->push_event = GST_DEBUG_FUNCPTR (push_event);
  ts_class->program_started = GST_DEBUG_FUNCPTR (amlts_parse_program_started);
  ts_class->program_stopped = GST_DEBUG_FUNCPTR (amlts_parse_program_stopped);
  ts_class->reset = GST_DEBUG_FUNCPTR (amlts_parse_reset);
  ts_class->input_done = GST_DEBUG_FUNCPTR (amlts_parse_input_done);
  ts_class->inspect_packet = GST_DEBUG_FUNCPTR (amlts_parse_inspect_packet);
}

static void
amlts_parse_init (AmlTSParse2 * parse)
{
  AmlTSBase *base = (AmlTSBase *) parse;

  base->program_size = sizeof (AmlTSParseProgram);
  base->push_data = TRUE;
  base->push_section = TRUE;
  base->push_unknown = TRUE;

  parse->user_pcr_pid = parse->pcr_pid = -1;

  parse->flowcombiner = gst_flow_combiner_new ();

  parse->srcpad = gst_pad_new_from_static_template (&src_template, "src");
  gst_flow_combiner_add_pad (parse->flowcombiner, parse->srcpad);
  parse->first = TRUE;
  gst_pad_set_query_function (parse->srcpad,
      GST_DEBUG_FUNCPTR (amlts_parse_src_pad_query));
  gst_element_add_pad (GST_ELEMENT (parse), parse->srcpad);

  parse->have_group_id = FALSE;
  parse->group_id = G_MAXUINT;

  parse->ts_adapter.adapter = gst_adapter_new ();
  parse->ts_adapter.packets_in_adapter = 0;
  parse->ts_adapter.first_is_keyframe = TRUE;
  parse->alignment = 0;
  parse->is_eos = FALSE;
  parse->header = 0;
  parse->split_on_rai = FALSE;
}

static void
amlts_parse_reset (AmlTSBase * base)
{
  AmlTSParse2 *parse = (AmlTSParse2 *) base;

  /* Set the various know PIDs we are interested in */

  /* CAT */
  MPEGTS_BIT_SET (base->known_psi, 1);
  /* NIT, ST */
  MPEGTS_BIT_SET (base->known_psi, 0x10);
  /* SDT, BAT, ST */
  MPEGTS_BIT_SET (base->known_psi, 0x11);
  /* EIT, ST, CIT (TS 102 323) */
  MPEGTS_BIT_SET (base->known_psi, 0x12);
  /* RST, ST */
  MPEGTS_BIT_SET (base->known_psi, 0x13);
  /* RNT (TS 102 323) */
  MPEGTS_BIT_SET (base->known_psi, 0x16);
  /* inband signalling */
  MPEGTS_BIT_SET (base->known_psi, 0x1c);
  /* measurement */
  MPEGTS_BIT_SET (base->known_psi, 0x1d);
  /* DIT */
  MPEGTS_BIT_SET (base->known_psi, 0x1e);
  /* SIT */
  MPEGTS_BIT_SET (base->known_psi, 0x1f);

  parse->first = TRUE;
  parse->have_group_id = FALSE;
  parse->group_id = G_MAXUINT;

  g_list_free_full (parse->pending_buffers, (GDestroyNotify) gst_buffer_unref);
  parse->pending_buffers = NULL;

  parse->current_pcr = GST_CLOCK_TIME_NONE;
  parse->previous_pcr = GST_CLOCK_TIME_NONE;
  parse->base_pcr = GST_CLOCK_TIME_NONE;
  parse->bytes_since_pcr = 0;
  parse->pcr_pid = parse->user_pcr_pid;
  parse->ts_offset = 0;

  gst_adapter_clear (parse->ts_adapter.adapter);
  parse->ts_adapter.packets_in_adapter = 0;
  parse->ts_adapter.first_is_keyframe = TRUE;
  parse->is_eos = FALSE;
  parse->header = 0;
}

static void
amlts_parse_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec)
{
  AmlTSParse2 *parse = (AmlTSParse2 *) object;

  switch (prop_id) {
    case PROP_SET_TIMESTAMPS:
      parse->set_timestamps = g_value_get_boolean (value);
      break;
    case PROP_SMOOTHING_LATENCY:
      parse->smoothing_latency = GST_USECOND * g_value_get_uint (value);
      amlts_packetizer_set_pcr_discont_threshold (GST_AMLTS_BASE
          (parse)->packetizer, parse->smoothing_latency);
      break;
    case PROP_PCR_PID:
      parse->pcr_pid = parse->user_pcr_pid = g_value_get_int (value);
      break;
    case PROP_ALIGNMENT:
      parse->alignment = g_value_get_uint (value);
      break;
    case PROP_SPLIT_ON_RAI:
      parse->split_on_rai = g_value_get_boolean (value);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
  }
}

static void
amlts_parse_get_property (GObject * object, guint prop_id,
    GValue * value, GParamSpec * pspec)
{
  AmlTSParse2 *parse = (AmlTSParse2 *) object;

  switch (prop_id) {
    case PROP_SET_TIMESTAMPS:
      g_value_set_boolean (value, parse->set_timestamps);
      break;
    case PROP_SMOOTHING_LATENCY:
      g_value_set_uint (value, parse->smoothing_latency / GST_USECOND);
      break;
    case PROP_PCR_PID:
      g_value_set_int (value, parse->pcr_pid);
      break;
    case PROP_ALIGNMENT:
      g_value_set_uint (value, parse->alignment);
      break;
    case PROP_SPLIT_ON_RAI:
      g_value_set_boolean (value, parse->split_on_rai);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
  }
}

static gboolean
aml_prepare_src_pad (AmlTSBase * base, AmlTSParse2 * parse)
{
  GstEvent *event;
  gchar *stream_id;
  GstCaps *caps;

  if (!parse->first)
    return TRUE;

  /* If there's no packet_size yet, we can't set caps yet */
  if (G_UNLIKELY (base->packetizer->packet_size == 0))
    return FALSE;

  stream_id =
      gst_pad_create_stream_id (parse->srcpad, GST_ELEMENT_CAST (base),
      "multi-program");

  event =
      gst_pad_get_sticky_event (parse->parent.sinkpad, GST_EVENT_STREAM_START,
      0);
  if (event) {
    if (gst_event_parse_group_id (event, &parse->group_id))
      parse->have_group_id = TRUE;
    else
      parse->have_group_id = FALSE;
    gst_event_unref (event);
  } else if (!parse->have_group_id) {
    parse->have_group_id = TRUE;
    parse->group_id = gst_util_group_id_next ();
  }
  event = gst_event_new_stream_start (stream_id);
  if (parse->have_group_id)
    gst_event_set_group_id (event, parse->group_id);

  gst_pad_push_event (parse->srcpad, event);
  g_free (stream_id);

  caps = gst_caps_new_simple ("video/mpegts",
      "systemstream", G_TYPE_BOOLEAN, TRUE,
      "packetsize", G_TYPE_INT, base->packetizer->packet_size, NULL);

  gst_pad_set_caps (parse->srcpad, caps);
  gst_caps_unref (caps);

  /* If setting output timestamps, ensure that the output segment is TIME */
  if (parse->set_timestamps == FALSE || base->segment.format == GST_FORMAT_TIME)
    /* Just use the upstream segment */
    base->out_segment = base->segment;
  else {
    GstSegment *seg = &base->out_segment;
    gst_segment_init (seg, GST_FORMAT_TIME);
    GST_DEBUG_OBJECT (parse,
        "Generating time output segment %" GST_SEGMENT_FORMAT, seg);
  }
  gst_pad_push_event (parse->srcpad,
      gst_event_new_segment (&base->out_segment));

  parse->first = FALSE;

  return TRUE;
}

static gboolean
push_event (AmlTSBase * base, GstEvent * event)
{
  AmlTSParse2 *parse = (AmlTSParse2 *) base;
  GList *tmp;

  if (G_UNLIKELY (parse->first)) {
    /* We will send the segment when really starting  */
    if (G_UNLIKELY (GST_EVENT_TYPE (event) == GST_EVENT_SEGMENT)) {
      gst_event_unref (event);
      return TRUE;
    }
    aml_prepare_src_pad (base, parse);
  }
  if (G_UNLIKELY (GST_EVENT_TYPE (event) == GST_EVENT_EOS)) {
    gsize packet_size = base->packetizer->packet_size;

    parse->is_eos = TRUE;

    if (packet_size > 0 && parse->alignment > 0 &&
        parse->ts_adapter.packets_in_adapter > 0
        && parse->ts_adapter.packets_in_adapter < parse->alignment) {
      GstBuffer *buf;
      GstMapInfo map;
      guint8 *data;
      gint missing_packets =
          parse->alignment - parse->ts_adapter.packets_in_adapter;
      gint i = missing_packets;

      GST_DEBUG_OBJECT (parse, "Adding %d dummy packets", missing_packets);

      buf = gst_buffer_new_and_alloc (missing_packets * packet_size);
      gst_buffer_map (buf, &map, GST_MAP_READWRITE);
      data = map.data;

      for (; i > 0; i--) {
        gint offset;

        if (packet_size > MPEGTS_NORMAL_PACKETSIZE) {
          parse->header++;
          GST_WRITE_UINT32_BE (data, parse->header);
          offset = 4;
        } else {
          offset = 0;
        }
        GST_WRITE_UINT8 (data + offset, SYNC_BYTE);
        /* null packet PID */
        GST_WRITE_UINT16_BE (data + offset + 1, 0x1FFF);
        /* no adaptation field exists | continuity counter undefined */
        GST_WRITE_UINT8 (data + offset + 3, 0x10);
        /* payload */
        memset (data + offset + 4, 0, MPEGTS_NORMAL_PACKETSIZE - 4);
        data += packet_size;
      }
      gst_buffer_unmap (buf, &map);
      gst_adapter_push (parse->ts_adapter.adapter, buf);
      parse->ts_adapter.packets_in_adapter += missing_packets;
    }
    aml_drain_pending_buffers (parse, TRUE);
  }

  if (G_UNLIKELY (GST_EVENT_TYPE (event) == GST_EVENT_SEGMENT))
    parse->ts_offset = 0;

  for (tmp = parse->srcpads; tmp; tmp = tmp->next) {
    GstPad *pad = (GstPad *) tmp->data;
    if (pad) {
      gst_event_ref (event);
      gst_pad_push_event (pad, event);
    }
  }

  gst_pad_push_event (parse->srcpad, event);

  return TRUE;
}

static AmlTSParsePad *
amlts_parse_create_tspad (AmlTSParse2 * parse, const gchar * pad_name)
{
  GstPad *pad;
  AmlTSParsePad *tspad;

  pad = gst_pad_new_from_static_template (&program_template, pad_name);
  gst_pad_set_query_function (pad,
      GST_DEBUG_FUNCPTR (amlts_parse_src_pad_query));

  /* create our wrapper */
  tspad = g_new0 (AmlTSParsePad, 1);
  tspad->pad = pad;
  tspad->program_number = -1;
  tspad->program = NULL;
  tspad->pushed = FALSE;
  tspad->flow_return = GST_FLOW_NOT_LINKED;
  tspad->ts_adapter.adapter = gst_adapter_new ();
  tspad->ts_adapter.packets_in_adapter = 0;
  tspad->ts_adapter.first_is_keyframe = TRUE;
  gst_pad_set_element_private (pad, tspad);
  gst_flow_combiner_add_pad (parse->flowcombiner, pad);

  return tspad;
}

static void
amlts_parse_destroy_tspad (AmlTSParse2 * parse, AmlTSParsePad * tspad)
{
  gst_adapter_clear (tspad->ts_adapter.adapter);
  g_object_unref (tspad->ts_adapter.adapter);

  /* free the wrapper */
  g_free (tspad);
}

static void
amlts_parse_pad_removed (GstElement * element, GstPad * pad)
{
  AmlTSParsePad *tspad;
  AmlTSParse2 *parse = GST_AMLTS_PARSE (element);

  if (gst_pad_get_direction (pad) == GST_PAD_SINK)
    return;

  tspad = (AmlTSParsePad *) gst_pad_get_element_private (pad);
  if (tspad) {
    amlts_parse_destroy_tspad (parse, tspad);

    parse->srcpads = g_list_remove_all (parse->srcpads, pad);
  }

  if (GST_ELEMENT_CLASS (parent_class)->pad_removed)
    GST_ELEMENT_CLASS (parent_class)->pad_removed (element, pad);
}

static GstPad *
amlts_parse_request_new_pad (GstElement * element, GstPadTemplate * template,
    const gchar * padname, const GstCaps * caps)
{
  AmlTSParse2 *parse;
  AmlTSParsePad *tspad;
  AmlTSParseProgram *parseprogram;
  GstPad *pad;
  gint program_num = -1;
  GstEvent *event;
  gchar *stream_id;

  g_return_val_if_fail (template != NULL, NULL);
  g_return_val_if_fail (GST_IS_AMLTS_PARSE (element), NULL);
  g_return_val_if_fail (padname != NULL, NULL);

  sscanf (padname + 8, "%d", &program_num);

  GST_DEBUG_OBJECT (element, "padname:%s, program:%d", padname, program_num);

  parse = GST_AMLTS_PARSE (element);

  tspad = amlts_parse_create_tspad (parse, padname);
  tspad->program_number = program_num;

  /* Find if the program is already active */
  parseprogram =
      (AmlTSParseProgram *) amlts_base_get_program (GST_AMLTS_BASE (parse),
      program_num);
  if (parseprogram) {
    tspad->program = parseprogram;
    parseprogram->tspad = tspad;
  }

  pad = tspad->pad;
  parse->srcpads = g_list_append (parse->srcpads, pad);

  gst_pad_set_active (pad, TRUE);

  stream_id = gst_pad_create_stream_id (pad, element, padname + 8);

  event =
      gst_pad_get_sticky_event (parse->parent.sinkpad, GST_EVENT_STREAM_START,
      0);
  if (event) {
    if (gst_event_parse_group_id (event, &parse->group_id))
      parse->have_group_id = TRUE;
    else
      parse->have_group_id = FALSE;
    gst_event_unref (event);
  } else if (!parse->have_group_id) {
    parse->have_group_id = TRUE;
    parse->group_id = gst_util_group_id_next ();
  }
  event = gst_event_new_stream_start (stream_id);
  if (parse->have_group_id)
    gst_event_set_group_id (event, parse->group_id);

  gst_pad_push_event (pad, event);
  g_free (stream_id);

  gst_element_add_pad (element, pad);

  return pad;
}

static GstBuffer *
amlts_packet_to_buffer (AmlTSPacketizerPacket * packet)
{
  GstBuffer *buf =
      gst_buffer_new_and_alloc (packet->data_end - packet->data_start);
  gst_buffer_fill (buf, 0, packet->data_start,
      packet->data_end - packet->data_start);
  return buf;
}

static void
amlts_parse_release_pad (GstElement * element, GstPad * pad)
{
  AmlTSParse2 *parse = (AmlTSParse2 *) element;

  gst_pad_set_active (pad, FALSE);
  /* we do the cleanup in GstElement::pad-removed */
  gst_flow_combiner_remove_pad (parse->flowcombiner, pad);
  gst_element_remove_pad (element, pad);
}

static GstFlowReturn
aml_empty_adapter_into_pad (AmlTSParse2 * parse, AmlTSParse2Adapter * ts_adapter,
    GstPad * pad)
{
  GstAdapter *adapter = ts_adapter->adapter;
  GstBuffer *buf = NULL;
  guint64 pts_dist, dts_dist;
  GstClockTime pts, dts;
  gsize avail = gst_adapter_available (adapter);
  GstFlowReturn ret = GST_FLOW_OK;
  gsize offset;

  if (avail > 0)
    buf = gst_adapter_take_buffer (adapter, avail);
  /* Find the previous PTS/DTS. We also handle un-aligned input since want to
   * use the most recent PTS/DTS if present */
  offset = MIN (GST_AMLTS_BASE (parse)->packetizer->packet_size, 188);
  pts = gst_adapter_prev_pts_at_offset (adapter, offset, &pts_dist);
  dts = gst_adapter_prev_dts_at_offset (adapter, offset, &dts_dist);

  GST_LOG_OBJECT (pad,
      "prev pts:%" GST_TIME_FORMAT " (dist:%" G_GUINT64_FORMAT ") dts:%"
      GST_TIME_FORMAT " (dist:%" G_GUINT64_FORMAT ")",
      GST_TIME_ARGS (pts), pts_dist, GST_TIME_ARGS (dts), dts_dist);

  ts_adapter->packets_in_adapter = 0;

  if (buf) {
    GST_BUFFER_PTS (buf) = pts;
    GST_BUFFER_DTS (buf) = dts;
    if (!ts_adapter->first_is_keyframe)
      gst_buffer_set_flags (buf, GST_BUFFER_FLAG_DELTA_UNIT);
    ret = gst_pad_push (pad, buf);
  }

  return ret;
}

static GstFlowReturn
enqueue_and_maybe_push_buffer (AmlTSParse2 * parse, GstPad * pad,
    AmlTSParse2Adapter * ts_adapter, GstBuffer * buffer)
{
  GstFlowReturn ret = GST_FLOW_OK;

  if (buffer != NULL) {
    if (parse->alignment == 1) {
      ret = gst_pad_push (pad, buffer);
      ret = gst_flow_combiner_update_flow (parse->flowcombiner, ret);
    } else {
      if (!GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT)
          && parse->split_on_rai) {
        ret = aml_empty_adapter_into_pad (parse, ts_adapter, pad);
        ret = gst_flow_combiner_update_flow (parse->flowcombiner, ret);
      }
      gst_adapter_push (ts_adapter->adapter, buffer);
      ts_adapter->packets_in_adapter++;
      if (ts_adapter->packets_in_adapter == 1 && parse->split_on_rai) {
        ts_adapter->first_is_keyframe =
            !GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT);
      }

      if (ts_adapter->packets_in_adapter == parse->alignment
          && ts_adapter->packets_in_adapter > 0) {
        ret = aml_empty_adapter_into_pad (parse, ts_adapter, pad);
        ret = gst_flow_combiner_update_flow (parse->flowcombiner, ret);
      }
    }
  }

  return ret;
}

static GstFlowReturn
amlts_parse_tspad_push_section (AmlTSParse2 * parse, AmlTSParsePad * tspad,
    GstMpegtsSection * section, AmlTSPacketizerPacket * packet,
    GstBuffer * buf)
{
  GstFlowReturn ret = GST_FLOW_OK;
  gboolean to_push = TRUE;

  if (tspad->program_number != -1) {
    if (tspad->program) {
      /* we push all sections to all pads except PMTs which we
       * only push to pads meant to receive that program number */
      if (section->table_id == 0x02) {
        /* PMT */
        if (section->subtable_extension != tspad->program_number)
          to_push = FALSE;
      }
    } else if (section->table_id != 0x00) {
      /* there's a program filter on the pad but the PMT for the program has not
       * been parsed yet, ignore the pad until we get a PMT.
       * But we always allow PAT to go through */
      to_push = FALSE;
    }
  }

  GST_DEBUG_OBJECT (parse,
      "pushing section: %d program number: %d table_id: %d", to_push,
      tspad->program_number, section->table_id);

  if (to_push) {
    ret =
        enqueue_and_maybe_push_buffer (parse, tspad->pad,
        &tspad->ts_adapter, gst_buffer_ref (buf));
  }

  GST_LOG_OBJECT (parse, "Returning %s", gst_flow_get_name (ret));
  return ret;
}

static GstFlowReturn
amlts_parse_tspad_push (AmlTSParse2 * parse, AmlTSParsePad * tspad,
    AmlTSPacketizerPacket * packet, GstBuffer * buf)
{
  GstFlowReturn ret = GST_FLOW_OK;
  AmlTSBaseProgram *bp = NULL;

  if (tspad->program_number != -1) {
    if (tspad->program)
      bp = (AmlTSBaseProgram *) tspad->program;
    else
      bp = amlts_base_get_program ((AmlTSBase *) parse,
          tspad->program_number);
  }

  if (bp) {
    if (packet->pid == bp->pmt_pid || bp->streams == NULL
        || bp->streams[packet->pid]) {
      /* push if there's no filter or if the pid is in the filter */
      ret = gst_pad_push (tspad->pad, gst_buffer_ref (buf));
      ret = gst_flow_combiner_update_flow (parse->flowcombiner, ret);
    }
  }
  GST_DEBUG_OBJECT (parse, "Returning %s", gst_flow_get_name (ret));

  return ret;
}

static void
aml_pad_clear_for_push (GstPad * pad, AmlTSParse2 * parse)
{
  AmlTSParsePad *tspad = (AmlTSParsePad *) gst_pad_get_element_private (pad);

  tspad->flow_return = GST_FLOW_NOT_LINKED;
  tspad->pushed = FALSE;
}

static GstFlowReturn
amlts_parse_push (AmlTSBase * base, AmlTSPacketizerPacket * packet,
    GstMpegtsSection * section)
{
  AmlTSParse2 *parse = (AmlTSParse2 *) base;
  guint32 pads_cookie;
  gboolean done = FALSE;
  GstPad *pad = NULL;
  AmlTSParsePad *tspad;
  GstFlowReturn ret;
  GList *srcpads;
  GstBuffer *buf;

  GST_OBJECT_LOCK (parse);
  srcpads = parse->srcpads;

  /* clear tspad->pushed on pads */
  g_list_foreach (srcpads, (GFunc) aml_pad_clear_for_push, parse);
  if (srcpads)
    ret = GST_FLOW_NOT_LINKED;
  else
    ret = GST_FLOW_OK;

  /* Get cookie and source pads list */
  pads_cookie = GST_ELEMENT_CAST (parse)->pads_cookie;
  if (G_LIKELY (srcpads)) {
    pad = GST_PAD_CAST (srcpads->data);
    g_object_ref (pad);
  }
  GST_OBJECT_UNLOCK (parse);

  buf = amlts_packet_to_buffer (packet);
  if (parse->split_on_rai
      && !(packet->afc_flags & MPEGTS_AFC_RANDOM_ACCESS_FLAG)) {
    gst_buffer_set_flags (buf, GST_BUFFER_FLAG_DELTA_UNIT);
  }

  /* Copy over input PTS/DTS (if present) */
  GST_BUFFER_DTS (buf) = base->packetizer->last_dts;
  GST_BUFFER_PTS (buf) = base->packetizer->last_pts;
  ret = amlts_parse_have_buffer (base, gst_buffer_ref (buf));

  while (pad && !done) {
    tspad = gst_pad_get_element_private (pad);

    if (G_LIKELY (!tspad->pushed)) {
      if (section) {
        tspad->flow_return =
            amlts_parse_tspad_push_section (parse, tspad, section, packet,
            buf);
      } else {
        tspad->flow_return =
            amlts_parse_tspad_push (parse, tspad, packet, buf);
      }
      tspad->pushed = TRUE;

      if (G_UNLIKELY (tspad->flow_return != GST_FLOW_OK
              && tspad->flow_return != GST_FLOW_NOT_LINKED)) {
        /* return the error upstream */
        ret = tspad->flow_return;
        done = TRUE;
      }

    }

    if (ret == GST_FLOW_NOT_LINKED)
      ret = tspad->flow_return;

    g_object_unref (pad);

    if (G_UNLIKELY (!done)) {
      GST_OBJECT_LOCK (parse);
      if (G_UNLIKELY (pads_cookie != GST_ELEMENT_CAST (parse)->pads_cookie)) {
        /* resync */
        GST_DEBUG ("resync");
        pads_cookie = GST_ELEMENT_CAST (parse)->pads_cookie;
        srcpads = parse->srcpads;
      } else {
        GST_DEBUG ("getting next pad");
        /* Get next pad */
        srcpads = g_list_next (srcpads);
      }

      if (srcpads) {
        pad = GST_PAD_CAST (srcpads->data);
        g_object_ref (pad);
      } else
        done = TRUE;
      GST_OBJECT_UNLOCK (parse);
    }
  }

  gst_buffer_unref (buf);
  return ret;
}

static void
amlts_parse_inspect_packet (AmlTSBase * base, AmlTSPacketizerPacket * packet)
{
  AmlTSParse2 *parse = GST_AMLTS_PARSE (base);
  GST_LOG ("pid 0x%04x pusi:%d, afc:%d, cont:%d, payload:%p PCR %"
      G_GUINT64_FORMAT, packet->pid, packet->payload_unit_start_indicator,
      packet->scram_afc_cc & 0x30,
      FLAGS_CONTINUITY_COUNTER (packet->scram_afc_cc), packet->payload,
      packet->pcr);

  /* Store the PCR if desired */
  if (parse->current_pcr == GST_CLOCK_TIME_NONE &&
      packet->afc_flags & MPEGTS_AFC_PCR_FLAG) {
    /* Take this as the pcr_pid if set to auto-select */
    if (parse->pcr_pid == -1)
      parse->pcr_pid = packet->pid;
    /* Check the PCR-PID matches the program we want for multiple programs */
    if (parse->pcr_pid == packet->pid) {
      parse->current_pcr = amlts_packetizer_pts_to_ts (base->packetizer,
          PCRTIME_TO_GSTTIME (packet->pcr), parse->pcr_pid);
      GST_DEBUG ("Got new PCR %" GST_TIME_FORMAT " raw %" G_GUINT64_FORMAT,
          GST_TIME_ARGS (parse->current_pcr), packet->pcr);
      if (parse->base_pcr == GST_CLOCK_TIME_NONE) {
        parse->base_pcr = parse->current_pcr;
      }
    }
  }
}

static GstClockTime
aml_get_pending_timestamp_diff (AmlTSParse2 * parse)
{
  GList *l;
  GstClockTime first_ts, last_ts;

  if (parse->pending_buffers == NULL)
    return GST_CLOCK_TIME_NONE;

  l = g_list_last (parse->pending_buffers);
  first_ts = GST_BUFFER_PTS (l->data);
  if (first_ts == GST_CLOCK_TIME_NONE)
    return GST_CLOCK_TIME_NONE;

  l = g_list_first (parse->pending_buffers);
  last_ts = GST_BUFFER_PTS (l->data);
  if (last_ts == GST_CLOCK_TIME_NONE)
    return GST_CLOCK_TIME_NONE;

  return last_ts - first_ts;
}

static GstFlowReturn
aml_drain_pending_buffers (AmlTSParse2 * parse, gboolean drain_all)
{
  GstFlowReturn ret = GST_FLOW_OK;
  GstClockTime start_ts;
  GstClockTime pcr = GST_CLOCK_TIME_NONE;
  GstClockTime pcr_diff = 0;
  gsize pcr_bytes, bytes_since_pcr, pos;
  GstBuffer *buffer;
  GList *l, *end = NULL;

  if (parse->pending_buffers == NULL)
    return GST_FLOW_OK;         /* Nothing to push */

  /*
   * There are 4 cases:
   *  1 We get a buffer with no PCR -> it's the head of the list
   *      -> Do nothing, unless it's EOS
   *  2 We get a buffer with a PCR, it's the first PCR we've seen, and belongs
   *    to the buffer at the head of the list
   *    -> Push any buffers in the list except the head,
   *       using a smoothing of their timestamps to land at the PCR
   *    -> store new PCR as the previous PCR, bytes_since_pcr = sizeof (buffer);
   *  3 It's EOS (drain_all == TRUE, current_pcr == NONE)
   *    -> Push any buffers in the list using a smoothing of their timestamps
   *       starting at the previous PCR or first TS
   *  4 We get a buffer with a PCR, and have a previous PCR
   *    -> If distance > smoothing_latency,
   *       output buffers except the last in the pending queue using
   *       piecewise-linear timestamps
   *    -> store new PCR as the previous PCR, bytes_since_pcr = sizeof (buffer);
   */

  /* Case 1 */
  if (!GST_CLOCK_TIME_IS_VALID (parse->current_pcr) && !drain_all)
    return GST_FLOW_OK;

  if (GST_CLOCK_TIME_IS_VALID (parse->current_pcr)) {
    pcr = parse->current_pcr;
    parse->current_pcr = GST_CLOCK_TIME_NONE;
  }

  /* The bytes of the last buffer are after the PCR */
  buffer = GST_BUFFER (g_list_nth_data (parse->pending_buffers, 0));
  bytes_since_pcr = gst_buffer_get_size (buffer);

  pcr_bytes = parse->bytes_since_pcr - bytes_since_pcr;

  if (!drain_all)
    end = g_list_first (parse->pending_buffers);

  /* Case 2 */
  if (!GST_CLOCK_TIME_IS_VALID (parse->previous_pcr)) {
    pcr_diff = aml_get_pending_timestamp_diff (parse);

    /* Calculate the start_ts that ends at the end timestamp */
    start_ts = GST_CLOCK_TIME_NONE;
    if (end) {
      start_ts = GST_BUFFER_PTS (GST_BUFFER (end->data));
      if (start_ts > pcr_diff)
        start_ts -= pcr_diff;
    }
  } else if (drain_all) {       /* Case 3 */
    start_ts = parse->previous_pcr;
    pcr_diff = aml_get_pending_timestamp_diff (parse);
  } else {                      /* Case 4 */
    start_ts = parse->previous_pcr;
    if (GST_CLOCK_TIME_IS_VALID (pcr) && pcr > start_ts)
      pcr_diff = GST_CLOCK_DIFF (start_ts, pcr);

    /* Make sure PCR observations are sufficiently far apart */
    if (drain_all == FALSE && pcr_diff < parse->smoothing_latency)
      return GST_FLOW_OK;
  }

  GST_INFO_OBJECT (parse, "Pushing buffers - startTS %" GST_TIME_FORMAT
      " duration %" GST_TIME_FORMAT " %" G_GSIZE_FORMAT " bytes",
      GST_TIME_ARGS (start_ts), GST_TIME_ARGS (pcr_diff), pcr_bytes);

  /* Now, push buffers out pacing timestamps over pcr_diff time and pcr_bytes */
  pos = 0;
  l = g_list_last (parse->pending_buffers);
  while (l != end) {
    GList *p;
    GstClockTime out_ts = start_ts;

    buffer = gst_buffer_make_writable (GST_BUFFER (l->data));

    if (out_ts != GST_CLOCK_TIME_NONE && pcr_diff != GST_CLOCK_TIME_NONE &&
        pcr_bytes && pos)
      out_ts += gst_util_uint64_scale (pcr_diff, pos, pcr_bytes);

    pos += gst_buffer_get_size (buffer);

    GST_DEBUG_OBJECT (parse,
        "InputTS %" GST_TIME_FORMAT " out %" GST_TIME_FORMAT,
        GST_TIME_ARGS (GST_BUFFER_PTS (buffer)), GST_TIME_ARGS (out_ts));

    GST_BUFFER_PTS (buffer) = out_ts + parse->ts_offset;
    GST_BUFFER_DTS (buffer) = out_ts + parse->ts_offset;
    if (ret == GST_FLOW_OK) {
      ret =
          enqueue_and_maybe_push_buffer (parse, parse->srcpad,
          &parse->ts_adapter, buffer);
    } else {
      gst_buffer_unref (buffer);
    }

    /* Free this list node and move to the next */
    p = g_list_previous (l);
    parse->pending_buffers = g_list_delete_link (parse->pending_buffers, l);
    l = p;
  }

  if (parse->is_eos) {
    aml_empty_adapter_into_pad (parse, &parse->ts_adapter, parse->srcpad);
  }

  parse->pending_buffers = end;
  parse->bytes_since_pcr = bytes_since_pcr;
  parse->previous_pcr = pcr;
  return ret;
}

static GstFlowReturn
amlts_parse_have_buffer (AmlTSBase * base, GstBuffer * buffer)
{
  AmlTSParse2 *parse = GST_AMLTS_PARSE (base);
  GstFlowReturn ret = GST_FLOW_OK;

  GST_LOG_OBJECT (parse, "Received buffer %" GST_PTR_FORMAT, buffer);

  /* Assume all packets have equal size */
  if (parse->alignment > 0 &&
      base->packetizer->packet_size != MPEGTS_NORMAL_PACKETSIZE) {
    GstMapInfo map;
    guint8 *data;

    gst_buffer_map (buffer, &map, GST_MAP_READ);
    data = map.data;

    parse->header = GST_READ_UINT32_BE (data);
    gst_buffer_unmap (buffer, &map);
  }

  if (parse->current_pcr != GST_CLOCK_TIME_NONE) {
    GST_DEBUG_OBJECT (parse,
        "InputTS %" GST_TIME_FORMAT " PCR %" GST_TIME_FORMAT,
        GST_TIME_ARGS (GST_BUFFER_PTS (buffer)),
        GST_TIME_ARGS (parse->current_pcr));
  }

  if (parse->set_timestamps || parse->first) {
    parse->pending_buffers = g_list_prepend (parse->pending_buffers, buffer);
    parse->bytes_since_pcr += gst_buffer_get_size (buffer);
    buffer = NULL;
  }

  if (!aml_prepare_src_pad (base, parse))
    return GST_FLOW_OK;

  if (parse->pending_buffers != NULL) {
    /* Don't keep pending_buffers if not setting output timestamps */
    gboolean drain_all = (parse->set_timestamps == FALSE);
    ret = aml_drain_pending_buffers (parse, drain_all);
    if (ret != GST_FLOW_OK) {
      if (buffer)
        gst_buffer_unref (buffer);
      return ret;
    }
  }

  ret =
      enqueue_and_maybe_push_buffer (parse, parse->srcpad,
      &parse->ts_adapter, buffer);
  return ret;
}

static void
aml_empty_pad (GstPad * pad, AmlTSParse2 * parse)
{
  AmlTSParsePad *tspad = (AmlTSParsePad *) gst_pad_get_element_private (pad);
  GstFlowReturn ret;
  ret = aml_empty_adapter_into_pad (parse, &tspad->ts_adapter, tspad->pad);
  ret = gst_flow_combiner_update_flow (parse->flowcombiner, ret);
}

static GstFlowReturn
amlts_parse_input_done (AmlTSBase * base)
{
  AmlTSParse2 *parse = GST_AMLTS_PARSE (base);
  GstFlowReturn ret = GST_FLOW_OK;

  if (!aml_prepare_src_pad (base, parse))
    return GST_FLOW_OK;

  if (parse->alignment == 0) {
    ret = aml_empty_adapter_into_pad (parse, &parse->ts_adapter, parse->srcpad);
    ret = gst_flow_combiner_update_flow (parse->flowcombiner, ret);
    g_list_foreach (parse->srcpads, (GFunc) aml_empty_pad, parse);
  }
  return ret;
}

static AmlTSParsePad *
aml_find_pad_for_program (AmlTSParse2 * parse, guint program_number)
{
  GList *tmp;

  for (tmp = parse->srcpads; tmp; tmp = tmp->next) {
    AmlTSParsePad *tspad = gst_pad_get_element_private ((GstPad *) tmp->data);

    if (tspad->program_number == program_number)
      return tspad;
  }

  return NULL;
}

static void
amlts_parse_program_started (AmlTSBase * base, AmlTSBaseProgram * program)
{
  AmlTSParse2 *parse = GST_AMLTS_PARSE (base);
  AmlTSParseProgram *parseprogram = (AmlTSParseProgram *) program;
  AmlTSParsePad *tspad;

  /* If we have a request pad for that program, activate it */
  tspad = aml_find_pad_for_program (parse, program->program_number);

  if (tspad) {
    tspad->program = parseprogram;
    parseprogram->tspad = tspad;
  }
}

static void
amlts_parse_program_stopped (AmlTSBase * base, AmlTSBaseProgram * program)
{
  AmlTSParse2 *parse = GST_AMLTS_PARSE (base);
  AmlTSParseProgram *parseprogram = (AmlTSParseProgram *) program;
  AmlTSParsePad *tspad;

  /* If we have a request pad for that program, activate it */
  tspad = aml_find_pad_for_program (parse, program->program_number);

  if (tspad) {
    tspad->program = NULL;
    parseprogram->tspad = NULL;
  }

  parse->pcr_pid = -1;
  parse->ts_offset += parse->current_pcr - parse->base_pcr;
  parse->base_pcr = GST_CLOCK_TIME_NONE;
}

static gboolean
amlts_parse_src_pad_query (GstPad * pad, GstObject * parent, GstQuery * query)
{
  AmlTSParse2 *parse = GST_AMLTS_PARSE (parent);
  gboolean res;

  switch (GST_QUERY_TYPE (query)) {
    case GST_QUERY_LATENCY:
    {
      if ((res = gst_pad_peer_query (((AmlTSBase *) parse)->sinkpad, query))) {
        gboolean is_live;
        GstClockTime min_latency, max_latency;

        gst_query_parse_latency (query, &is_live, &min_latency, &max_latency);
        if (is_live) {
          GstClockTime extra_latency = TS_LATENCY * GST_MSECOND;
          if (parse->set_timestamps) {
            extra_latency = MAX (extra_latency, parse->smoothing_latency);
          }
          min_latency += extra_latency;
          if (max_latency != GST_CLOCK_TIME_NONE)
            max_latency += extra_latency;
        }

        gst_query_set_latency (query, is_live, min_latency, max_latency);
      }
      break;
    }
    default:
      res = gst_pad_query_default (pad, parent, query);
  }
  return res;
}

gboolean
gst_amltsparse_plugin_init (GstPlugin * plugin)
{
  GST_DEBUG_CATEGORY_INIT (mpegts_parse_debug, "amltsparse", 0,
      "MPEG transport stream parser");

  return gst_element_register (plugin, "amltsparse",
      GST_RANK_NONE, GST_TYPE_AMLTS_PARSE);
}