gst/wavparse/gstwavparse.*: Add push (streaming) mode to wavparse (fixes #337625)

Original commit message from CVS:
* gst/wavparse/gstwavparse.c: (gst_wavparse_base_init),
(gst_wavparse_class_init), (gst_wavparse_dispose),
(gst_wavparse_reset), (gst_wavparse_init),
(gst_wavparse_create_sourcepad), (gst_wavparse_parse_adtl),
(gst_wavparse_parse_cues), (gst_wavparse_parse_file_header),
(gst_wavparse_stream_init), (gst_wavparse_perform_seek),
(gst_wavparse_peek_chunk_info), (gst_wavparse_peek_chunk),
(gst_wavparse_stream_headers), (gst_wavparse_parse_stream_init),
(gst_wavparse_send_event), (gst_wavparse_add_src_pad),
(gst_wavparse_stream_data), (gst_wavparse_loop),
(gst_wavparse_chain), (gst_wavparse_srcpad_event),
(gst_wavparse_sink_activate), (gst_wavparse_sink_activate_pull),
(gst_wavparse_change_state), (plugin_init):
* gst/wavparse/gstwavparse.h:
Add push (streaming) mode to wavparse (fixes #337625)

ChangeLog

@@ -1,3 +1,21 @@
+2006-04-29  Stefan Kost  <ensonic@users.sf.net>
+
+	* gst/wavparse/gstwavparse.c: (gst_wavparse_base_init),
+	(gst_wavparse_class_init), (gst_wavparse_dispose),
+	(gst_wavparse_reset), (gst_wavparse_init),
+	(gst_wavparse_create_sourcepad), (gst_wavparse_parse_adtl),
+	(gst_wavparse_parse_cues), (gst_wavparse_parse_file_header),
+	(gst_wavparse_stream_init), (gst_wavparse_perform_seek),
+	(gst_wavparse_peek_chunk_info), (gst_wavparse_peek_chunk),
+	(gst_wavparse_stream_headers), (gst_wavparse_parse_stream_init),
+	(gst_wavparse_send_event), (gst_wavparse_add_src_pad),
+	(gst_wavparse_stream_data), (gst_wavparse_loop),
+	(gst_wavparse_chain), (gst_wavparse_srcpad_event),
+	(gst_wavparse_sink_activate), (gst_wavparse_sink_activate_pull),
+	(gst_wavparse_change_state), (plugin_init):
+	* gst/wavparse/gstwavparse.h:
+	  Add push (streaming) mode to wavparse (fixes #337625)
+
 2006-04-28  Thomas Vander Stichele  <thomas at apestaart dot org>
 
 	* configure.ac:

gst/wavparse/gstwavparse.c

@@ -1,6 +1,7 @@
 /* -*- Mode: C; tab-width: 2; indent-tabs-mode: t; c-basic-offset: 2 -*- */
 /* GStreamer
  * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
+ * Copyright (C) <2006> Nokia Corporation, Stefan Kost <stefan.kost@nokia.com>.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Library General Public
@@ -31,11 +32,18 @@
  * <title>Example launch line</title>
  * <para>
  * <programlisting>
- * gst-launch filesrc sine.wav ! wavparse ! audioconvert ! alsasink
+ * gst-launch filesrc location=sine.wav ! wavparse ! audioconvert ! alsasink
  * </programlisting>
  * Read a wav file and output to the soundcard using the ALSA element. The
  * wav file is assumed to contain raw uncompressed samples.
  * </para>
+ * <para>
+ * <programlisting>
+ * gst-launch gnomevfssrc location=http://www.example.org/sine.wav ! queue ! wavparse ! audioconvert ! alsasink
+ * </programlisting>
+ * Stream data from
+ * </para>
+ *
  * </refsect2>
  *
  * Last reviewed on 2006-03-03 (0.10.3)
@@ -51,18 +59,24 @@
 #include "gst/riff/riff-media.h"
 #include <gst/gst-i18n-plugin.h>
 
+#ifndef G_MAXUINT32
+#define G_MAXUINT32 0xffffffff
+#endif
+
 GST_DEBUG_CATEGORY_STATIC (wavparse_debug);
 #define GST_CAT_DEFAULT (wavparse_debug)
 
 static void gst_wavparse_base_init (gpointer g_class);
 static void gst_wavparse_class_init (GstWavParseClass * klass);
 static void gst_wavparse_init (GstWavParse * wavparse);
+static void gst_wavparse_dispose (GObject * object);
 
 static gboolean gst_wavparse_sink_activate (GstPad * sinkpad);
 static gboolean gst_wavparse_sink_activate_pull (GstPad * sinkpad,
     gboolean active);
 static gboolean gst_wavparse_send_event (GstElement * element,
     GstEvent * event);
+static GstFlowReturn gst_wavparse_chain (GstPad * pad, GstBuffer * buf);
 static GstStateChangeReturn gst_wavparse_change_state (GstElement * element,
     GstStateChange transition);
 
@@ -77,6 +91,12 @@ static gboolean gst_wavparse_srcpad_event (GstPad * pad, GstEvent * event);
 static void gst_wavparse_get_property (GObject * object, guint prop_id,
     GValue * value, GParamSpec * pspec);
 
+static const GstElementDetails gst_wavparse_details =
+GST_ELEMENT_DETAILS ("WAV audio demuxer",
+    "Codec/Demuxer/Audio",
+    "Parse a .wav file into raw audio",
+    "Erik Walthinsen <omega@cse.ogi.edu>");
+
 static GstStaticPadTemplate sink_template_factory =
 GST_STATIC_PAD_TEMPLATE ("wavparse_sink",
     GST_PAD_SINK,
@@ -155,22 +175,13 @@ static void
 gst_wavparse_base_init (gpointer g_class)
 {
   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
-  GstPadTemplate *templ;
-  static GstElementDetails gst_wavparse_details =
-      GST_ELEMENT_DETAILS ("WAV audio demuxer",
-      "Codec/Demuxer/Audio",
-      "Parse a .wav file into raw audio",
-      "Erik Walthinsen <omega@cse.ogi.edu>");
-
-  gst_element_class_set_details (element_class, &gst_wavparse_details);
 
   /* register src pads */
-  templ = gst_static_pad_template_get (&sink_template_factory);
+  gst_element_class_add_pad_template (element_class,
-  gst_element_class_add_pad_template (element_class, templ);
+      gst_static_pad_template_get (&sink_template_factory));
-  gst_object_unref (templ);
+  gst_element_class_add_pad_template (element_class,
-  templ = gst_static_pad_template_get (&src_template_factory);
+      gst_static_pad_template_get (&src_template_factory));
-  gst_element_class_add_pad_template (element_class, templ);
+  gst_element_class_set_details (element_class, &gst_wavparse_details);
-  gst_object_unref (templ);
 }
 
 static void
@@ -185,13 +196,28 @@ gst_wavparse_class_init (GstWavParseClass * klass)
   parent_class = g_type_class_peek_parent (klass);
 
   object_class->get_property = gst_wavparse_get_property;
+  object_class->dispose = gst_wavparse_dispose;
 
   gstelement_class->change_state = gst_wavparse_change_state;
   gstelement_class->send_event = gst_wavparse_send_event;
-
-  GST_DEBUG_CATEGORY_INIT (wavparse_debug, "wavparse", 0, "WAV parser");
 }
 
+
+static void
+gst_wavparse_dispose (GObject * object)
+{
+  GST_DEBUG ("WAV: Dispose\n");
+  GstWavParse *wav = GST_WAVPARSE (object);
+
+  if (wav->adapter) {
+    g_object_unref (wav->adapter);
+    wav->adapter = NULL;
+  }
+
+  G_OBJECT_CLASS (parent_class)->dispose (object);
+}
+
+
 static void
 gst_wavparse_reset (GstWavParse * wavparse)
 {
@@ -209,6 +235,12 @@ gst_wavparse_reset (GstWavParse * wavparse)
   wavparse->dataleft = 0;
   wavparse->datasize = 0;
   wavparse->datastart = 0;
+  wavparse->got_fmt = FALSE;
+  wavparse->first = TRUE;
+
+  if (wavparse->seek_event)
+    gst_event_unref (wavparse->seek_event);
+  wavparse->seek_event = NULL;
 
   /* we keep the segment info in time */
   gst_segment_init (&wavparse->segment, GST_FORMAT_TIME);
@@ -226,7 +258,12 @@ gst_wavparse_init (GstWavParse * wavparse)
       GST_DEBUG_FUNCPTR (gst_wavparse_sink_activate));
   gst_pad_set_activatepull_function (wavparse->sinkpad,
       GST_DEBUG_FUNCPTR (gst_wavparse_sink_activate_pull));
+  gst_pad_set_chain_function (wavparse->sinkpad,
+      GST_DEBUG_FUNCPTR (gst_wavparse_chain));
   gst_element_add_pad (GST_ELEMENT (wavparse), wavparse->sinkpad);
+
+  /* src, will be created later */
+  wavparse->srcpad = NULL;
 }
 
 static void
