Add pulseaudio GStreamer element from gst-pulse. Development will continue here instead of pulseaudio SVN. Fixes bug ...

Original commit message from CVS:
* configure.ac:
* ext/pulse/Makefile.am:
* ext/pulse/plugin.c: (plugin_init):
* ext/pulse/pulsemixer.c: (gst_pulsemixer_interface_supported),
(gst_pulsemixer_implements_interface_init),
(gst_pulsemixer_init_interfaces), (gst_pulsemixer_base_init),
(gst_pulsemixer_class_init), (gst_pulsemixer_init),
(gst_pulsemixer_finalize), (gst_pulsemixer_set_property),
(gst_pulsemixer_get_property), (gst_pulsemixer_change_state):
* ext/pulse/pulsemixer.h:
* ext/pulse/pulsemixerctrl.c:
(gst_pulsemixer_ctrl_context_state_cb),
(gst_pulsemixer_ctrl_sink_info_cb),
(gst_pulsemixer_ctrl_source_info_cb),
(gst_pulsemixer_ctrl_subscribe_cb),
(gst_pulsemixer_ctrl_success_cb), (gst_pulsemixer_ctrl_open),
(gst_pulsemixer_ctrl_close), (gst_pulsemixer_ctrl_new),
(gst_pulsemixer_ctrl_free), (gst_pulsemixer_ctrl_list_tracks),
(gst_pulsemixer_ctrl_timeout_event), (restart_time_event),
(gst_pulsemixer_ctrl_set_volume), (gst_pulsemixer_ctrl_get_volume),
(gst_pulsemixer_ctrl_set_record), (gst_pulsemixer_ctrl_set_mute):
* ext/pulse/pulsemixerctrl.h:
* ext/pulse/pulsemixertrack.c: (gst_pulsemixer_track_class_init),
(gst_pulsemixer_track_init), (gst_pulsemixer_track_new):
* ext/pulse/pulsemixertrack.h:
* ext/pulse/pulseprobe.c: (gst_pulseprobe_context_state_cb),
(gst_pulseprobe_sink_info_cb), (gst_pulseprobe_source_info_cb),
(gst_pulseprobe_invalidate), (gst_pulseprobe_open),
(gst_pulseprobe_enumerate), (gst_pulseprobe_close),
(gst_pulseprobe_new), (gst_pulseprobe_free),
(gst_pulseprobe_get_properties), (gst_pulseprobe_needs_probe),
(gst_pulseprobe_probe_property), (gst_pulseprobe_get_values),
(gst_pulseprobe_set_server):
* ext/pulse/pulseprobe.h:
* ext/pulse/pulsesink.c: (gst_pulsesink_base_init),
(gst_pulsesink_class_init), (gst_pulsesink_init),
(gst_pulsesink_destroy_stream), (gst_pulsesink_destroy_context),
(gst_pulsesink_finalize), (gst_pulsesink_dispose),
(gst_pulsesink_set_property), (gst_pulsesink_get_property),
(gst_pulsesink_context_state_cb), (gst_pulsesink_stream_state_cb),
(gst_pulsesink_stream_request_cb),
(gst_pulsesink_stream_latency_update_cb), (gst_pulsesink_open),
(gst_pulsesink_close), (gst_pulsesink_prepare),
(gst_pulsesink_unprepare), (gst_pulsesink_write),
(gst_pulsesink_delay), (gst_pulsesink_success_cb),
(gst_pulsesink_reset), (gst_pulsesink_change_title),
(gst_pulsesink_event), (gst_pulsesink_get_type):
* ext/pulse/pulsesink.h:
* ext/pulse/pulsesrc.c: (gst_pulsesrc_interface_supported),
(gst_pulsesrc_implements_interface_init),
(gst_pulsesrc_init_interfaces), (gst_pulsesrc_base_init),
(gst_pulsesrc_class_init), (gst_pulsesrc_init),
(gst_pulsesrc_destroy_stream), (gst_pulsesrc_destroy_context),
(gst_pulsesrc_finalize), (gst_pulsesrc_dispose),
(gst_pulsesrc_set_property), (gst_pulsesrc_get_property),
(gst_pulsesrc_context_state_cb), (gst_pulsesrc_stream_state_cb),
(gst_pulsesrc_stream_request_cb), (gst_pulsesrc_open),
(gst_pulsesrc_close), (gst_pulsesrc_prepare),
(gst_pulsesrc_unprepare), (gst_pulsesrc_read),
(gst_pulsesrc_delay), (gst_pulsesrc_change_state),
(gst_pulsesrc_get_type):
* ext/pulse/pulsesrc.h:
* ext/pulse/pulseutil.c: (gst_pulse_fill_sample_spec),
(gst_pulse_client_name), (gst_pulse_gst_to_channel_map):
* ext/pulse/pulseutil.h:
Add pulseaudio GStreamer element from gst-pulse. Development will
continue here instead of pulseaudio SVN. Fixes bug #400679.
Only changes over gst-pulse SVN are added copyright to the top of
files and coding style changes.

