pulse: code rewrite using the asynchronous libpulse API

This is a complete rewrite of the PulseAudio output plugin.  It uses
the asynchronous API, which gives us more control over everything.
Additionally, it connects to the PulseAudio server on startup, and
keeps this connection up while MPD runs.  During pause, instead of
closing the stream, it enables "cork".

3 files changed, 765 insertions, 282 deletions

diff --git a/src/mixer/pulse_mixer_plugin.c b/src/mixer/pulse_mixer_plugin.c
index ecc0fc75b..b33ef80ae 100644
--- a/src/mixer/pulse_mixer_plugin.c
+++ b/src/mixer/pulse_mixer_plugin.c
@@ -18,12 +18,20 @@
  */
 
 #include "mixer_api.h"
+#include "output/pulse_output_plugin.h"
 #include "conf.h"
+#include "event_pipe.h"
 
 #include <glib.h>
-#include <pulse/volume.h>
-#include <pulse/pulseaudio.h>
 
+#include <pulse/thread-mainloop.h>
+#include <pulse/context.h>
+#include <pulse/introspect.h>
+#include <pulse/stream.h>
+#include <pulse/subscribe.h>
+#include <pulse/error.h>
+
+#include <assert.h>
 #include <string.h>
 
 #undef G_LOG_DOMAIN
@@ -32,15 +40,9 @@
 struct pulse_mixer {
 	struct mixer base;
 
-	const char *server;
-	const char *sink;
-	const char *output_name;
+	struct pulse_output *output;
 
-	uint32_t index;
 	bool online;
-
-	struct pa_context *context;
-	struct pa_threaded_mainloop *mainloop;
 	struct pa_cvolume volume;
 
 };
@@ -54,175 +56,159 @@ pulse_mixer_quark(void)
 	return g_quark_from_static_string("pulse_mixer");
 }
 
-/**
- * \brief waits for a pulseaudio operation to finish, frees it and
- *     unlocks the mainloop
- * \param operation the operation to wait for
- * \return true if operation has finished normally (DONE state),
- *     false otherwise
- */
-static bool
-pulse_wait_for_operation(struct pa_threaded_mainloop *mainloop,
-			 struct pa_operation *operation)
+static void
+pulse_mixer_offline(struct pulse_mixer *pm)
 {
-	pa_operation_state_t state;
-
-	assert(mainloop != NULL);
-	assert(operation != NULL);
-
-	state = pa_operation_get_state(operation);
-	while (state == PA_OPERATION_RUNNING) {
-		pa_threaded_mainloop_wait(mainloop);
-		state = pa_operation_get_state(operation);
-	}
+	if (!pm->online)
+		return;
 
-	pa_operation_unref(operation);
+	pm->online = false;
 
-	return state == PA_OPERATION_DONE;
+	event_pipe_emit(PIPE_EVENT_MIXER);
 }
 
+/**
+ * Callback invoked by pulse_mixer_update().  Receives the new mixer
+ * value.
+ */
 static void
-sink_input_cb(G_GNUC_UNUSED pa_context *context, const pa_sink_input_info *i,
-	      int eol, void *userdata)
+pulse_mixer_volume_cb(G_GNUC_UNUSED pa_context *context, const pa_sink_input_info *i,
+		      int eol, void *userdata)
 {
-
 	struct pulse_mixer *pm = userdata;
 
-	if (eol) {
-		g_debug("eol error sink_input_cb");
+	if (eol)
 		return;
-	}
 
 	if (i == NULL) {
-		g_debug("Sink input callback failure");
+		pulse_mixer_offline(pm);
 		return;
 	}
 
-	g_debug("sink input cb %s, index %d ",i->name,i->index);
+	pm->online = true;
+	pm->volume = i->volume;
 
-	if (strcmp(i->name,pm->output_name) == 0) {
-		pm->index = i->index;
-		pm->online = true;
-		pm->volume = i->volume;
-	} else
-		g_debug("bad name");
+	event_pipe_emit(PIPE_EVENT_MIXER);
 }
 
 static void