@@ -254,6 +291,8 @@ gst_wavparse_create_sourcepad (GstWavParse * wavparse)
       GST_DEBUG_FUNCPTR (gst_wavparse_pad_query));
   gst_pad_set_event_function (wavparse->srcpad,
       GST_DEBUG_FUNCPTR (gst_wavparse_srcpad_event));
+
+  GST_DEBUG_OBJECT (wavparse, "srcpad created");
 }
 
 static void
@@ -270,6 +309,8 @@ gst_wavparse_get_property (GObject * object,
   }
 }
 
+
+
 #if 0
 static void
 gst_wavparse_parse_adtl (GstWavParse * wavparse, int len)
@@ -453,9 +494,7 @@ gst_wavparse_parse_adtl (GstWavParse * wavparse, int len)
 
   g_object_notify (G_OBJECT (wavparse), "metadata");
 }
-#endif
 
-#if 0
 static void
 gst_wavparse_parse_cues (GstWavParse * wavparse, int len)
 {
@@ -518,49 +557,7 @@ gst_wavparse_parse_cues (GstWavParse * wavparse, int len)
 
   g_object_notify (G_OBJECT (wavparse), "metadata");
 }
-#endif
 
-static gboolean
-gst_wavparse_parse_file_header (GstElement * element, GstBuffer * buf)
-{
-  guint32 doctype;
-
-  if (!gst_riff_parse_file_header (element, buf, &doctype))
-    return FALSE;
-
-  if (doctype != GST_RIFF_RIFF_WAVE)
-    goto not_wav;
-
-  return TRUE;
-
-  /* ERRORS */
-not_wav:
-  {
-    GST_ELEMENT_ERROR (element, STREAM, WRONG_TYPE, (NULL),
-        ("File is not an WAVE file: %" GST_FOURCC_FORMAT,
-            GST_FOURCC_ARGS (doctype)));
-    return FALSE;
-  }
-}
-
-static GstFlowReturn
-gst_wavparse_stream_init (GstWavParse * wav)
-{
-  GstFlowReturn res;
-  GstBuffer *buf = NULL;
-
-  if ((res = gst_pad_pull_range (wav->sinkpad,
-              wav->offset, 12, &buf)) != GST_FLOW_OK)
-    return res;
-  else if (!gst_wavparse_parse_file_header (GST_ELEMENT (wav), buf))
-    return GST_FLOW_ERROR;
-
-  wav->offset += 12;
-
-  return GST_FLOW_OK;
-}
-
-#if 0
 /* Read 'fmt ' header */
 static gboolean
 gst_wavparse_fmt (GstWavParse * wav)
