diff options
Diffstat (limited to 'src/output/plugins/PulseOutputPlugin.cxx')
| -rw-r--r-- | src/output/plugins/PulseOutputPlugin.cxx | 889 |
1 files changed, 889 insertions, 0 deletions
diff --git a/src/output/plugins/PulseOutputPlugin.cxx b/src/output/plugins/PulseOutputPlugin.cxx new file mode 100644 index 000000000..c133d9796 --- /dev/null +++ b/src/output/plugins/PulseOutputPlugin.cxx @@ -0,0 +1,889 @@ +/* + * Copyright (C) 2003-2014 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. + */ + +#include "config.h" +#include "PulseOutputPlugin.hxx" +#include "../OutputAPI.hxx" +#include "MixerList.hxx" +#include "mixer/PulseMixerPlugin.hxx" +#include "util/Error.hxx" +#include "util/Domain.hxx" +#include "Log.hxx" + +#include <glib.h> + +#include <pulse/thread-mainloop.h> +#include <pulse/context.h> +#include <pulse/stream.h> +#include <pulse/introspect.h> +#include <pulse/subscribe.h> +#include <pulse/error.h> +#include <pulse/version.h> + +#include <assert.h> +#include <stddef.h> + +#define MPD_PULSE_NAME "Music Player Daemon" + +struct PulseOutput { + struct audio_output base; + + const char *name; + const char *server; + const char *sink; + + PulseMixer *mixer; + + struct pa_threaded_mainloop *mainloop; + struct pa_context *context; + struct pa_stream *stream; + + size_t writable; +}; + +static constexpr Domain pulse_output_domain("pulse_output"); + +static void +SetError(Error &error, pa_context *context, const char *msg) +{ + const int e = pa_context_errno(context); + error.Format(pulse_output_domain, e, "%s: %s", msg, pa_strerror(e)); +} + +void +pulse_output_lock(PulseOutput *po) +{ + pa_threaded_mainloop_lock(po->mainloop); +} + +void +pulse_output_unlock(PulseOutput *po) +{ + pa_threaded_mainloop_unlock(po->mainloop); +} + +void +pulse_output_set_mixer(PulseOutput *po, PulseMixer *pm) +{ + assert(po != nullptr); + assert(po->mixer == nullptr); + assert(pm != nullptr); + + po->mixer = pm; + + if (po->mainloop == nullptr) + return; + + pa_threaded_mainloop_lock(po->mainloop); + + if (po->context != nullptr && + pa_context_get_state(po->context) == PA_CONTEXT_READY) { + pulse_mixer_on_connect(pm, po->context); + + if (po->stream != nullptr && + pa_stream_get_state(po->stream) == PA_STREAM_READY) + pulse_mixer_on_change(pm, po->context, po->stream); + } + + pa_threaded_mainloop_unlock(po->mainloop); +} + +void +pulse_output_clear_mixer(PulseOutput *po, gcc_unused PulseMixer *pm) +{ + assert(po != nullptr); + assert(pm != nullptr); + assert(po->mixer == pm); + + po->mixer = nullptr; +} + +bool +pulse_output_set_volume(PulseOutput *po, const pa_cvolume *volume, + Error &error) +{ + pa_operation *o; + + if (po->context == nullptr || po->stream == nullptr || + pa_stream_get_state(po->stream) != PA_STREAM_READY) { + error.Set(pulse_output_domain, "disconnected"); + return false; + } + + o = pa_context_set_sink_input_volume(po->context, + pa_stream_get_index(po->stream), + volume, nullptr, nullptr); + if (o == nullptr) { + SetError(error, po->context, + "failed to set PulseAudio volume"); + return false; + } + + pa_operation_unref(o); + return true; +} + +/** + * \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 != nullptr); + assert(operation != nullptr); + + 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(gcc_unused pa_stream *s, + gcc_unused int success, void *userdata) +{ + PulseOutput *po = (PulseOutput *)userdata; + + pa_threaded_mainloop_signal(po->mainloop, 0); +} + +static void +pulse_output_context_state_cb(struct pa_context *context, void *userdata) +{ + PulseOutput *po = (PulseOutput *)userdata; + + switch (pa_context_get_state(context)) { + case PA_CONTEXT_READY: + if (po->mixer != nullptr) + pulse_mixer_on_connect(po->mixer, context); + + pa_threaded_mainloop_signal(po->mainloop, 0); + break; + + case PA_CONTEXT_TERMINATED: + case PA_CONTEXT_FAILED: + if (po->mixer != nullptr) + pulse_mixer_on_disconnect(po->mixer); + + /* 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; + } +} + +static void +pulse_output_subscribe_cb(pa_context *context, + pa_subscription_event_type_t t, + uint32_t idx, void *userdata) +{ + PulseOutput *po = (PulseOutput *)userdata; + pa_subscription_event_type_t facility = + pa_subscription_event_type_t(t & PA_SUBSCRIPTION_EVENT_FACILITY_MASK); + pa_subscription_event_type_t type = + pa_subscription_event_type_t(t & PA_SUBSCRIPTION_EVENT_TYPE_MASK); + + if (po->mixer != nullptr && + facility == PA_SUBSCRIPTION_EVENT_SINK_INPUT && + po->stream != nullptr && + pa_stream_get_state(po->stream) == PA_STREAM_READY && + idx == pa_stream_get_index(po->stream) && + (type == PA_SUBSCRIPTION_EVENT_NEW || + type == PA_SUBSCRIPTION_EVENT_CHANGE)) + pulse_mixer_on_change(po->mixer, context, po->stream); +} + +/** + * Attempt to connect asynchronously to the PulseAudio server. + * + * @return true on success, false on error + */ +static bool +pulse_output_connect(PulseOutput *po, Error &error) +{ + assert(po != nullptr); + assert(po->context != nullptr); + + if (pa_context_connect(po->context, po->server, + (pa_context_flags_t)0, nullptr) < 0) { + SetError(error, po->context, + "pa_context_connect() has failed"); + return false; + } + + return true; +} + +/** + * Frees and clears the stream. + */ +static void +pulse_output_delete_stream(PulseOutput *po) +{ + assert(po != nullptr); + assert(po->stream != nullptr); + + pa_stream_set_suspended_callback(po->stream, nullptr, nullptr); + + pa_stream_set_state_callback(po->stream, nullptr, nullptr); + pa_stream_set_write_callback(po->stream, nullptr, nullptr); + + pa_stream_disconnect(po->stream); + pa_stream_unref(po->stream); + po->stream = nullptr; +} + +/** + * Frees and clears the context. + * + * Caller must lock the main loop. + */ +static void +pulse_output_delete_context(PulseOutput *po) +{ + assert(po != nullptr); + assert(po->context != nullptr); + + pa_context_set_state_callback(po->context, nullptr, nullptr); + pa_context_set_subscribe_callback(po->context, nullptr, nullptr); + + pa_context_disconnect(po->context); + pa_context_unref(po->context); + po->context = nullptr; +} + +/** + * Create, set up and connect a context. + * + * Caller must lock the main loop. + * + * @return true on success, false on error + */ +static bool +pulse_output_setup_context(PulseOutput *po, Error &error) +{ + assert(po != nullptr); + assert(po->mainloop != nullptr); + + po->context = pa_context_new(pa_threaded_mainloop_get_api(po->mainloop), + MPD_PULSE_NAME); + if (po->context == nullptr) { + error.Set(pulse_output_domain, "pa_context_new() has failed"); + return false; + } + + pa_context_set_state_callback(po->context, + pulse_output_context_state_cb, po); + pa_context_set_subscribe_callback(po->context, + pulse_output_subscribe_cb, po); + + if (!pulse_output_connect(po, error)) { + pulse_output_delete_context(po); + return false; + } + + return true; +} + +static struct audio_output * +pulse_output_init(const config_param ¶m, Error &error) +{ + PulseOutput *po; + + g_setenv("PULSE_PROP_media.role", "music", true); + + po = new PulseOutput(); + if (!ao_base_init(&po->base, &pulse_output_plugin, param, error)) { + delete po; + return nullptr; + } + + po->name = param.GetBlockValue("name", "mpd_pulse"); + po->server = param.GetBlockValue("server"); + po->sink = param.GetBlockValue("sink"); + + po->mixer = nullptr; + po->mainloop = nullptr; + po->context = nullptr; + po->stream = nullptr; + + return &po->base; +} + +static void +pulse_output_finish(struct audio_output *ao) +{ + PulseOutput *po = (PulseOutput *)ao; + + ao_base_finish(&po->base); + delete po; +} + +static bool +pulse_output_enable(struct audio_output *ao, Error &error) +{ + PulseOutput *po = (PulseOutput *)ao; + + assert(po->mainloop == nullptr); + assert(po->context == nullptr); + + /* create the libpulse mainloop and start the thread */ + + po->mainloop = pa_threaded_mainloop_new(); + if (po->mainloop == nullptr) { + g_free(po); + + error.Set(pulse_output_domain, + "pa_threaded_mainloop_new() has failed"); + return false; + } + + 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); + po->mainloop = nullptr; + + error.Set(pulse_output_domain, + "pa_threaded_mainloop_start() has failed"); + return false; + } + + /* create the libpulse context and connect it */ + + if (!pulse_output_setup_context(po, error)) { + pa_threaded_mainloop_unlock(po->mainloop); + pa_threaded_mainloop_stop(po->mainloop); + pa_threaded_mainloop_free(po->mainloop); + po->mainloop = nullptr; + return false; + } + + pa_threaded_mainloop_unlock(po->mainloop); + + return true; +} + +static void +pulse_output_disable(struct audio_output *ao) +{ + PulseOutput *po = (PulseOutput *)ao; + + assert(po->mainloop != nullptr); + + pa_threaded_mainloop_stop(po->mainloop); + if (po->context != nullptr) + pulse_output_delete_context(po); + pa_threaded_mainloop_free(po->mainloop); + po->mainloop = nullptr; +} + +/** + * Check if the context is (already) connected, and waits if not. If + * the context has been disconnected, retry to connect. + * + * Caller must lock the main loop. + * + * @return true on success, false on error + */ +static bool +pulse_output_wait_connection(PulseOutput *po, Error &error) +{ + assert(po->mainloop != nullptr); + + pa_context_state_t state; + + if (po->context == nullptr && !pulse_output_setup_context(po, error)) + return false; + + while (true) { + state = pa_context_get_state(po->context); + switch (state) { + case PA_CONTEXT_READY: + /* nothing to do */ + return true; + + case PA_CONTEXT_UNCONNECTED: + case PA_CONTEXT_TERMINATED: + case PA_CONTEXT_FAILED: + /* failure */ + SetError(error, po->context, "failed to connect"); + pulse_output_delete_context(po); + 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; + } + } +} + +static void +pulse_output_stream_suspended_cb(gcc_unused pa_stream *stream, void *userdata) +{ + PulseOutput *po = (PulseOutput *)userdata; + + assert(stream == po->stream || po->stream == nullptr); + assert(po->mainloop != nullptr); + + /* wake up the main loop to break out of the loop in + pulse_output_play() */ + pa_threaded_mainloop_signal(po->mainloop, 0); +} + +static void +pulse_output_stream_state_cb(pa_stream *stream, void *userdata) +{ + PulseOutput *po = (PulseOutput *)userdata; + + assert(stream == po->stream || po->stream == nullptr); + assert(po->mainloop != nullptr); + assert(po->context != nullptr); + + switch (pa_stream_get_state(stream)) { + case PA_STREAM_READY: + if (po->mixer != nullptr) + pulse_mixer_on_change(po->mixer, po->context, stream); + + pa_threaded_mainloop_signal(po->mainloop, 0); + break; + + case PA_STREAM_FAILED: + case PA_STREAM_TERMINATED: + if (po->mixer != nullptr) + pulse_mixer_on_disconnect(po->mixer); + + pa_threaded_mainloop_signal(po->mainloop, 0); + break; + + case PA_STREAM_UNCONNECTED: + case PA_STREAM_CREATING: + break; + } +} + +static void +pulse_output_stream_write_cb(gcc_unused pa_stream *stream, size_t nbytes, + void *userdata) +{ + PulseOutput *po = (PulseOutput *)userdata; + + assert(po->mainloop != nullptr); + + po->writable = nbytes; + pa_threaded_mainloop_signal(po->mainloop, 0); +} + +/** + * Create, set up and connect a context. + * + * Caller must lock the main loop. + * + * @return true on success, false on error + */ +static bool +pulse_output_setup_stream(PulseOutput *po, const pa_sample_spec *ss, + Error &error) +{ + assert(po != nullptr); + assert(po->context != nullptr); + + po->stream = pa_stream_new(po->context, po->name, ss, nullptr); + if (po->stream == nullptr) { + SetError(error, po->context, "pa_stream_new() has failed"); + return false; + } + + pa_stream_set_suspended_callback(po->stream, + pulse_output_stream_suspended_cb, po); + + 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); + + return true; +} + +static bool +pulse_output_open(struct audio_output *ao, AudioFormat &audio_format, + Error &error) +{ + PulseOutput *po = (PulseOutput *)ao; + pa_sample_spec ss; + + assert(po->mainloop != nullptr); + + pa_threaded_mainloop_lock(po->mainloop); + + if (po->context != nullptr) { + 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)) { + pa_threaded_mainloop_unlock(po->mainloop); + 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.format = SampleFormat::S16; + + ss.format = PA_SAMPLE_S16NE; + ss.rate = audio_format.sample_rate; + ss.channels = audio_format.channels; + + /* create a stream .. */ + + if (!pulse_output_setup_stream(po, &ss, error)) { + pa_threaded_mainloop_unlock(po->mainloop); + return false; + } + + /* .. and connect it (asynchronously) */ + + if (pa_stream_connect_playback(po->stream, po->sink, + nullptr, pa_stream_flags_t(0), + nullptr, nullptr) < 0) { + pulse_output_delete_stream(po); + + SetError(error, po->context, + "pa_stream_connect_playback() has failed"); + pa_threaded_mainloop_unlock(po->mainloop); + return false; + } + + pa_threaded_mainloop_unlock(po->mainloop); + + return true; +} + +static void +pulse_output_close(struct audio_output *ao) +{ + PulseOutput *po = (PulseOutput *)ao; + pa_operation *o; + + assert(po->mainloop != nullptr); + + 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 == nullptr) { + FormatWarning(pulse_output_domain, + "pa_stream_drain() has failed: %s", + pa_strerror(pa_context_errno(po->context))); + } else + pulse_wait_for_operation(po->mainloop, o); + } + + pulse_output_delete_stream(po); + + if (po->context != nullptr && + 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 if not. The + * mainloop must be locked before calling this function. + * + * @return true on success, false on error + */ +static bool +pulse_output_wait_stream(PulseOutput *po, Error &error) +{ + while (true) { + switch (pa_stream_get_state(po->stream)) { + case PA_STREAM_READY: + return true; + + case PA_STREAM_FAILED: + case PA_STREAM_TERMINATED: + case PA_STREAM_UNCONNECTED: + SetError(error, po->context, + "failed to connect the stream"); + return false; + + case PA_STREAM_CREATING: + pa_threaded_mainloop_wait(po->mainloop); + break; + } + } +} + +/** + * Sets cork mode on the stream. + */ +static bool +pulse_output_stream_pause(PulseOutput *po, bool pause, + Error &error) +{ + pa_operation *o; + + assert(po->mainloop != nullptr); + assert(po->context != nullptr); + assert(po->stream != nullptr); + + o = pa_stream_cork(po->stream, pause, + pulse_output_stream_success_cb, po); + if (o == nullptr) { + SetError(error, po->context, "pa_stream_cork() has failed"); + return false; + } + + if (!pulse_wait_for_operation(po->mainloop, o)) { + SetError(error, po->context, "pa_stream_cork() has failed"); + return false; + } + + return true; +} + +static unsigned +pulse_output_delay(struct audio_output *ao) +{ + PulseOutput *po = (PulseOutput *)ao; + unsigned result = 0; + + pa_threaded_mainloop_lock(po->mainloop); + + if (po->base.pause && pa_stream_is_corked(po->stream) && + pa_stream_get_state(po->stream) == PA_STREAM_READY) + /* idle while paused */ + result = 1000; + + pa_threaded_mainloop_unlock(po->mainloop); + + return result; +} + +static size_t +pulse_output_play(struct audio_output *ao, const void *chunk, size_t size, + Error &error) +{ + PulseOutput *po = (PulseOutput *)ao; + + assert(po->mainloop != nullptr); + assert(po->stream != nullptr); + + pa_threaded_mainloop_lock(po->mainloop); + + /* check if the stream is (already) connected */ + + if (!pulse_output_wait_stream(po, error)) { + pa_threaded_mainloop_unlock(po->mainloop); + return 0; + } + + assert(po->context != nullptr); + + /* unpause if previously paused */ + + if (pa_stream_is_corked(po->stream) && + !pulse_output_stream_pause(po, false, error)) { + pa_threaded_mainloop_unlock(po->mainloop); + return 0; + } + + /* wait until the server allows us to write */ + + while (po->writable == 0) { + if (pa_stream_is_suspended(po->stream)) { + pa_threaded_mainloop_unlock(po->mainloop); + error.Set(pulse_output_domain, "suspended"); + return 0; + } + + pa_threaded_mainloop_wait(po->mainloop); + + if (pa_stream_get_state(po->stream) != PA_STREAM_READY) { + pa_threaded_mainloop_unlock(po->mainloop); + error.Set(pulse_output_domain, "disconnected"); + return 0; + } + } + + /* now write */ + + if (size > po->writable) + /* don't send more than possible */ + size = po->writable; + + po->writable -= size; + + int result = pa_stream_write(po->stream, chunk, size, nullptr, + 0, PA_SEEK_RELATIVE); + pa_threaded_mainloop_unlock(po->mainloop); + if (result < 0) { + SetError(error, po->context, "pa_stream_write() failed"); + return 0; + } + + return size; +} + +static void +pulse_output_cancel(struct audio_output *ao) +{ + PulseOutput *po = (PulseOutput *)ao; + pa_operation *o; + + assert(po->mainloop != nullptr); + assert(po->stream != nullptr); + + 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 != nullptr); + + o = pa_stream_flush(po->stream, pulse_output_stream_success_cb, po); + if (o == nullptr) { + FormatWarning(pulse_output_domain, + "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(struct audio_output *ao) +{ + PulseOutput *po = (PulseOutput *)ao; + + assert(po->mainloop != nullptr); + assert(po->stream != nullptr); + + pa_threaded_mainloop_lock(po->mainloop); + + /* check if the stream is (already/still) connected */ + + Error error; + if (!pulse_output_wait_stream(po, error)) { + pa_threaded_mainloop_unlock(po->mainloop); + LogError(error); + return false; + } + + assert(po->context != nullptr); + + /* cork the stream */ + + if (!pa_stream_is_corked(po->stream) && + !pulse_output_stream_pause(po, true, error)) { + pa_threaded_mainloop_unlock(po->mainloop); + LogError(error); + return false; + } + + pa_threaded_mainloop_unlock(po->mainloop); + + return true; +} + +static bool +pulse_output_test_default_device(void) +{ + bool success; + + const config_param empty; + PulseOutput *po = (PulseOutput *) + pulse_output_init(empty, IgnoreError()); + if (po == nullptr) + return false; + + success = pulse_output_wait_connection(po, IgnoreError()); + pulse_output_finish(&po->base); + + return success; +} + +const struct audio_output_plugin pulse_output_plugin = { + "pulse", + pulse_output_test_default_device, + pulse_output_init, + pulse_output_finish, + pulse_output_enable, + pulse_output_disable, + pulse_output_open, + pulse_output_close, + pulse_output_delay, + nullptr, + pulse_output_play, + nullptr, + pulse_output_cancel, + pulse_output_pause, + + &pulse_mixer_plugin, +}; |