1 files changed, 703 insertions, 0 deletions

diff --git a/ext/pulse/pulsesrc.c b/ext/pulse/pulsesrc.c
new file mode 100644
index 00000000..e69c5edd
--- /dev/null
+++ b/ext/pulse/pulsesrc.c
@@ -0,0 +1,703 @@
+/*
+ *  GStreamer pulseaudio plugin
+ *
+ *  Copyright (c) 2004-2008 Lennart Poettering
+ *
+ *  gst-pulse is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU Lesser General Public License as
+ *  published by the Free Software Foundation; either version 2.1 of the
+ *  License, or (at your option) any later version.
+ *
+ *  gst-pulse 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
+ *  Lesser General Public License for more details.
+ *
+ *  You should have received a copy of the GNU Lesser General Public
+ *  License along with gst-pulse; if not, write to the Free Software
+ *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+ *  USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <string.h>
+#include <stdio.h>
+
+#include <gst/base/gstbasesrc.h>
+#include <gst/gsttaglist.h>
+
+#include "pulsesrc.h"
+#include "pulseutil.h"
+#include "pulsemixerctrl.h"
+
+GST_DEBUG_CATEGORY_EXTERN (pulse_debug);
+#define GST_CAT_DEFAULT pulse_debug
+
+enum
+{
+  PROP_SERVER = 1,
+  PROP_DEVICE
+};
+
+static GstAudioSrcClass *parent_class = NULL;
+
+GST_IMPLEMENT_PULSEMIXER_CTRL_METHODS (GstPulseSrc, gst_pulsesrc)
+
+     static void gst_pulsesrc_destroy_stream (GstPulseSrc * pulsesrc);
+
+     static void gst_pulsesrc_destroy_context (GstPulseSrc * pulsesrc);
+
+     static void gst_pulsesrc_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec);
+     static void gst_pulsesrc_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec);
+     static void gst_pulsesrc_finalize (GObject * object);
+
+     static void gst_pulsesrc_dispose (GObject * object);
+
+     static gboolean gst_pulsesrc_open (GstAudioSrc * asrc);
+
+     static gboolean gst_pulsesrc_close (GstAudioSrc * asrc);
+
+     static gboolean gst_pulsesrc_prepare (GstAudioSrc * asrc,
+    GstRingBufferSpec * spec);
+     static gboolean gst_pulsesrc_unprepare (GstAudioSrc * asrc);
+
+     static guint gst_pulsesrc_read (GstAudioSrc * asrc, gpointer data,
+    guint length);
+     static guint gst_pulsesrc_delay (GstAudioSrc * asrc);
+
+     static GstStateChangeReturn gst_pulsesrc_change_state (GstElement *
+    element, GstStateChange transition);
+
+#if (G_BYTE_ORDER == G_LITTLE_ENDIAN)
+# define ENDIANNESS   "LITTLE_ENDIAN, BIG_ENDIAN"
+#else
+# define ENDIANNESS   "BIG_ENDIAN, LITTLE_ENDIAN"
+#endif
+
+     static gboolean gst_pulsesrc_interface_supported (GstImplementsInterface *
+    iface, GType interface_type)
+{
+  GstPulseSrc *this = GST_PULSESRC (iface);
+
+  if (interface_type == GST_TYPE_MIXER && this->mixer)
+    return TRUE;
+
+  return FALSE;
+}
+
+static void
+gst_pulsesrc_implements_interface_init (GstImplementsInterfaceClass * klass)
+{
+  klass->supported = gst_pulsesrc_interface_supported;
+}
+
+static void
+gst_pulsesrc_init_interfaces (GType type)
+{
+  static const GInterfaceInfo implements_iface_info = {
+    (GInterfaceInitFunc) gst_pulsesrc_implements_interface_init,
+    NULL,
+    NULL,
+  };
+  static const GInterfaceInfo mixer_iface_info = {
+    (GInterfaceInitFunc) gst_pulsesrc_mixer_interface_init,
+    NULL,
+    NULL,
+  };
+
+  g_type_add_interface_static (type, GST_TYPE_IMPLEMENTS_INTERFACE,
+      &implements_iface_info);
+  g_type_add_interface_static (type, GST_TYPE_MIXER, &mixer_iface_info);
+}
+
+static void
+gst_pulsesrc_base_init (gpointer g_class)
+{
+
+  static GstStaticPadTemplate pad_template = GST_STATIC_PAD_TEMPLATE ("src",
+      GST_PAD_SRC,
+      GST_PAD_ALWAYS,
+      GST_STATIC_CAPS ("audio/x-raw-int, "
+          "endianness = (int) { " ENDIANNESS " }, "
+          "signed = (boolean) TRUE, "
+          "width = (int) 16, "
+          "depth = (int) 16, "
+          "rate = (int) [ 1, MAX ], "
+          "channels = (int) [ 1, 16 ];"
+          "audio/x-raw-int, "
+          "endianness = (int) { " ENDIANNESS " }, "
+          "signed = (boolean) TRUE, "
+          "width = (int) 32, "
+          "depth = (int) 32, "
+          "rate = (int) [ 1, MAX ], "
+          "channels = (int) [ 1, 16 ];"
+          "audio/x-raw-float, "
+          "endianness = (int) { " ENDIANNESS " }, "
+          "width = (int) 32, "
+          "rate = (int) [ 1, MAX ], "
+          "channels = (int) [ 1, 16 ];"
+          "audio/x-raw-int, "
+          "signed = (boolean) FALSE, "
+          "width = (int) 8, "
+          "depth = (int) 8, "
+          "rate = (int) [ 1, MAX ], "
+          "channels = (int) [ 1, 16 ];"
+          "audio/x-alaw, "
+          "rate = (int) [ 1, MAX], "
+          "channels = (int) [ 1, 16 ];"
+          "audio/x-mulaw, "
+          "rate = (int) [ 1, MAX], " "channels = (int) [ 1, 16 ]")
+      );
+
+  static const GstElementDetails details =
+      GST_ELEMENT_DETAILS ("PulseAudio Audio Source",
+      "Source/Audio",
+      "Captures audio from a PulseAudio server",
+      "Lennart Poettering");
+
+  GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
+
+  gst_element_class_set_details (element_class, &details);
+  gst_element_class_add_pad_template (element_class,
+      gst_static_pad_template_get (&pad_template));
+}
+
+static void
+gst_pulsesrc_class_init (gpointer g_class, gpointer class_data)
+{
+
+  GObjectClass *gobject_class = G_OBJECT_CLASS (g_class);
+
+  GstAudioSrcClass *gstaudiosrc_class = GST_AUDIO_SRC_CLASS (g_class);
+
+  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
+
+  parent_class = g_type_class_peek_parent (g_class);
+
+  gstelement_class->change_state =
+      GST_DEBUG_FUNCPTR (gst_pulsesrc_change_state);
+
+  gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_pulsesrc_dispose);
+  gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_pulsesrc_finalize);
+  gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_pulsesrc_set_property);
+  gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_pulsesrc_get_property);
+
+  gstaudiosrc_class->open = GST_DEBUG_FUNCPTR (gst_pulsesrc_open);
+  gstaudiosrc_class->close = GST_DEBUG_FUNCPTR (gst_pulsesrc_close);
+  gstaudiosrc_class->prepare = GST_DEBUG_FUNCPTR (gst_pulsesrc_prepare);
+  gstaudiosrc_class->unprepare = GST_DEBUG_FUNCPTR (gst_pulsesrc_unprepare);
+  gstaudiosrc_class->read = GST_DEBUG_FUNCPTR (gst_pulsesrc_read);
+  gstaudiosrc_class->delay = GST_DEBUG_FUNCPTR (gst_pulsesrc_delay);
+
+  /* Overwrite GObject fields */
+  g_object_class_install_property (gobject_class,
+      PROP_SERVER,
+      g_param_spec_string ("server", "Server",
+          "The PulseAudio server to connect to", NULL, G_PARAM_READWRITE));
+  g_object_class_install_property (gobject_class, PROP_DEVICE,
+      g_param_spec_string ("device", "Source",
+          "The PulseAudio source device to connect to", NULL,
+          G_PARAM_READWRITE));
+}
+
+static void
+gst_pulsesrc_init (GTypeInstance * instance, gpointer g_class)
+{
+
+  GstPulseSrc *pulsesrc = GST_PULSESRC (instance);
+
+  int e;
+
+  pulsesrc->server = pulsesrc->device = NULL;
+
+  pulsesrc->context = NULL;
+  pulsesrc->stream = NULL;
+
+  pulsesrc->read_buffer = NULL;
+  pulsesrc->read_buffer_length = 0;
+
+  pulsesrc->mainloop = pa_threaded_mainloop_new ();
+  g_assert (pulsesrc->mainloop);
+
+  e = pa_threaded_mainloop_start (pulsesrc->mainloop);
+  g_assert (e == 0);
+
+  pulsesrc->mixer = NULL;
+}
+
+static void
+gst_pulsesrc_destroy_stream (GstPulseSrc * pulsesrc)
+{
+  if (pulsesrc->stream) {
+    pa_stream_disconnect (pulsesrc->stream);
+    pa_stream_unref (pulsesrc->stream);
+    pulsesrc->stream = NULL;
+  }
+}
+
+static void
+gst_pulsesrc_destroy_context (GstPulseSrc * pulsesrc)
+{
+
+  gst_pulsesrc_destroy_stream (pulsesrc);
+
+  if (pulsesrc->context) {
+    pa_context_disconnect (pulsesrc->context);
+    pa_context_unref (pulsesrc->context);
+    pulsesrc->context = NULL;
+  }
+}
+
+static void
+gst_pulsesrc_finalize (GObject * object)
+{
+  GstPulseSrc *pulsesrc = GST_PULSESRC (object);
+
+  pa_threaded_mainloop_stop (pulsesrc->mainloop);
+
+  gst_pulsesrc_destroy_context (pulsesrc);
+
+  g_free (pulsesrc->server);
+  g_free (pulsesrc->device);
+
+  pa_threaded_mainloop_free (pulsesrc->mainloop);
+
+  if (pulsesrc->mixer)
+    gst_pulsemixer_ctrl_free (pulsesrc->mixer);
+
+  G_OBJECT_CLASS (parent_class)->finalize (object);
+}
+
+static void
+gst_pulsesrc_dispose (GObject * object)
+{
+  G_OBJECT_CLASS (parent_class)->dispose (object);
+}
+
+static void
+gst_pulsesrc_set_property (GObject * object,
+    guint prop_id, const GValue * value, GParamSpec * pspec)
+{
+
+  GstPulseSrc *pulsesrc = GST_PULSESRC (object);
+
+  switch (prop_id) {
+    case PROP_SERVER:
+      g_free (pulsesrc->server);
+      pulsesrc->server = g_value_dup_string (value);
+      break;
+
+    case PROP_DEVICE:
+      g_free (pulsesrc->device);
+      pulsesrc->device = g_value_dup_string (value);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pulsesrc_get_property (GObject * object,
+    guint prop_id, GValue * value, GParamSpec * pspec)
+{
+
+  GstPulseSrc *pulsesrc = GST_PULSESRC (object);
+
+  switch (prop_id) {
+    case PROP_SERVER:
+      g_value_set_string (value, pulsesrc->server);
+      break;
+
+    case PROP_DEVICE:
+      g_value_set_string (value, pulsesrc->device);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pulsesrc_context_state_cb (pa_context * c, void *userdata)
+{
+  GstPulseSrc *pulsesrc = GST_PULSESRC (userdata);
+
+  switch (pa_context_get_state (c)) {
+    case PA_CONTEXT_READY:
+    case PA_CONTEXT_TERMINATED:
+    case PA_CONTEXT_FAILED:
+      pa_threaded_mainloop_signal (pulsesrc->mainloop, 0);
+      break;
+
+    case PA_CONTEXT_UNCONNECTED:
+    case PA_CONTEXT_CONNECTING:
+    case PA_CONTEXT_AUTHORIZING:
+    case PA_CONTEXT_SETTING_NAME:
+      break;
+  }
+}
+
+static void
+gst_pulsesrc_stream_state_cb (pa_stream * s, void *userdata)
+{
+  GstPulseSrc *pulsesrc = GST_PULSESRC (userdata);
+
+  switch (pa_stream_get_state (s)) {
+
+    case PA_STREAM_READY:
+    case PA_STREAM_FAILED:
+    case PA_STREAM_TERMINATED:
+      pa_threaded_mainloop_signal (pulsesrc->mainloop, 0);
+      break;
+
+    case PA_STREAM_UNCONNECTED:
+    case PA_STREAM_CREATING:
+      break;
+  }
+}
+
+static void
+gst_pulsesrc_stream_request_cb (pa_stream * s, size_t length, void *userdata)
+{
+  GstPulseSrc *pulsesrc = GST_PULSESRC (userdata);
+
+  pa_threaded_mainloop_signal (pulsesrc->mainloop, 0);
+}
+
+static gboolean
+gst_pulsesrc_open (GstAudioSrc * asrc)
+{
+  GstPulseSrc *pulsesrc = GST_PULSESRC (asrc);
+
+  gchar *name = gst_pulse_client_name ();
+
+  pa_threaded_mainloop_lock (pulsesrc->mainloop);
+
+  if (!(pulsesrc->context =
+          pa_context_new (pa_threaded_mainloop_get_api (pulsesrc->mainloop),
+              name))) {
+    GST_ELEMENT_ERROR (pulsesrc, RESOURCE, FAILED, ("Failed to create context"),
+        (NULL));
+    goto unlock_and_fail;
+  }
+
+  pa_context_set_state_callback (pulsesrc->context,
+      gst_pulsesrc_context_state_cb, pulsesrc);
+
+  if (pa_context_connect (pulsesrc->context, pulsesrc->server, 0, NULL) < 0) {
+    GST_ELEMENT_ERROR (pulsesrc, RESOURCE, FAILED, ("Failed to connect: %s",
+            pa_strerror (pa_context_errno (pulsesrc->context))), (NULL));
+    goto unlock_and_fail;
+  }
+
+  /* Wait until the context is ready */
+  pa_threaded_mainloop_wait (pulsesrc->mainloop);
+
+  if (pa_context_get_state (pulsesrc->context) != PA_CONTEXT_READY) {
+    GST_ELEMENT_ERROR (pulsesrc, RESOURCE, FAILED, ("Failed to connect: %s",
+            pa_strerror (pa_context_errno (pulsesrc->context))), (NULL));
+    goto unlock_and_fail;
+  }
+
+  pa_threaded_mainloop_unlock (pulsesrc->mainloop);
+
+  g_free (name);
+  return TRUE;
+
+unlock_and_fail:
+
+  pa_threaded_mainloop_unlock (pulsesrc->mainloop);
+
+  g_free (name);
+  return FALSE;
+}
+
+static gboolean
+gst_pulsesrc_close (GstAudioSrc * asrc)
+{
+  GstPulseSrc *pulsesrc = GST_PULSESRC (asrc);
+
+  pa_threaded_mainloop_lock (pulsesrc->mainloop);
+  gst_pulsesrc_destroy_context (pulsesrc);
+  pa_threaded_mainloop_unlock (pulsesrc->mainloop);
+
+  return TRUE;
+}
+
+static gboolean
+gst_pulsesrc_prepare (GstAudioSrc * asrc, GstRingBufferSpec * spec)
+{
+  pa_buffer_attr buf_attr;
+
+  pa_channel_map channel_map;
+
+  GstPulseSrc *pulsesrc = GST_PULSESRC (asrc);
+
+  if (!gst_pulse_fill_sample_spec (spec, &pulsesrc->sample_spec)) {
+    GST_ELEMENT_ERROR (pulsesrc, RESOURCE, SETTINGS,
+        ("Invalid sample specification."), (NULL));
+    goto unlock_and_fail;
+  }
+
+  pa_threaded_mainloop_lock (pulsesrc->mainloop);
+
+  if (!pulsesrc->context
+      || pa_context_get_state (pulsesrc->context) != PA_CONTEXT_READY) {
+    GST_ELEMENT_ERROR (pulsesrc, RESOURCE, FAILED, ("Bad context state: %s",
+            pulsesrc->context ? pa_strerror (pa_context_errno (pulsesrc->
+                    context)) : NULL), (NULL));
+    goto unlock_and_fail;
+  }
+
+  if (!(pulsesrc->stream = pa_stream_new (pulsesrc->context,
+              "Record Stream",
+              &pulsesrc->sample_spec,
+              gst_pulse_gst_to_channel_map (&channel_map, spec)))) {
+    GST_ELEMENT_ERROR (pulsesrc, RESOURCE, FAILED,
+        ("Failed to create stream: %s",
+            pa_strerror (pa_context_errno (pulsesrc->context))), (NULL));
+    goto unlock_and_fail;
+  }
+
+  pa_stream_set_state_callback (pulsesrc->stream, gst_pulsesrc_stream_state_cb,
+      pulsesrc);
+  pa_stream_set_read_callback (pulsesrc->stream, gst_pulsesrc_stream_request_cb,
+      pulsesrc);
+
+  memset (&buf_attr, 0, sizeof (buf_attr));
+  buf_attr.maxlength = spec->segtotal * spec->segsize * 2;
+  buf_attr.fragsize = spec->segsize;
+
+  if (pa_stream_connect_record (pulsesrc->stream, pulsesrc->device, &buf_attr,
+          PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_AUTO_TIMING_UPDATE |
+          PA_STREAM_NOT_MONOTONOUS) < 0) {
+    GST_ELEMENT_ERROR (pulsesrc, RESOURCE, FAILED,
+        ("Failed to connect stream: %s",
+            pa_strerror (pa_context_errno (pulsesrc->context))), (NULL));
+    goto unlock_and_fail;
+  }
+
+  /* Wait until the stream is ready */
+  pa_threaded_mainloop_wait (pulsesrc->mainloop);
+
+  if (pa_stream_get_state (pulsesrc->stream) != PA_STREAM_READY) {
+    GST_ELEMENT_ERROR (pulsesrc, RESOURCE, FAILED,
+        ("Failed to connect stream: %s",
+            pa_strerror (pa_context_errno (pulsesrc->context))), (NULL));
+    goto unlock_and_fail;
+  }
+
+  pa_threaded_mainloop_unlock (pulsesrc->mainloop);
+
+  spec->bytes_per_sample = pa_frame_size (&pulsesrc->sample_spec);
+  memset (spec->silence_sample, 0, spec->bytes_per_sample);
+
+  return TRUE;
+
+unlock_and_fail:
+
+  pa_threaded_mainloop_unlock (pulsesrc->mainloop);
+  return FALSE;
+}
+
+static gboolean
+gst_pulsesrc_unprepare (GstAudioSrc * asrc)
+{
+  GstPulseSrc *pulsesrc = GST_PULSESRC (asrc);
+
+  pa_threaded_mainloop_lock (pulsesrc->mainloop);
+  gst_pulsesrc_destroy_stream (pulsesrc);
+
+  pa_threaded_mainloop_unlock (pulsesrc->mainloop);
+
+  pulsesrc->read_buffer = NULL;
+  pulsesrc->read_buffer_length = 0;
+
+  return TRUE;
+}
+
+#define CHECK_DEAD_GOTO(pulsesrc, label) \
+if (!(pulsesrc)->context || pa_context_get_state((pulsesrc)->context) != PA_CONTEXT_READY || \
+    !(pulsesrc)->stream || pa_stream_get_state((pulsesrc)->stream) != PA_STREAM_READY) { \
+    GST_ELEMENT_ERROR((pulsesrc), RESOURCE, FAILED, ("Disconnected: %s", (pulsesrc)->context ? pa_strerror(pa_context_errno((pulsesrc)->context)) : NULL), (NULL)); \
+    goto label; \
+}
+
+static guint
+gst_pulsesrc_read (GstAudioSrc * asrc, gpointer data, guint length)
+{
+  GstPulseSrc *pulsesrc = GST_PULSESRC (asrc);
+
+  size_t sum = 0;
+
+  pa_threaded_mainloop_lock (pulsesrc->mainloop);
+
+  CHECK_DEAD_GOTO (pulsesrc, unlock_and_fail);
+
+  while (length > 0) {
+    size_t l;
+
+    if (!pulsesrc->read_buffer) {
+
+      for (;;) {
+        if (pa_stream_peek (pulsesrc->stream, &pulsesrc->read_buffer,
+                &pulsesrc->read_buffer_length) < 0) {
+          GST_ELEMENT_ERROR (pulsesrc, RESOURCE, FAILED,
+              ("pa_stream_peek() failed: %s",
+                  pa_strerror (pa_context_errno (pulsesrc->context))), (NULL));
+          goto unlock_and_fail;
+        }
+
+        if (pulsesrc->read_buffer)
+          break;
+
+        pa_threaded_mainloop_wait (pulsesrc->mainloop);
+
+        CHECK_DEAD_GOTO (pulsesrc, unlock_and_fail);
+      }
+    }
+
+    g_assert (pulsesrc->read_buffer && pulsesrc->read_buffer_length);
+
+    l = pulsesrc->read_buffer_length >
+        length ? length : pulsesrc->read_buffer_length;
+
+    memcpy (data, pulsesrc->read_buffer, l);
+
+    pulsesrc->read_buffer = (const guint8 *) pulsesrc->read_buffer + l;
+    pulsesrc->read_buffer_length -= l;
+
+    data = (guint8 *) data + l;
+    length -= l;
+
+    sum += l;
+
+    if (pulsesrc->read_buffer_length <= 0) {
+
+      if (pa_stream_drop (pulsesrc->stream) < 0) {
+        GST_ELEMENT_ERROR (pulsesrc, RESOURCE, FAILED,
+            ("pa_stream_drop() failed: %s",
+                pa_strerror (pa_context_errno (pulsesrc->context))), (NULL));
+        goto unlock_and_fail;
+      }
+
+      pulsesrc->read_buffer = NULL;
+      pulsesrc->read_buffer_length = 0;
+    }
+  }
+
+  pa_threaded_mainloop_unlock (pulsesrc->mainloop);
+
+  return sum;
+
+unlock_and_fail:
+  pa_threaded_mainloop_unlock (pulsesrc->mainloop);
+  return 0;
+}
+
+static guint
+gst_pulsesrc_delay (GstAudioSrc * asrc)
+{
+  GstPulseSrc *pulsesrc = GST_PULSESRC (asrc);
+
+  pa_usec_t t;
+
+  int negative;
+
+  pa_threaded_mainloop_lock (pulsesrc->mainloop);
+
+  CHECK_DEAD_GOTO (pulsesrc, unlock_and_fail);
+
+  if (pa_stream_get_latency (pulsesrc->stream, &t, &negative) < 0) {
+
+    if (pa_context_errno (pulsesrc->context) != PA_ERR_NODATA) {
+      GST_ELEMENT_ERROR (pulsesrc, RESOURCE, FAILED,
+          ("pa_stream_get_latency() failed: %s",
+              pa_strerror (pa_context_errno (pulsesrc->context))), (NULL));
+      goto unlock_and_fail;
+    }
+
+    GST_WARNING ("Not data while querying latency");
+    t = 0;
+  } else if (negative)
+    t = 0;
+
+  pa_threaded_mainloop_unlock (pulsesrc->mainloop);
+
+  return (guint) ((t * pulsesrc->sample_spec.rate) / 1000000LL);
+
+unlock_and_fail:
+
+  pa_threaded_mainloop_unlock (pulsesrc->mainloop);
+  return 0;
+}
+
+static GstStateChangeReturn
+gst_pulsesrc_change_state (GstElement * element, GstStateChange transition)
+{
+  GstPulseSrc *this = GST_PULSESRC (element);
+
+  switch (transition) {
+    case GST_STATE_CHANGE_NULL_TO_READY:
+
+      if (!this->mixer)
+        this->mixer =
+            gst_pulsemixer_ctrl_new (this->server, this->device,
+            GST_PULSEMIXER_SOURCE);
+
+      break;
+
+    case GST_STATE_CHANGE_READY_TO_NULL:
+
+      if (this->mixer) {
+        gst_pulsemixer_ctrl_free (this->mixer);
+        this->mixer = NULL;
+      }
+
+      break;
+
+    default:
+      ;
+  }
+
+  if (GST_ELEMENT_CLASS (parent_class)->change_state)
+    return GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
+
+  return GST_STATE_CHANGE_SUCCESS;
+}
+
+GType
+gst_pulsesrc_get_type (void)
+{
+  static GType pulsesrc_type = 0;
+
+  if (!pulsesrc_type) {
+
+    static const GTypeInfo pulsesrc_info = {
+      sizeof (GstPulseSrcClass),
+      gst_pulsesrc_base_init,
+      NULL,
+      gst_pulsesrc_class_init,
+      NULL,
+      NULL,
+      sizeof (GstPulseSrc),
+      0,
+      gst_pulsesrc_init,
+    };
+
+    pulsesrc_type = g_type_register_static (GST_TYPE_AUDIO_SRC,
+        "GstPulseSrc", &pulsesrc_info, 0);
+
+    gst_pulsesrc_init_interfaces (pulsesrc_type);
+  }
+
+  return pulsesrc_type;
+}