@@ -718,6 +715,48 @@ gst_wavparse_other (GstWavParse * wav)
 }
 #endif
 
+
+
+static gboolean
+gst_wavparse_parse_file_header (GstElement * element, GstBuffer * buf)
+{
+  guint32 doctype;
+
+  if (!gst_riff_parse_file_header (element, buf, &doctype))
+    return FALSE;
+
+  if (doctype != GST_RIFF_RIFF_WAVE)
+    goto not_wav;
+
+  return TRUE;
+
+  /* ERRORS */
+not_wav:
+  {
+    GST_ELEMENT_ERROR (element, STREAM, WRONG_TYPE, (NULL),
+        ("File is not an WAVE file: %" GST_FOURCC_FORMAT,
+            GST_FOURCC_ARGS (doctype)));
+    return FALSE;
+  }
+}
+
+static GstFlowReturn
+gst_wavparse_stream_init (GstWavParse * wav)
+{
+  GstFlowReturn res;
+  GstBuffer *buf = NULL;
+
+  if ((res = gst_pad_pull_range (wav->sinkpad,
+              wav->offset, 12, &buf)) != GST_FLOW_OK)
+    return res;
+  else if (!gst_wavparse_parse_file_header (GST_ELEMENT (wav), buf))
+    return GST_FLOW_ERROR;
+
+  wav->offset += 12;
+
+  return GST_FLOW_OK;
+}
+
 /* This function is used to perform seeks on the element in 
  * pull mode.
  *
@@ -740,7 +779,6 @@ gst_wavparse_perform_seek (GstWavParse * wav, GstEvent * event)
   gboolean update;
   GstSegment seeksegment;
 
-
   if (event) {
     GST_DEBUG_OBJECT (wav, "doing seek with event");
 
@@ -770,10 +808,12 @@ gst_wavparse_perform_seek (GstWavParse * wav, GstEvent * event)
 
   flush = flags & GST_SEEK_FLAG_FLUSH;
 
-  if (flush)
+  if (flush && wav->srcpad) {
+    GST_DEBUG_OBJECT (wav, "sending flush start");
     gst_pad_push_event (wav->srcpad, gst_event_new_flush_start ());
-  else
+  } else {
     gst_pad_pause_task (wav->sinkpad);
+  }
 
   GST_PAD_STREAM_LOCK (wav->sinkpad);
 
@@ -814,18 +854,21 @@ gst_wavparse_perform_seek (GstWavParse * wav, GstEvent * event)
       GST_TIME_ARGS (seeksegment.start), GST_TIME_ARGS (stop));
 
   /* prepare for streaming again */
-  if (flush) {
+  if (wav->srcpad) {
-    gst_pad_push_event (wav->srcpad, gst_event_new_flush_stop ());
+    if (flush) {
-  } else if (wav->segment_running) {
+      GST_DEBUG_OBJECT (wav, "sending flush stop");
-    /* we are running the current segment and doing a non-flushing seek,
+      gst_pad_push_event (wav->srcpad, gst_event_new_flush_stop ());
-     * close the segment first based on the last_stop. */
+    } else if (wav->segment_running) {
-    GST_DEBUG_OBJECT (wav, "closing running segment %" G_GINT64_FORMAT
+      /* we are running the current segment and doing a non-flushing seek,
-        " to %" G_GINT64_FORMAT, wav->segment.start, wav->segment.last_stop);
+       * close the segment first based on the last_stop. */
+      GST_DEBUG_OBJECT (wav, "closing running segment %" G_GINT64_FORMAT
+          " to %" G_GINT64_FORMAT, wav->segment.start, wav->segment.last_stop);
 
-    gst_pad_push_event (wav->srcpad,
+      gst_pad_push_event (wav->srcpad,
-        gst_event_new_new_segment (TRUE,
+          gst_event_new_new_segment (TRUE,
-            wav->segment.rate, wav->segment.format,
+              wav->segment.rate, wav->segment.format,
-            wav->segment.start, wav->segment.last_stop, wav->segment.time));
+              wav->segment.start, wav->segment.last_stop, wav->segment.time));
+    }
   }
 
   memcpy (&wav->segment, &seeksegment, sizeof (GstSegment));
@@ -853,8 +896,10 @@ gst_wavparse_perform_seek (GstWavParse * wav, GstEvent * event)
   }
 
   wav->segment_running = TRUE;
-  gst_pad_start_task (wav->sinkpad, (GstTaskFunction) gst_wavparse_loop,
+  if (!wav->streaming) {
-      wav->sinkpad);
+    gst_pad_start_task (wav->sinkpad, (GstTaskFunction) gst_wavparse_loop,
+        wav->sinkpad);
+  }
 
   GST_PAD_STREAM_UNLOCK (wav->sinkpad);
 
@@ -868,6 +913,59 @@ no_format:
   }
 }
 
