tsdemux: Fix several leaks

* dont' leak buffers when a stream is in discont state
* don't leak buffers when a program is removed/deactivated
* remove all programs when disposing

gst/mpegtsdemux/mpegtsbase.c

@@ -93,6 +93,9 @@ static void mpegts_base_get_tags_from_sdt (MpegTSBase * base,
     GstStructure * sdt_info);
 static void mpegts_base_get_tags_from_eit (MpegTSBase * base,
     GstStructure * eit_info);
+static gboolean
+remove_each_program (gpointer key, MpegTSBaseProgram * program,
+    MpegTSBase * base);
 
 GST_BOILERPLATE_FULL (MpegTSBase, mpegts_base, GstElement, GST_TYPE_ELEMENT,
     _extra_init);
@@ -226,6 +229,9 @@ mpegts_base_reset (MpegTSBase * base)
   base->upstream_live = FALSE;
   base->queried_latency = FALSE;
 
+  g_hash_table_foreach_remove (base->programs, (GHRFunc) remove_each_program,
+      base);
+
   if (klass->reset)
     klass->reset (base);
 }
@@ -1237,20 +1243,26 @@ mpegts_base_get_tags_from_eit (MpegTSBase * base, GstStructure * eit_info)
   }
 }
 
-static void
+static gboolean
 remove_each_program (gpointer key, MpegTSBaseProgram * program,
     MpegTSBase * base)
 {
+  MpegTSBaseClass *klass = GST_MPEGTS_BASE_GET_CLASS (base);
+
   /* First deactivate it */
   mpegts_base_deactivate_program (base, program);
   /* Then remove it */
-  mpegts_base_remove_program (base, program->program_number);
+  if (klass->program_stopped)
+    klass->program_stopped (base, program);
+
+  return TRUE;
 }
 
 static gboolean
 gst_mpegts_base_handle_eos (MpegTSBase * base)
 {
-  g_hash_table_foreach (base->programs, (GHFunc) remove_each_program, base);
+  g_hash_table_foreach_remove (base->programs, (GHRFunc) remove_each_program,
+      base);
   /* finally remove  */
   return TRUE;
 }

gst/mpegtsdemux/tsdemux.c

@@ -249,6 +249,7 @@ static void gst_ts_demux_get_property (GObject * object, guint prop_id,
 static void gst_ts_demux_flush_streams (GstTSDemux * tsdemux);
 static GstFlowReturn
 gst_ts_demux_push_pending_data (GstTSDemux * demux, TSDemuxStream * stream);
+static void gst_ts_demux_stream_flush (TSDemuxStream * stream);
 
 static gboolean push_event (MpegTSBase * base, GstEvent * event);
 static void _extra_init (GType type);
@@ -998,6 +999,7 @@ gst_ts_demux_stream_removed (MpegTSBase * base, MpegTSBaseStream * bstream)
     }
     stream->pad = NULL;
   }
+  gst_ts_demux_stream_flush (stream);
   stream->flow_return = GST_FLOW_NOT_LINKED;
 }
 
@@ -1030,6 +1032,12 @@ gst_ts_demux_stream_flush (TSDemuxStream * stream)
   memset (stream->pendingbuffers, 0, TS_MAX_PENDING_BUFFERS);
   stream->nbpending = 0;
 
+  if (stream->currentlist) {
+    g_list_foreach (stream->currentlist, (GFunc) gst_buffer_unref, NULL);
+    g_list_free (stream->currentlist);
+    stream->currentlist = NULL;
+  }
+
   stream->expected_size = 0;
   stream->current_size = 0;
   stream->current = NULL;
@@ -1346,7 +1354,9 @@ gst_ts_demux_queue_data (GstTSDemux * demux, TSDemuxStream * stream,
     }
   }
 
-  if (stream->state == PENDING_PACKET_HEADER) {
+  switch (stream->state) {
+    case PENDING_PACKET_HEADER:
+    {
       GST_LOG ("HEADER: appending data to array");
       /* Append to the array */
       stream->pendingbuffers[stream->nbpending++] = buf;
@@ -1354,12 +1364,24 @@ gst_ts_demux_queue_data (GstTSDemux * demux, TSDemuxStream * stream,
 
       /* parse the header */
       gst_ts_demux_parse_pes_header (demux, stream);
-  } else if (stream->state == PENDING_PACKET_BUFFER) {
+      break;
+    }
+    case PENDING_PACKET_BUFFER:
+    {
       GST_LOG ("BUFFER: appending data to bufferlist");
       stream->currentlist = g_list_prepend (stream->currentlist, buf);
       stream->current_size += GST_BUFFER_SIZE (buf);
+      break;
+    }
+    case PENDING_PACKET_DISCONT:
+    {
+      GST_LOG ("DISCONT: dropping buffer");
+      gst_buffer_unref (packet->buffer);
+      break;
+    }
+    default:
+      break;
   }
-
 
   return;
 }
@@ -1464,8 +1486,10 @@ gst_ts_demux_push_pending_data (GstTSDemux * demux, TSDemuxStream * stream)
     goto beach;
   }
 
-  if (G_UNLIKELY (stream->state != PENDING_PACKET_BUFFER))
+  if (G_UNLIKELY (stream->state != PENDING_PACKET_BUFFER)) {
+    GST_LOG ("state:%d, returning", stream->state);
     goto beach;
+  }
 
   if (G_UNLIKELY (!stream->active))
     activate_pad_for_stream (demux, stream);
@@ -1485,6 +1509,7 @@ gst_ts_demux_push_pending_data (GstTSDemux * demux, TSDemuxStream * stream)
   GST_LOG_OBJECT (stream->pad, "Putting pending data into GstBufferList");
   stream->currentlist = g_list_reverse (stream->currentlist);
   gst_buffer_list_iterator_add_list (stream->currentit, stream->currentlist);
+  stream->currentlist = NULL;
   gst_buffer_list_iterator_free (stream->currentit);
 
   firstbuffer = gst_buffer_list_get (stream->current, 0, 0);