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".

1 files changed, 126 insertions, 209 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");