+/*
+ * gst_wavparse_peek_chunk_info:
+ * @wav Wavparse object
+ * @tag holder for tag
+ * @size holder for tag size
+ *                         
+ * Peek next chunk info (tag and size)                         
+ *
+ * Returns: %TRUE when one chunk info has been got from the adapter
+ */
+static gboolean
+gst_wavparse_peek_chunk_info (GstWavParse * wav, guint32 * tag, guint32 * size)
+{
+  const guint8 *data = NULL;
+
+  if (gst_adapter_available (wav->adapter) < 8) {
+    return FALSE;
+  }
+
+  GST_DEBUG ("Next chunk size is %d bytes", *size);
+  data = gst_adapter_peek (wav->adapter, 8);
+  *tag = GST_READ_UINT32_LE (data);
+  *size = GST_READ_UINT32_LE (data + 4);
+
+  return TRUE;
+}
+
+/*
+ * gst_wavparse_peek_chunk:
+ * @wav Wavparse object
+ * @tag holder for tag
+ * @size holder for tag size
+ *
+ * Peek enough data for one full chunk
+ *
+ * Returns: %TRUE when one chunk has been got
+ */
+static gboolean
+gst_wavparse_peek_chunk (GstWavParse * wav, guint32 * tag, guint32 * size)
+{
+  guint32 peek_size = 0;
+
+  gst_wavparse_peek_chunk_info (wav, tag, size);
+  GST_DEBUG ("Need to peek chunk of %d bytes", *size);
+  peek_size = (*size + 1) & ~1;
+
+  if (gst_adapter_available (wav->adapter) >= (8 + peek_size)) {
+    return TRUE;
+  } else {
+    return FALSE;
+  }
+}
+
 static gboolean
 gst_wavparse_get_upstream_size (GstWavParse * wav, gint64 * len)
 {
@@ -887,97 +985,127 @@ static GstFlowReturn
 gst_wavparse_stream_headers (GstWavParse * wav)
 {
   GstFlowReturn res;
-  GstBuffer *buf, *extra;
+  GstBuffer *buf;
   gst_riff_strf_auds *header = NULL;
-  guint32 tag;
+  guint32 tag, size;
   gboolean gotdata = FALSE;
   GstCaps *caps;
   gint64 duration;
   gchar *codec_name = NULL;
   GstEvent **event_p;
 
-  /* The header start with a 'fmt ' tag */
+  if (!wav->got_fmt) {
-  if ((res = gst_riff_read_chunk (GST_ELEMENT (wav), wav->sinkpad,
+    GstBuffer *extra;
-              &wav->offset, &tag, &buf)) != GST_FLOW_OK)
-    return res;
 
-  else if (tag != GST_RIFF_TAG_fmt)
+    /* The header start with a 'fmt ' tag */
-    goto invalid_wav;
 
-  if (!(gst_riff_parse_strf_auds (GST_ELEMENT (wav), buf, &header, &extra)))
+    if (wav->streaming) {
-    goto parse_header_error;
+      if (!gst_wavparse_peek_chunk (wav, &tag, &size))
+        return GST_FLOW_OK;
 
-  /* Note: gst_riff_create_audio_caps might nedd to fix values in
+      buf = gst_buffer_new ();
-   * the header header depending on the format, so call it first */
+      gst_buffer_ref (buf);
-  caps =
+      gst_adapter_flush (wav->adapter, 8);
-      gst_riff_create_audio_caps (header->format, NULL, header, extra,
+      wav->offset += 8;
-      NULL, &codec_name);
+      GST_BUFFER_DATA (buf) = (guint8 *) gst_adapter_peek (wav->adapter, size);
+      GST_BUFFER_SIZE (buf) = size;
 
-  if (extra)
+    } else {
-    gst_buffer_unref (extra);
+      if ((res = gst_riff_read_chunk (GST_ELEMENT (wav), wav->sinkpad,
+                  &wav->offset, &tag, &buf)) != GST_FLOW_OK)
+        return res;
+    }
 
-  wav->format = header->format;
+    if (tag != GST_RIFF_TAG_fmt)
-  wav->rate = header->rate;
+      goto invalid_wav;
-  wav->channels = header->channels;
 
-  if (wav->channels == 0)
+    if (!(gst_riff_parse_strf_auds (GST_ELEMENT (wav), buf, &header, &extra)))
-    goto no_channels;
+      goto parse_header_error;
 
-  wav->blockalign = header->blockalign;
+    if (extra)
-  wav->width = (header->blockalign * 8) / header->channels;
+      gst_buffer_unref (extra);
-  wav->depth = header->size;
-  wav->bps = header->av_bps;
 
-  if (wav->bps <= 0)
+    if (wav->streaming) {
-    goto no_bitrate;
+      gst_adapter_flush (wav->adapter, size);
+      wav->offset += size;
+      GST_BUFFER_DATA (buf) = NULL;
+      gst_buffer_unref (buf);
+    }
 
-  wav->bytes_per_sample = wav->channels * wav->width / 8;
+    /* Note: gst_riff_create_audio_caps might nedd to fix values in
-  if (wav->bytes_per_sample <= 0)
+     * the header header depending on the format, so call it first */
-    goto no_bytes_per_sample;
+    caps =
+        gst_riff_create_audio_caps (header->format, NULL, header, NULL,
+        NULL, &codec_name);
 
-  g_free (header);
+    wav->format = header->format;
+    wav->rate = header->rate;
+    wav->channels = header->channels;
 
-  if (!caps)
+    if (wav->channels == 0)
-    goto unknown_format;
+      goto no_channels;
 
-  GST_DEBUG_OBJECT (wav, "blockalign = %u", (guint) wav->blockalign);
+    wav->blockalign = header->blockalign;
-  GST_DEBUG_OBJECT (wav, "width      = %u", (guint) wav->width);
+    wav->width = (header->blockalign * 8) / header->channels;
-  GST_DEBUG_OBJECT (wav, "depth      = %u", (guint) wav->depth);
+    wav->depth = header->size;
-  GST_DEBUG_OBJECT (wav, "bps        = %u", (guint) wav->bps);
+    wav->bps = header->av_bps;
 
-  /* create pad later so we can sniff the first few bytes
+    if (wav->bps <= 0)
-   * of the real data and correct our caps if necessary */
+      goto no_bitrate;
-  gst_caps_replace (&wav->caps, caps);
-  gst_caps_replace (&caps, NULL);
 
-  if (codec_name) {
+    wav->bytes_per_sample = wav->channels * wav->width / 8;
-    wav->tags = gst_tag_list_new ();
+    if (wav->bytes_per_sample <= 0)
+      goto no_bytes_per_sample;
 
-    gst_tag_list_add (wav->tags, GST_TAG_MERGE_REPLACE,
+    g_free (header);
-        GST_TAG_AUDIO_CODEC, codec_name, NULL);
 
-    g_free (codec_name);
+    if (!caps)
-    codec_name = NULL;
+      goto unknown_format;
+
+    GST_DEBUG_OBJECT (wav, "blockalign = %u", (guint) wav->blockalign);
+    GST_DEBUG_OBJECT (wav, "width      = %u", (guint) wav->width);
+    GST_DEBUG_OBJECT (wav, "depth      = %u", (guint) wav->depth);
+    GST_DEBUG_OBJECT (wav, "bps        = %u", (guint) wav->bps);
+
+    /* create pad later so we can sniff the first few bytes
+     * of the real data and correct our caps if necessary */
+    gst_caps_replace (&wav->caps, caps);
+    gst_caps_replace (&caps, NULL);
+
+    wav->got_fmt = TRUE;
+
+    if (codec_name) {
+      wav->tags = gst_tag_list_new ();
+
+      gst_tag_list_add (wav->tags, GST_TAG_MERGE_REPLACE,
+          GST_TAG_AUDIO_CODEC, codec_name, NULL);
+
+      g_free (codec_name);
+      codec_name = NULL;
+    }
+
+    GST_DEBUG_OBJECT (wav, "frequency %d, channels %d", wav->rate,
+        wav->channels);
   }
 
-  GST_DEBUG_OBJECT (wav, "frequency %d, channels %d", wav->rate, wav->channels);
-
   /* loop headers until we get data */
   while (!gotdata) {
-    guint size;
+    if (wav->streaming) {
-    guint32 tag;
+      if (!gst_wavparse_peek_chunk_info (wav, &tag, &size))
-
+        return GST_FLOW_OK;
-    if ((res =
+    } else {
-            gst_pad_pull_range (wav->sinkpad, wav->offset, 8,
+      if ((res =
-                &buf)) != GST_FLOW_OK)
+              gst_pad_pull_range (wav->sinkpad, wav->offset, 8,
-      goto header_read_error;
+                  &buf)) != GST_FLOW_OK)
+        goto header_read_error;
+      tag = GST_READ_UINT32_LE (GST_BUFFER_DATA (buf));
+      size = GST_READ_UINT32_LE (GST_BUFFER_DATA (buf) + 4);
+    }
 
     /*
        wav is a st00pid format, we don't know for sure where data starts.
        So we have to go bit by bit until we find the 'data' header
      */
-    tag = GST_READ_UINT32_LE (GST_BUFFER_DATA (buf));
-    size = GST_READ_UINT32_LE (GST_BUFFER_DATA (buf) + 4);
 
     switch (tag) {
         /* TODO : Implement the various cases */
@@ -986,6 +1114,11 @@ gst_wavparse_stream_headers (GstWavParse * wav)
 
         GST_DEBUG_OBJECT (wav, "Got 'data' TAG, size : %d", size);
         gotdata = TRUE;
+        if (wav->streaming) {
+          gst_adapter_flush (wav->adapter, 8);
+        } else {
+          gst_buffer_unref (buf);
+        }
         wav->offset += 8;
         wav->datastart = wav->offset;
         /* file might be truncated */
@@ -998,12 +1131,19 @@ gst_wavparse_stream_headers (GstWavParse * wav)
         break;
       }
       default:
+        if (wav->streaming) {
+          if (!gst_wavparse_peek_chunk (wav, &tag, &size))
+            return GST_FLOW_OK;
+        }
         GST_DEBUG_OBJECT (wav, "Ignoring tag %" GST_FOURCC_FORMAT,
             GST_FOURCC_ARGS (tag));
         wav->offset += 8 + ((size + 1) & ~1);
-        break;
+        if (wav->streaming) {
+          gst_adapter_flush (wav->adapter, 8 + ((size + 1) & ~1));
+        } else {
+          gst_buffer_unref (buf);
+        }
     }
-    gst_buffer_unref (buf);
   }
 
   GST_DEBUG_OBJECT (wav, "Finished parsing headers");
@@ -1021,6 +1161,7 @@ gst_wavparse_stream_headers (GstWavParse * wav)
   event_p = &wav->seek_event;
   gst_event_replace (event_p, NULL);
 
+  wav->state = GST_WAVPARSE_DATA;
   return GST_FLOW_OK;
 
   /* ERROR */
@@ -1080,6 +1221,32 @@ header_read_error:
   }
 }
 