-sink_input_vol(G_GNUC_UNUSED pa_context *context, const pa_sink_input_info *i,
-	       int eol, void *userdata)
+pulse_mixer_update(struct pulse_mixer *pm)
 {
+	pa_operation *o;
 
-	struct pulse_mixer *pm = userdata;
+	assert(pm->output->stream != NULL);
 
-	if (eol) {
-		g_debug("eol error sink_input_vol");
+	if (pm->output->context == NULL)
 		return;
-	}
 
-	if (i == NULL) {
-		g_debug("Sink input callback failure");
+	o = pa_context_get_sink_input_info(pm->output->context,
+					   pa_stream_get_index(pm->output->stream),
+					   pulse_mixer_volume_cb, pm);
+	if (o == NULL) {
+		g_warning("pa_context_get_sink_input_info() failed: %s",
+			  pa_strerror(pa_context_errno(pm->output->context)));
+		pulse_mixer_offline(pm);
 		return;
 	}
 
-	g_debug("sink input vol %s, index %d ", i->name, i->index);
+	pa_operation_unref(o);
+}
 
-	pm->volume = i->volume;
+static void
+pulse_mixer_handle_sink_input(struct pulse_mixer *pm,
+			      pa_subscription_event_type_t t,
+			      uint32_t idx)
+{
+	if (pm->output->stream == NULL) {
+		pulse_mixer_offline(pm);
+		return;
+	}
+
+	if (idx != pa_stream_get_index(pm->output->stream))
+		return;
 
-	pa_threaded_mainloop_signal(pm->mainloop, 0);
+	if (t == PA_SUBSCRIPTION_EVENT_NEW ||
+	    t == PA_SUBSCRIPTION_EVENT_CHANGE)
+		pulse_mixer_update(pm);
 }
 
 static void
-subscribe_cb(pa_context *c, pa_subscription_event_type_t t,
+pulse_mixer_subscribe_cb(G_GNUC_UNUSED pa_context *c, pa_subscription_event_type_t t,
 	     uint32_t idx, void *userdata)
 {
-
 	struct pulse_mixer *pm = userdata;
 
-	g_debug("subscribe call back");
-
 	switch (t & PA_SUBSCRIPTION_EVENT_FACILITY_MASK) {
 	case PA_SUBSCRIPTION_EVENT_SINK_INPUT:
-		if ((t & PA_SUBSCRIPTION_EVENT_TYPE_MASK) ==
-		    PA_SUBSCRIPTION_EVENT_REMOVE &&
-		    pm->index == idx)
-			pm->online = false;
-		else {
-			pa_operation *o;
-
-			o = pa_context_get_sink_input_info(c, idx,
-							   sink_input_cb, pm);
-			if (o == NULL) {
-				g_debug("pa_context_get_sink_input_info() failed");
-				return;
-			}
-
-			pa_operation_unref(o);
-		}
-
+		pulse_mixer_handle_sink_input(pm,
+					      t & PA_SUBSCRIPTION_EVENT_TYPE_MASK,
+					      idx);
 		break;
 	}
 }
 
 static void
-context_state_cb(pa_context *context, void *userdata)
+pulxe_mixer_context_state_cb(pa_context *context, void *userdata)
 {
 	struct pulse_mixer *pm = userdata;
+	pa_operation *o;
 
-	switch (pa_context_get_state(context)) {
-	case PA_CONTEXT_READY: {
-		pa_operation *o;
+	/* pass event to the output's callback function */
+	pulse_output_context_state_cb(context, pm->output);
 
-		pa_context_set_subscribe_callback(context, subscribe_cb, pm);
+	if (pa_context_get_state(context) == PA_CONTEXT_READY) {
+		/* subscribe to sink_input events after the connection
+		   has been established */
 
 		o = pa_context_subscribe(context,
 					 (pa_subscription_mask_t)PA_SUBSCRIPTION_MASK_SINK_INPUT,
 					 NULL, NULL);
 		if (o == NULL) {
-			g_debug("pa_context_subscribe() failed");
+			g_warning("pa_context_subscribe() failed: %s",
+				  pa_strerror(pa_context_errno(context)));
 			return;
 		}
 
 		pa_operation_unref(o);
 
-		o = pa_context_get_sink_input_info_list(context,
-							sink_input_cb, pm);
-		if (o == NULL) {
-			g_debug("pa_context_get_sink_input_info_list() failed");
-			return;
-		}
-
-		pa_operation_unref(o);
-
-		pa_threaded_mainloop_signal(pm->mainloop, 0);
-		break;
-	}
-
-	case PA_CONTEXT_UNCONNECTED:
-	case PA_CONTEXT_CONNECTING:
-	case PA_CONTEXT_AUTHORIZING:
-	case PA_CONTEXT_SETTING_NAME:
-		break;
-	case PA_CONTEXT_TERMINATED:
-	case PA_CONTEXT_FAILED:
-		pa_threaded_mainloop_signal(pm->mainloop, 0);
-		break;
+		if (pm->output->stream != NULL)
+			pulse_mixer_update(pm);
 	}
 }
 
-
 static struct mixer *
-pulse_mixer_init(G_GNUC_UNUSED void *ao, const struct config_param *param,
-		 G_GNUC_UNUSED GError **error_r)
+pulse_mixer_init(void *ao, G_GNUC_UNUSED const struct config_param *param,
+		 GError **error_r)
 {
-	struct pulse_mixer *pm = g_new(struct pulse_mixer,1);
+	struct pulse_mixer *pm;
+
+	if (ao == NULL) {
+		g_set_error(error_r, pulse_mixer_quark(), 0,
+			    "The pulse mixer cannot work without the audio output");
+		return false;
+	}
+
+	pm = g_new(struct pulse_mixer,1);
 	mixer_init(&pm->base, &pulse_mixer_plugin);
 
+	pm->output = ao;
 	pm->online = false;
 
-	pm->server = config_get_block_string(param, "server", NULL);
-	pm->sink = config_get_block_string(param, "sink", NULL);
-	pm->output_name = config_get_block_string(param, "name", NULL);
+	pa_threaded_mainloop_lock(pm->output->mainloop);
+
+	/* register callbacks (override the output's context state
+	   callback) */
+
+	pa_context_set_state_callback(pm->output->context,
+				      pulxe_mixer_context_state_cb, pm);
+	pa_context_set_subscribe_callback(pm->output->context,
+					  pulse_mixer_subscribe_cb, pm);
+
+	/* check the current state now (we might have missed the first
+	   events!) */
+	pulxe_mixer_context_state_cb(pm->output->context, pm);
+
+	pa_threaded_mainloop_unlock(pm->output->mainloop);
 
 	return &pm->base;
 }
@@ -232,79 +218,35 @@ pulse_mixer_finish(struct mixer *data)
 {
 	struct pulse_mixer *pm = (struct pulse_mixer *) data;
 
-	g_free(pm);
-}
+	/* restore callbacks */
 
-static bool
-pulse_mixer_setup(struct pulse_mixer *pm, GError **error_r)
-{
-	pa_context_set_state_callback(pm->context, context_state_cb, pm);
+	pa_threaded_mainloop_lock(pm->output->mainloop);
 
-	if (pa_context_connect(pm->context, pm->server,
-			       (pa_context_flags_t)0, NULL) < 0) {
-		g_set_error(error_r, pulse_mixer_quark(), 0,
-			    "pa_context_connect() has failed");
-		return false;
+	if (pm->output->context != NULL) {
+		pa_context_set_state_callback(pm->output->context,
+					      pulse_output_context_state_cb,
+					      pm->output);
+		pa_context_set_subscribe_callback(pm->output->context,
+						  NULL, NULL);
 	}
 
-	pa_threaded_mainloop_lock(pm->mainloop);
+	pa_threaded_mainloop_unlock(pm->output->mainloop);
 
-	if (pa_threaded_mainloop_start(pm->mainloop) < 0) {
-		pa_threaded_mainloop_unlock(pm->mainloop);
-		g_set_error(error_r, pulse_mixer_quark(), 0,
-			    "pa_threaded_mainloop_start() has failed");
-		return false;
-	}
-
-	pa_threaded_mainloop_wait(pm->mainloop);
-
-	if (pa_context_get_state(pm->context) != PA_CONTEXT_READY) {
-		g_set_error(error_r, pulse_mixer_quark(), 0,
-			    "failed to connect: %s",
-			    pa_strerror(pa_context_errno(pm->context)));
-		pa_threaded_mainloop_unlock(pm->mainloop);
-		return false;
-	}
+	/* free resources */
 
-	pa_threaded_mainloop_unlock(pm->mainloop);
-
-	return true;
+	g_free(pm);
 }
 
 static bool
-pulse_mixer_open(struct mixer *data, GError **error_r)
+pulse_mixer_open(struct mixer *data, G_GNUC_UNUSED GError **error_r)
 {
 	struct pulse_mixer *pm = (struct pulse_mixer *) data;
 
-	g_debug("pulse mixer open");
-
-	pm->index = 0;
-	pm->online = false;
-
-	pm->mainloop = pa_threaded_mainloop_new();
-	if (pm->mainloop == NULL) {
-		g_set_error(error_r, pulse_mixer_quark(), 0,
-			    "pa_threaded_mainloop_new() has failed");
-		return false;
-	}
-
-	pm->context = pa_context_new(pa_threaded_mainloop_get_api(pm->mainloop),
-				     "Mixer mpd");
-	if (pm->context == NULL) {
-		pa_threaded_mainloop_stop(pm->mainloop);
-		pa_threaded_mainloop_free(pm->mainloop);
-		g_set_error(error_r, pulse_mixer_quark(), 0,
-			    "pa_context_new() has failed");
-		return false;
-	}
-
-	if (!pulse_mixer_setup(pm, error_r)) {
-		pa_threaded_mainloop_stop(pm->mainloop);
-		pa_context_disconnect(pm->context);
-		pa_context_unref(pm->context);
-		pa_threaded_mainloop_free(pm->mainloop);
-		return false;
-	}
+	pa_threaded_mainloop_lock(pm->output->mainloop);
+	if (pm->output->stream != NULL &&
+	    pa_stream_get_state(pm->output->stream) == PA_STREAM_READY)
+		pulse_mixer_update(pm);
+	pa_threaded_mainloop_unlock(pm->output->mainloop);
 
 	return true;
 }
@@ -314,49 +256,22 @@ pulse_mixer_close(struct mixer *data)
 {
 	struct pulse_mixer *pm = (struct pulse_mixer *) data;
 
-	pa_threaded_mainloop_stop(pm->mainloop);
-	pa_context_disconnect(pm->context);
-	pa_context_unref(pm->context);
-	pa_threaded_mainloop_free(pm->mainloop);
-
-	pm->online = false;
+	pulse_mixer_offline(pm);
 }
 
 static int
-pulse_mixer_get_volume(struct mixer *mixer, GError **error_r)
+pulse_mixer_get_volume(struct mixer *mixer, G_GNUC_UNUSED GError **error_r)
 {
 	struct pulse_mixer *pm = (struct pulse_mixer *) mixer;
 	int ret;
-	pa_operation *o;
 
-	pa_threaded_mainloop_lock(pm->mainloop);
-
-	if (!pm->online) {
-		pa_threaded_mainloop_unlock(pm->mainloop);
-		return false;
-	}
-
-	o = pa_context_get_sink_input_info(pm->context, pm->index,
-					   sink_input_vol, pm);
-	if (o == NULL) {
-		pa_threaded_mainloop_unlock(pm->mainloop);
-		g_set_error(error_r, pulse_mixer_quark(), 0,
-			    "pa_context_get_sink_input_info() has failed");
-		return false;
-	}
-
-	if (!pulse_wait_for_operation(pm->mainloop, o)) {
-		pa_threaded_mainloop_unlock(pm->mainloop);
-		g_set_error(error_r, pulse_mixer_quark(), 0,
-			    "failed to read PulseAudio volume");
-		return false;
-	}
+	pa_threaded_mainloop_lock(pm->output->mainloop);
 
 	ret = pm->online
 		? (int)((100*(pa_cvolume_avg(&pm->volume)+1))/PA_VOLUME_NORM)
 		: -1;
 
-	pa_threaded_mainloop_unlock(pm->mainloop);
+	pa_threaded_mainloop_unlock(pm->output->mainloop);
 
 	return ret;
 }
@@ -368,10 +283,11 @@ pulse_mixer_set_volume(struct mixer *mixer, unsigned volume, GError **error_r)
 	struct pa_cvolume cvolume;
 	pa_operation *o;
 
-	pa_threaded_mainloop_lock(pm->mainloop);
+	pa_threaded_mainloop_lock(pm->output->mainloop);
 
-	if (!pm->online) {
-		pa_threaded_mainloop_unlock(pm->mainloop);
+	if (!pm->online || pm->output->stream == NULL ||
+	    pm->output->context == NULL) {
+		pa_threaded_mainloop_unlock(pm->output->mainloop);
 		g_set_error(error_r, pulse_mixer_quark(), 0, "disconnected");
 		return false;
 	}
@@ -379,9 +295,10 @@ pulse_mixer_set_volume(struct mixer *mixer, unsigned volume, GError **error_r)
 	pa_cvolume_set(&cvolume, pm->volume.channels,
 		       (pa_volume_t)volume * PA_VOLUME_NORM / 100 + 0.5);
 
-	o = pa_context_set_sink_input_volume(pm->context, pm->index,
+	o = pa_context_set_sink_input_volume(pm->output->context,
+					     pa_stream_get_index(pm->output->stream),
 					     &cvolume, NULL, NULL);
-	pa_threaded_mainloop_unlock(pm->mainloop);
+	pa_threaded_mainloop_unlock(pm->output->mainloop);
 	if (o == NULL) {
 		g_set_error(error_r, pulse_mixer_quark(), 0,
 			    "failed to set PulseAudio volume");
diff --git a/src/output/pulse_output_plugin.c b/src/output/pulse_output_plugin.c
index 1b1c27575..c24a356f2 100644
--- a/src/output/pulse_output_plugin.c
+++ b/src/output/pulse_output_plugin.c
@@ -17,22 +17,20 @@
  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  */
 
+#include "pulse_output_plugin.h"
 #include "output_api.h"
 #include "mixer_list.h"
 
 #include <glib.h>
-#include <pulse/simple.h>
-#include <pulse/error.h>
 
-#define MPD_PULSE_NAME "Music Player Daemon"
+#include <pulse/thread-mainloop.h>
+#include <pulse/context.h>
+#include <pulse/stream.h>
+#include <pulse/error.h>
 
-struct pulse_data {
-	const char *name;
-	const char *server;
-	const char *sink;
+#include <assert.h>
 
-	pa_simple *s;
-};
+#define MPD_PULSE_NAME "Music Player Daemon"
 
 /**
  * The quark used for GError.domain.
@@ -43,76 +41,299 @@ pulse_output_quark(void)
 	return g_quark_from_static_string("pulse_output");
 }
 
-static struct pulse_data *pulse_new_data(void)
+/**
+ * \brief waits for a pulseaudio operation to finish, frees it and
+ *     unlocks the mainloop
+ * \param operation the operation to wait for
+ * \return true if operation has finished normally (DONE state),
+ *     false otherwise
+ */
+static bool
+pulse_wait_for_operation(struct pa_threaded_mainloop *mainloop,
+			 struct pa_operation *operation)
+{
+	pa_operation_state_t state;
+
+	assert(mainloop != NULL);
+	assert(operation != NULL);
+
+	state = pa_operation_get_state(operation);
+	while (state == PA_OPERATION_RUNNING) {
+		pa_threaded_mainloop_wait(mainloop);
+		state = pa_operation_get_state(operation);
+	}
+
+	pa_operation_unref(operation);
+
+	return state == PA_OPERATION_DONE;
+}
+
+/**
+ * Callback function for stream operation.  It just sends a signal to
+ * the caller thread, to wake pulse_wait_for_operation() up.
+ */
+static void
+pulse_output_stream_success_cb(G_GNUC_UNUSED pa_stream *s,
+			       G_GNUC_UNUSED int success, void *userdata)
+{
+	struct pulse_output *po = userdata;
+
+	pa_threaded_mainloop_signal(po->mainloop, 0);
+}
+
+void
+pulse_output_context_state_cb(struct pa_context *context, void *userdata)
+{
+	struct pulse_output *po = userdata;
+
+	switch (pa_context_get_state(context)) {
+	case PA_CONTEXT_READY:
+	case PA_CONTEXT_TERMINATED:
+	case PA_CONTEXT_FAILED:
+		/* the caller thread might be waiting for these
+		   states */
+		pa_threaded_mainloop_signal(po->mainloop, 0);
+		break;
+
+	case PA_CONTEXT_UNCONNECTED:
+	case PA_CONTEXT_CONNECTING:
+	case PA_CONTEXT_AUTHORIZING:
+	case PA_CONTEXT_SETTING_NAME:
+		break;
+	}
+}
+
+/**
+ * Attempt to connect asynchronously to the PulseAudio server.
+ *
+ * @return true on success, false on error
+ */
+static bool
+pulse_output_connect(struct pulse_output *po, GError **error_r)
 {
-	struct pulse_data *ret;
+	int error;
 
-	ret = g_new(struct pulse_data, 1);
+	error = pa_context_connect(po->context, po->server,
+				   (pa_context_flags_t)0, NULL);
+	if (error < 0) {
+		g_set_error(error_r, pulse_output_quark(), 0,
+			    "pa_context_connect() has failed: %s",
+			    pa_strerror(pa_context_errno(po->context)));
+		return false;
+	}
 
-	ret->server = NULL;
-	ret->sink = NULL;
+	return true;
+}
 
-	return ret;
+/**
+ * Create, set up and connect a context.
+ *
+ * @return true on success, false on error
+ */
+static bool
+pulse_output_setup_context(struct pulse_output *po, GError **error_r)
+{
+	po->context = pa_context_new(pa_threaded_mainloop_get_api(po->mainloop),
+				     MPD_PULSE_NAME);
+	if (po->context == NULL) {
+		g_set_error(error_r, pulse_output_quark(), 0,
+			    "pa_context_new() has failed");
+		return false;
+	}
+
+	pa_context_set_state_callback(po->context,
+				      pulse_output_context_state_cb, po);
+
+	if (!pulse_output_connect(po, error_r)) {
+		pa_context_unref(po->context);
+		return false;
+	}
+
+	return true;
 }
 
-static void pulse_free_data(struct pulse_data *pd)
+/**
+ * Frees and clears the context.
+ */
+static void
+pulse_output_delete_context(struct pulse_output *po)
 {
-	g_free(pd);
+	pa_context_disconnect(po->context);
+	pa_context_unref(po->context);
+	po->context = NULL;
 }
 
 static void *
-pulse_init(G_GNUC_UNUSED const struct audio_format *audio_format,
-	   const struct config_param *param, G_GNUC_UNUSED GError **error)
+pulse_output_init(G_GNUC_UNUSED const struct audio_format *audio_format,
+		  const struct config_param *param, GError **error_r)
 {
-	struct pulse_data *pd;
+	struct pulse_output *po;
 
 	g_setenv("PULSE_PROP_media.role", "music", true);
 
-	pd = pulse_new_data();
-	pd->name = config_get_block_string(param, "name", "mpd_pulse");
-	pd->server = config_get_block_string(param, "server", NULL);
-	pd->sink = config_get_block_string(param, "sink", NULL);
+	po = g_new(struct pulse_output, 1);
+	po->name = config_get_block_string(param, "name", "mpd_pulse");
+	po->server = config_get_block_string(param, "server", NULL);
+	po->sink = config_get_block_string(param, "sink", NULL);
 
-	return pd;
+	/* create the libpulse mainloop and start the thread */
+
+	po->mainloop = pa_threaded_mainloop_new();
+	if (po->mainloop == NULL) {
+		g_free(po);
+
+		g_set_error(error_r, pulse_output_quark(), 0,
+			    "pa_threaded_mainloop_new() has failed");
+		return NULL;
+	}
+
+	pa_threaded_mainloop_lock(po->mainloop);
+
+	if (pa_threaded_mainloop_start(po->mainloop) < 0) {
+		pa_threaded_mainloop_unlock(po->mainloop);
+		pa_threaded_mainloop_free(po->mainloop);
+		g_free(po);
+
+		g_set_error(error_r, pulse_output_quark(), 0,
+			    "pa_threaded_mainloop_start() has failed");
+		return false;
+	}
+
+	pa_threaded_mainloop_unlock(po->mainloop);
+
+	/* create the libpulse context and connect it */
+
+	pa_threaded_mainloop_lock(po->mainloop);
+
+	if (!pulse_output_setup_context(po, error_r)) {
+		pa_threaded_mainloop_unlock(po->mainloop);
+		pa_threaded_mainloop_stop(po->mainloop);
+		pa_threaded_mainloop_free(po->mainloop);
+		g_free(po);
+		return NULL;
+	}
+
+	pa_threaded_mainloop_unlock(po->mainloop);
+
+	return po;
 }
 
-static void pulse_finish(void *data)
+static void
+pulse_output_finish(void *data)
 {
-	struct pulse_data *pd = data;
+	struct pulse_output *po = data;
+
+	pa_threaded_mainloop_stop(po->mainloop);
+	if (po->context != NULL)
+		pulse_output_delete_context(po);
+	pa_threaded_mainloop_free(po->mainloop);
 
-	pulse_free_data(pd);
+	g_free(po);
 }
 
-static bool pulse_test_default_device(void)
+/**
+ * Check if the context is (already) connected, and waits if not.  If
+ * the context has been disconnected, retry to connect.
+ *
+ * @return true on success, false on error
+ */
+static bool
+pulse_output_wait_connection(struct pulse_output *po, GError **error_r)
 {
-	pa_simple *s;
-	pa_sample_spec ss;
-	int error;
+	pa_context_state_t state;
 
-	ss.format = PA_SAMPLE_S16NE;
-	ss.rate = 44100;
-	ss.channels = 2;
-
-	s = pa_simple_new(NULL, MPD_PULSE_NAME, PA_STREAM_PLAYBACK, NULL,
-			  MPD_PULSE_NAME, &ss, NULL, NULL, &error);
-	if (!s) {
-		g_message("Cannot connect to default PulseAudio server: %s\n",
-			  pa_strerror(error));
+	pa_threaded_mainloop_lock(po->mainloop);
+
+	if (po->context == NULL && !pulse_output_setup_context(po, error_r))
 		return false;
+
+	while (true) {
+		state = pa_context_get_state(po->context);
+		switch (state) {
+		case PA_CONTEXT_READY:
+			/* nothing to do */
+			pa_threaded_mainloop_unlock(po->mainloop);
+			return true;
+
+		case PA_CONTEXT_UNCONNECTED:
+		case PA_CONTEXT_TERMINATED:
+		case PA_CONTEXT_FAILED:
+			/* failure */
+			g_set_error(error_r, pulse_output_quark(), 0,
+				    "failed to connect: %s",
+				    pa_strerror(pa_context_errno(po->context)));
+			pulse_output_delete_context(po);
+			pa_threaded_mainloop_unlock(po->mainloop);
+			return false;
+
+		case PA_CONTEXT_CONNECTING:
+		case PA_CONTEXT_AUTHORIZING:
+		case PA_CONTEXT_SETTING_NAME:
+			/* wait some more */
+			pa_threaded_mainloop_wait(po->mainloop);
+			break;
+		}
 	}
+}
 
-	pa_simple_free(s);
+static void
+pulse_output_stream_state_cb(pa_stream *stream, void *userdata)
+{
+	struct pulse_output *po = userdata;
+
+	switch (pa_stream_get_state(stream)) {
+	case PA_STREAM_READY:
+	case PA_STREAM_FAILED:
+	case PA_STREAM_TERMINATED:
+		pa_threaded_mainloop_signal(po->mainloop, 0);
+		break;
+
+	case PA_STREAM_UNCONNECTED:
+	case PA_STREAM_CREATING:
+		break;
+	}
+}
 
-	return true;
+static void
+pulse_output_stream_write_cb(G_GNUC_UNUSED pa_stream *stream, size_t nbytes,
+			     void *userdata)
+{
+	struct pulse_output *po = userdata;
+
+	po->writable = nbytes;
+	pa_threaded_mainloop_signal(po->mainloop, 0);
 }
 
 static bool
-pulse_open(void *data, struct audio_format *audio_format, GError **error_r)
+pulse_output_open(void *data, struct audio_format *audio_format,
+		  GError **error_r)
 {
-	struct pulse_data *pd = data;
+	struct pulse_output *po = data;
 	pa_sample_spec ss;
 	int error;
 
+	if (po->context != NULL) {
+		switch (pa_context_get_state(po->context)) {
+		case PA_CONTEXT_UNCONNECTED:
+		case PA_CONTEXT_TERMINATED:
+		case PA_CONTEXT_FAILED:
+			/* the connection was closed meanwhile; delete
+			   it, and pulse_output_wait_connection() will
+			   reopen it */
+			pulse_output_delete_context(po);
+			break;
+
+		case PA_CONTEXT_READY:
+		case PA_CONTEXT_CONNECTING:
+		case PA_CONTEXT_AUTHORIZING:
+		case PA_CONTEXT_SETTING_NAME:
+			break;
+		}
+	}
+
+	if (!pulse_output_wait_connection(po, error_r))
+		return false;
+
 	/* MPD doesn't support the other pulseaudio sample formats, so
 	   we just force MPD to send us everything as 16 bit */
 	audio_format->bits = 16;
@@ -121,45 +342,243 @@ pulse_open(void *data, struct audio_format *audio_format, GError **error_r)
 	ss.rate = audio_format->sample_rate;
 	ss.channels = audio_format->channels;
 
-	pd->s = pa_simple_new(pd->server, MPD_PULSE_NAME, PA_STREAM_PLAYBACK,
-			      pd->sink, pd->name,
-			      &ss, NULL, NULL,
-			      &error);
-	if (!pd->s) {
-		g_set_error(error_r, pulse_output_quark(), error,
-			    "Cannot connect to PulseAudio server: %s",
-			    pa_strerror(error));
+	pa_threaded_mainloop_lock(po->mainloop);
+
+	/* create a stream .. */
+
+	po->stream = pa_stream_new(po->context, po->name, &ss, NULL);
+	if (po->stream == NULL) {
+		g_set_error(error_r, pulse_output_quark(), 0,
+			    "pa_stream_new() has failed: %s",
+			    pa_strerror(pa_context_errno(po->context)));
+		pa_threaded_mainloop_unlock(po->mainloop);
+		return false;
+	}
+
+	pa_stream_set_state_callback(po->stream,
+				     pulse_output_stream_state_cb, po);
+	pa_stream_set_write_callback(po->stream,
+				     pulse_output_stream_write_cb, po);
+
+	/* .. and connect it (asynchronously) */
+
+	error = pa_stream_connect_playback(po->stream, po->sink,
+					   NULL, 0, NULL, NULL);
+	if (error < 0) {
+		pa_stream_unref(po->stream);
+		po->stream = NULL;
+
+		g_set_error(error_r, pulse_output_quark(), 0,
+			    "pa_stream_connect_playback() has failed: %s",
+			    pa_strerror(pa_context_errno(po->context)));
+		pa_threaded_mainloop_unlock(po->mainloop);
 		return false;
 	}
 
+	pa_threaded_mainloop_unlock(po->mainloop);
+
+#if !PA_CHECK_VERSION(0,9,11)
+	po->pause = false;
+#endif
+
 	return true;
 }
 
-static void pulse_cancel(void *data)
+static void
+pulse_output_close(void *data)
 {
-	struct pulse_data *pd = data;
-	int error;
+	struct pulse_output *po = data;
+	pa_operation *o;
+
+	pa_threaded_mainloop_lock(po->mainloop);
+
+	if (pa_stream_get_state(po->stream) == PA_STREAM_READY) {
+		o = pa_stream_drain(po->stream,
+				    pulse_output_stream_success_cb, po);
+		if (o == NULL) {
+			g_warning("pa_stream_drain() has failed: %s",
+				  pa_strerror(pa_context_errno(po->context)));
+		} else
+			pulse_wait_for_operation(po->mainloop, o);
+	}
+
+	pa_stream_disconnect(po->stream);
+	pa_stream_unref(po->stream);
+	po->stream = NULL;
 
-	if (pa_simple_flush(pd->s, &error) < 0)
-		g_warning("Flush failed in PulseAudio output \"%s\": %s\n",
-			  pd->name, pa_strerror(error));
+	if (po->context != NULL &&
+	    pa_context_get_state(po->context) != PA_CONTEXT_READY)
+		pulse_output_delete_context(po);
+
+	pa_threaded_mainloop_unlock(po->mainloop);
+}
+
+/**
+ * Check if the stream is (already) connected, and waits for a signal
+ * if not.  The mainloop must be locked before calling this function.
+ *
+ * @return the current stream state
+ */
+static pa_stream_state_t
+pulse_output_check_stream(struct pulse_output *po)
+{
+	pa_stream_state_t state = pa_stream_get_state(po->stream);
+
+	switch (state) {
+	case PA_STREAM_READY:
+	case PA_STREAM_FAILED:
+	case PA_STREAM_TERMINATED:
+	case PA_STREAM_UNCONNECTED:
+		break;
+
+	case PA_STREAM_CREATING:
+		pa_threaded_mainloop_wait(po->mainloop);
+		state = pa_stream_get_state(po->stream);
+		break;
+	}
+
+	return state;
+}
+
+/**
+ * Check if the stream is (already) connected, and waits if not.  The
+ * mainloop must be locked before calling this function.
+ *
+ * @return true on success, false on error
+ */
+static bool
+pulse_output_wait_stream(struct pulse_output *po, GError **error_r)
+{
+	pa_stream_state_t state = pa_stream_get_state(po->stream);
+
+	switch (state) {
+	case PA_STREAM_READY:
+		return true;
+
+	case PA_STREAM_FAILED:
+	case PA_STREAM_TERMINATED:
+	case PA_STREAM_UNCONNECTED:
+		g_set_error(error_r, pulse_output_quark(), 0,
+			    "disconnected");
+		return false;
+
+	case PA_STREAM_CREATING:
+		break;
+	}
+
+	do {
+		state = pulse_output_check_stream(po);
+	} while (state == PA_STREAM_CREATING);
+
+	if (state != PA_STREAM_READY) {
+		g_set_error(error_r, pulse_output_quark(), 0,
+			    "failed to connect the stream: %s",
+			    pa_strerror(pa_context_errno(po->context)));
+		return false;
+	}
+
+	return true;
 }
 
-static void pulse_close(void *data)
+/**
+ * Determines whether the stream is paused.  On libpulse older than
+ * 0.9.11, it uses a custom pause flag.
+ */
+static bool
+pulse_output_stream_is_paused(struct pulse_output *po)
 {
-	struct pulse_data *pd = data;
+	assert(po->stream != NULL);
 
-	pa_simple_drain(pd->s, NULL);
-	pa_simple_free(pd->s);
+#if !defined(PA_CHECK_VERSION) || !PA_CHECK_VERSION(0,9,11)
+	return po->pause;
+#else
+	return pa_stream_is_corked(po->stream);
+#endif
+}
+
+/**
+ * Sets cork mode on the stream.
+ */
+static bool
+pulse_output_stream_pause(struct pulse_output *po, bool pause,
+			  GError **error_r)
+{
+	pa_operation *o;
+
+	assert(po->stream != NULL);
+
+	o = pa_stream_cork(po->stream, pause,
+			   pulse_output_stream_success_cb, po);
+	if (o == NULL) {
+		g_set_error(error_r, pulse_output_quark(), 0,
+			    "pa_stream_cork() has failed: %s",
+			    pa_strerror(pa_context_errno(po->context)));
+		return false;
+	}
+
+	if (!pulse_wait_for_operation(po->mainloop, o)) {
+		g_set_error(error_r, pulse_output_quark(), 0,
+			    "pa_stream_cork() has failed: %s",
+			    pa_strerror(pa_context_errno(po->context)));
+		return false;
+	}
+
+#if !PA_CHECK_VERSION(0,9,11)
+	po->pause = pause;
+#endif
+	return true;
 }
 
 static size_t
-pulse_play(void *data, const void *chunk, size_t size, GError **error_r)
+pulse_output_play(void *data, const void *chunk, size_t size, GError **error_r)
 {
-	struct pulse_data *pd = data;
+	struct pulse_output *po = data;
 	int error;
 
-	if (pa_simple_write(pd->s, chunk, size, &error) < 0) {
+	assert(po->stream != NULL);
+
+	pa_threaded_mainloop_lock(po->mainloop);
+
+	/* check if the stream is (already) connected */
+
+	if (!pulse_output_wait_stream(po, error_r)) {
+		pa_threaded_mainloop_unlock(po->mainloop);
+		return 0;
+	}
+
+	assert(po->context != NULL);
+
+	/* unpause if previously paused */
+
+	if (pulse_output_stream_is_paused(po) &&
+	    !pulse_output_stream_pause(po, false, error_r))
+		return 0;
+
+	/* wait until the server allows us to write */
+
+	while (po->writable == 0) {
+		pa_threaded_mainloop_wait(po->mainloop);
+
+		if (pa_stream_get_state(po->stream) != PA_STREAM_READY) {
+			pa_threaded_mainloop_unlock(po->mainloop);
+			g_set_error(error_r, pulse_output_quark(), 0,
+				    "disconnected");
+			return false;
+		}
+	}
+
+	/* now write */
+
+	if (size > po->writable)
+		/* don't send more than possible */
+		size = po->writable;
+
+	po->writable -= size;
+
+	error = pa_stream_write(po->stream, chunk, size, NULL,
+				0, PA_SEEK_RELATIVE);
+	pa_threaded_mainloop_unlock(po->mainloop);
+	if (error < 0) {
 		g_set_error(error_r, pulse_output_quark(), error,
 			    "%s", pa_strerror(error));
 		return 0;
@@ -168,16 +587,105 @@ pulse_play(void *data, const void *chunk, size_t size, GError **error_r)
 	return size;
 }
 
+static void
+pulse_output_cancel(void *data)
+{
+	struct pulse_output *po = data;
+	pa_operation *o;
+
+	assert(po->stream != NULL);
+
+	pa_threaded_mainloop_lock(po->mainloop);
+
+	if (pa_stream_get_state(po->stream) != PA_STREAM_READY) {
+		/* no need to flush when the stream isn't connected
+		   yet */
+		pa_threaded_mainloop_unlock(po->mainloop);
+		return;
+	}
+
+	assert(po->context != NULL);
+
+	o = pa_stream_flush(po->stream, pulse_output_stream_success_cb, po);
+	if (o == NULL) {
+		g_warning("pa_stream_flush() has failed: %s",
+			  pa_strerror(pa_context_errno(po->context)));
+		pa_threaded_mainloop_unlock(po->mainloop);
+		return;
+	}
+
+	pulse_wait_for_operation(po->mainloop, o);
+	pa_threaded_mainloop_unlock(po->mainloop);
+}
+
+static bool
+pulse_output_pause(void *data)
+{
+	struct pulse_output *po = data;
+	GError *error = NULL;
+
+	assert(po->stream != NULL);
+
+	pa_threaded_mainloop_lock(po->mainloop);
+
+	/* check if the stream is (already/still) connected */
+
+	if (!pulse_output_wait_stream(po, &error)) {
+		pa_threaded_mainloop_unlock(po->mainloop);
+		g_warning("%s", error->message);
+		g_error_free(error);
+		return false;
+	}
+
+	assert(po->context != NULL);
+
+	/* cork the stream */
+
+	if (pulse_output_stream_is_paused(po)) {
+		/* already paused; due to a MPD API limitation, we
+		   have to sleep a little bit here, to avoid hogging
+		   the CPU */
+
+		g_usleep(50000);
+	} else if (!pulse_output_stream_pause(po, true, &error)) {
+		pa_threaded_mainloop_unlock(po->mainloop);
+		g_warning("%s", error->message);
+		g_error_free(error);
+		return false;
+	}
+
+	pa_threaded_mainloop_unlock(po->mainloop);
+
+	return true;
+}
+
+static bool
+pulse_output_test_default_device(void)
+{
+	struct pulse_output *po;
+	bool success;
+
+	po = pulse_output_init(NULL, NULL, NULL);
+	if (po == NULL)
+		return false;
+
+	success = pulse_output_wait_connection(po, NULL);
+	pulse_output_finish(po);
+
+	return success;
+}
+
 const struct audio_output_plugin pulse_output_plugin = {
 	.name = "pulse",
 
-	.test_default_device = pulse_test_default_device,
-	.init = pulse_init,
-	.finish = pulse_finish,
-	.open = pulse_open,
-	.play = pulse_play,
-	.cancel = pulse_cancel,
-	.close = pulse_close,
+	.test_default_device = pulse_output_test_default_device,
+	.init = pulse_output_init,
+	.finish = pulse_output_finish,
+	.open = pulse_output_open,
+	.play = pulse_output_play,
+	.cancel = pulse_output_cancel,
+	.pause = pulse_output_pause,
+	.close = pulse_output_close,
 
 	.mixer_plugin = &pulse_mixer_plugin,
 };
diff --git a/src/output/pulse_output_plugin.h b/src/output/pulse_output_plugin.h
new file mode 100644
index 000000000..fc2a7d4d5
--- /dev/null
+++ b/src/output/pulse_output_plugin.h
@@ -0,0 +1,58 @@
+/*
+ * Copyright (C) 2003-2009 The Music Player Daemon Project
+ * http://www.musicpd.org
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#ifndef MPD_PULSE_OUTPUT_PLUGIN_H
+#define MPD_PULSE_OUTPUT_PLUGIN_H
+
+#include <stdbool.h>
+#include <stddef.h>
+
+#if !defined(PA_CHECK_VERSION)
+/**
+ * This macro was implemented in libpulse 0.9.16.
+ */
+#define PA_CHECK_VERSION(a,b,c) false
+#endif
+
+struct pa_operation;
+
+struct pulse_output {
+	const char *name;
+	const char *server;
+	const char *sink;
+
+	struct pa_threaded_mainloop *mainloop;
+	struct pa_context *context;
+	struct pa_stream *stream;
+
+	size_t writable;
+
+#if !PA_CHECK_VERSION(0,9,11)
+	/**
+	 * We need this variable because pa_stream_is_corked() wasn't
+	 * added before 0.9.11.
+	 */
+	bool pause;
+#endif
+};
+
+void
+pulse_output_context_state_cb(struct pa_context *context, void *userdata);
+
+#endif