+
+/*                       
+ * Read WAV file tag when streaming
+ */
+static GstFlowReturn
+gst_wavparse_parse_stream_init (GstWavParse * wav)
+{
+  if (gst_adapter_available (wav->adapter) >= 12) {
+    GstBuffer *tmp = gst_buffer_new ();
+
+    /* _take flushes the data */
+    GST_BUFFER_DATA (tmp) = gst_adapter_take (wav->adapter, 12);
+    GST_BUFFER_SIZE (tmp) = 12;
+
+    GST_DEBUG ("Parsing wav header");
+    if (!gst_wavparse_parse_file_header (GST_ELEMENT (wav), tmp)) {
+      return GST_FLOW_ERROR;
+    }
+
+    wav->offset += 12;
+    /* Go to next state */
+    wav->state = GST_WAVPARSE_HEADER;
+  }
+  return GST_FLOW_OK;
+}
+
 /* handle an event sent directly to the element.
  *
  * This event can be sent either in the READY state or the
@@ -1100,6 +1267,8 @@ gst_wavparse_send_event (GstElement * element, GstEvent * event)
   gboolean res = FALSE;
   GstEvent **event_p;
 
+  GST_DEBUG_OBJECT (wav, "received event %s", GST_EVENT_TYPE_NAME (event));
+
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_SEEK:
       if (wav->state == GST_WAVPARSE_DATA) {
@@ -1149,6 +1318,7 @@ gst_wavparse_add_src_pad (GstWavParse * wav, GstBuffer * buf)
   gst_element_add_pad (GST_ELEMENT (wav), wav->srcpad);
   gst_element_no_more_pads (GST_ELEMENT (wav));
 
+  GST_DEBUG_OBJECT (wav, "Send newsegment event on newpad");
   gst_pad_push_event (wav->srcpad, wav->newsegment);
   wav->newsegment = NULL;
 
@@ -1161,7 +1331,7 @@ gst_wavparse_add_src_pad (GstWavParse * wav, GstBuffer * buf)
 #define MAX_BUFFER_SIZE 4096
 
 static GstFlowReturn
-gst_wavparse_stream_data (GstWavParse * wav, gboolean first)
+gst_wavparse_stream_data (GstWavParse * wav)
 {
   GstBuffer *buf = NULL;
   GstFlowReturn res = GST_FLOW_OK;
@@ -1169,8 +1339,10 @@ gst_wavparse_stream_data (GstWavParse * wav, gboolean first)
   GstClockTime timestamp, next_timestamp;
   guint64 pos, nextpos;
 
-  GST_LOG_OBJECT (wav, "offset: %" G_GINT64_FORMAT " , end: %" G_GINT64_FORMAT,
+iterate_adapter:
-      wav->offset, wav->end_offset);
+  GST_LOG_OBJECT (wav,
+      "offset: %" G_GINT64_FORMAT " , end: %" G_GINT64_FORMAT " , dataleft: %"
+      G_GINT64_FORMAT, wav->offset, wav->end_offset, wav->dataleft);
 
   /* Get the next n bytes and output them */
   if (wav->dataleft == 0 || wav->dataleft < wav->blockalign)
@@ -1187,18 +1359,32 @@ gst_wavparse_stream_data (GstWavParse * wav, gboolean first)
   GST_LOG_OBJECT (wav, "Fetching %" G_GINT64_FORMAT " bytes of data "
       "from the sinkpad", desired);
 
-  if ((res = gst_pad_pull_range (wav->sinkpad, wav->offset,
+  if (wav->streaming) {
-              desired, &buf)) != GST_FLOW_OK)
+    guint avail = gst_adapter_available (wav->adapter);
-    goto pull_error;
 
-  obtained = GST_BUFFER_SIZE (buf);
+    if (avail < desired) {
+      GST_LOG_OBJECT (wav, "Got only %d bytes of data from the sinkpad", avail);
+      return GST_FLOW_OK;
+    }
+
+    buf = gst_buffer_new ();
+    GST_BUFFER_DATA (buf) = gst_adapter_take (wav->adapter, desired);
+    GST_BUFFER_SIZE (buf) = desired;
+  } else {
+    if ((res = gst_pad_pull_range (wav->sinkpad, wav->offset,
+                desired, &buf)) != GST_FLOW_OK)
+      goto pull_error;
+  }
 
   /* first chunk of data? create the source pad. We do this only here so
    * we can detect broken .wav files with dts disguised as raw PCM (sigh) */
-  if (first) {
+  if (G_UNLIKELY (wav->first)) {
+    wav->first = FALSE;
     gst_wavparse_add_src_pad (wav, buf);
   }
 
+  obtained = GST_BUFFER_SIZE (buf);
+
   /* our positions */
   pos = wav->offset - wav->datastart;
   nextpos = pos + obtained;
@@ -1220,7 +1406,7 @@ gst_wavparse_stream_data (GstWavParse * wav, gboolean first)
   /* don't forget to set the caps on the buffer */
   gst_buffer_set_caps (buf, GST_PAD_CAPS (wav->srcpad));
 
-  GST_DEBUG_OBJECT (wav,
+  GST_LOG_OBJECT (wav,
       "Got buffer. timestamp:%" GST_TIME_FORMAT " , duration:%" GST_TIME_FORMAT
       ", size:%u", GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)),
       GST_TIME_ARGS (GST_BUFFER_DURATION (buf)), GST_BUFFER_SIZE (buf));
@@ -1230,10 +1416,18 @@ gst_wavparse_stream_data (GstWavParse * wav, gboolean first)
 
   if (obtained < wav->dataleft) {
     wav->dataleft -= obtained;
-    wav->offset += obtained;
   } else {
     wav->dataleft = 0;
   }
+  wav->offset += obtained;
+  /* Iterate until need more data, so adapter size won't grow */
+  if (wav->streaming) {
+    GST_LOG_OBJECT (wav,
+        "offset: %" G_GINT64_FORMAT " , end: %" G_GINT64_FORMAT, wav->offset,
+        wav->end_offset);
+    goto iterate_adapter;
+  }
+
   return res;
 
   /* ERROR */
@@ -1275,8 +1469,11 @@ gst_wavparse_loop (GstPad * pad)
   GstFlowReturn ret;
   GstWavParse *wav = GST_WAVPARSE (GST_PAD_PARENT (pad));
 
+  GST_LOG_OBJECT (wav, "process data");
+
   switch (wav->state) {
     case GST_WAVPARSE_START:
+      GST_DEBUG_OBJECT (wav, "GST_WAVPARSE_START");
       if ((ret = gst_wavparse_stream_init (wav)) != GST_FLOW_OK)
         goto pause;
 
@@ -1284,15 +1481,15 @@ gst_wavparse_loop (GstPad * pad)
       /* fall-through */
 
     case GST_WAVPARSE_HEADER:
+      GST_DEBUG_OBJECT (wav, "GST_WAVPARSE_HEADER");
       if ((ret = gst_wavparse_stream_headers (wav)) != GST_FLOW_OK)
         goto pause;
 
       wav->state = GST_WAVPARSE_DATA;
-      if ((ret = gst_wavparse_stream_data (wav, TRUE)) != GST_FLOW_OK)
+      /* fall-through */
-        goto pause;
+
-      break;
     case GST_WAVPARSE_DATA:
-      if ((ret = gst_wavparse_stream_data (wav, FALSE)) != GST_FLOW_OK)
+      if ((ret = gst_wavparse_stream_data (wav)) != GST_FLOW_OK)
         goto pause;
       break;
     default:
@@ -1315,6 +1512,58 @@ pause:
   }
 }
 
+static GstFlowReturn
+gst_wavparse_chain (GstPad * pad, GstBuffer * buf)
+{
+  GstFlowReturn ret;
+  GstWavParse *wav = GST_WAVPARSE (GST_PAD_PARENT (pad));
+
+  GST_LOG_OBJECT (wav, "adapter_push %" G_GINT64_FORMAT " bytes",
+      GST_BUFFER_SIZE (buf));
+
+  gst_adapter_push (wav->adapter, buf);
+
+  switch (wav->state) {
+    case GST_WAVPARSE_START:
+      GST_DEBUG_OBJECT (wav, "GST_WAVPARSE_START");
+      if ((ret = gst_wavparse_parse_stream_init (wav)) != GST_FLOW_OK)
+        goto pause;
+
+      wav->state = GST_WAVPARSE_HEADER;
+      /* fall-through */
+
+    case GST_WAVPARSE_HEADER:
+      GST_DEBUG_OBJECT (wav, "GST_WAVPARSE_HEADER");
+      if ((ret = gst_wavparse_stream_headers (wav)) != GST_FLOW_OK)
+        goto pause;
+
+      wav->state = GST_WAVPARSE_DATA;
+      /* fall-through */
+
+    case GST_WAVPARSE_DATA:
+      if ((ret = gst_wavparse_stream_data (wav)) != GST_FLOW_OK)
+        goto pause;
+      break;
+    default:
+      g_assert_not_reached ();
+  }
+
+  return ret;
+
+pause:
+  GST_LOG_OBJECT (wav, "pausing task %d", ret);
+  gst_pad_pause_task (wav->sinkpad);
+  if (GST_FLOW_IS_FATAL (ret)) {
+    /* for fatal errors we post an error message */
+    GST_ELEMENT_ERROR (wav, STREAM, FAILED,
+        (_("Internal data stream error.")),
+        ("streaming stopped, reason %s", gst_flow_get_name (ret)));
+    if (wav->srcpad != NULL)
+      gst_pad_push_event (wav->srcpad, gst_event_new_eos ());
+  }
+  return ret;
+}
+
 #if 0
 /* convert and query stuff */
 static const GstFormat *
@@ -1526,7 +1775,8 @@ gst_wavparse_srcpad_event (GstPad * pad, GstEvent * event)
   GstWavParse *wavparse = GST_WAVPARSE (GST_PAD_PARENT (pad));
   gboolean res = TRUE;
 
-  GST_DEBUG_OBJECT (wavparse, "event %d", GST_EVENT_TYPE (event));
+  GST_DEBUG_OBJECT (wavparse, "event %d, %s", GST_EVENT_TYPE (event),
+      GST_EVENT_TYPE_NAME (event));
 
   /* can only handle events when we are in the data state */
   if (wavparse->state != GST_WAVPARSE_DATA)
@@ -1551,20 +1801,35 @@ gst_wavparse_srcpad_event (GstPad * pad, GstEvent * event)
 static gboolean
 gst_wavparse_sink_activate (GstPad * sinkpad)
 {
-  if (gst_pad_check_pull_range (sinkpad))
+  GstWavParse *wav = GST_WAVPARSE (gst_pad_get_parent (sinkpad));
-    return gst_pad_activate_pull (sinkpad, TRUE);
+  gboolean res;
+
+  if (gst_pad_check_pull_range (sinkpad)) {
+    GST_DEBUG ("going to pull mode");
+    wav->streaming = FALSE;
+    wav->adapter = NULL;
+    res = gst_pad_activate_pull (sinkpad, TRUE);
+  } else {
+    GST_DEBUG ("going to push (streaming) mode");
+    wav->streaming = TRUE;
+    wav->adapter = gst_adapter_new ();
+    res = gst_pad_activate_push (sinkpad, TRUE);
+  }
+  gst_object_unref (wav);
+  return res;
+}
 
-  /* FIXME, we can only operate in pull mode for now */
-  GST_DEBUG_OBJECT (sinkpad, "pull_range not supported on sinkpad");
-  return FALSE;
-};
 
 static gboolean
 gst_wavparse_sink_activate_pull (GstPad * sinkpad, gboolean active)
 {
   GstWavParse *wav = GST_WAVPARSE (gst_pad_get_parent (sinkpad));
 
+  GST_DEBUG_OBJECT (wav, "activating pull");
+
   if (active) {
+    /* if we have a scheduler we can start the task */
+    wav->segment_running = TRUE;
     gst_pad_start_task (sinkpad, (GstTaskFunction) gst_wavparse_loop, sinkpad);
   } else {
     gst_pad_stop_task (sinkpad);
@@ -1580,6 +1845,10 @@ gst_wavparse_change_state (GstElement * element, GstStateChange transition)
   GstStateChangeReturn ret;
   GstWavParse *wav = GST_WAVPARSE (element);
 
+  GST_DEBUG_OBJECT (wav, "changing state %s - %s",
+      gst_element_state_get_name (GST_STATE_TRANSITION_CURRENT (transition)),
+      gst_element_state_get_name (GST_STATE_TRANSITION_NEXT (transition)));
+
   switch (transition) {
     case GST_STATE_CHANGE_NULL_TO_READY:
       break;
@@ -1603,8 +1872,11 @@ gst_wavparse_change_state (GstElement * element, GstStateChange transition)
       gst_wavparse_destroy_sourcepad (wav);
       gst_event_replace (event_p, NULL);
       gst_wavparse_reset (wav);
-    }
+      if (wav->adapter) {
+        gst_adapter_clear (wav->adapter);
+      }
       break;
+    }
     case GST_STATE_CHANGE_READY_TO_NULL:
       break;
     default:
@@ -1618,6 +1890,8 @@ plugin_init (GstPlugin * plugin)
 {
   gst_riff_init ();
 
+  GST_DEBUG_CATEGORY_INIT (wavparse_debug, "wavparse", 0, "WAV parser");
+
   return gst_element_register (plugin, "wavparse", GST_RANK_PRIMARY,
       GST_TYPE_WAVPARSE);
 }

gst/wavparse/gstwavparse.h

@@ -1,5 +1,6 @@
 /* GStreamer
  * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
+ * Copyright (C) <2006> Nokia Corporation, Stefan Kost <stefan.kost@nokia.com>.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Library General Public
@@ -25,6 +26,7 @@
 #include <gst/gst.h>
 #include "gst/riff/riff-ids.h"
 #include "gst/riff/riff-read.h"
+#include <gst/base/gstadapter.h>
 
 G_BEGIN_DECLS
 
@@ -93,9 +95,17 @@ struct _GstWavParse {
   /* pending seek */
   GstEvent *seek_event;
 
+  /* For streaming */
+  GstAdapter *adapter;
+  gboolean got_fmt;
+  gboolean streaming;
+
   /* configured segment, start/stop expressed in time */
   GstSegment segment;
   gboolean segment_running;
+  
+  /* for late pad configuration */
+  gboolean first;
 };
 
 struct _GstWavParseClass {