/*
* Copyright (C) 2003-2011 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 "pulse_output_plugin.h"
#include "output_api.h"
#include "mixer_list.h"
#include "mixer/pulse_mixer_plugin.h"
#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 <assert.h>
#define MPD_PULSE_NAME "Music Player Daemon"
/**
* The quark used for GError.domain.
*/
static inline GQuark
pulse_output_quark(void)
{
return g_quark_from_static_string("pulse_output");
}
void
pulse_output_set_mixer(struct pulse_output *po, struct pulse_mixer *pm)
{
assert(po != NULL);
assert(po->mixer == NULL);
assert(pm != NULL);
po->mixer = pm;
if (po->mainloop == NULL)
return;
pa_threaded_mainloop_lock(po->mainloop);
if (po->context != NULL &&
pa_context_get_state(po->context) == PA_CONTEXT_READY) {
pulse_mixer_on_connect(pm, po->context);
if (po->stream != NULL &&
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(struct pulse_output *po, struct pulse_mixer *pm)
{
assert(po != NULL);
assert(pm != NULL);
assert(po->mixer == pm);
po->mixer = NULL;
}
bool
pulse_output_set_volume(struct pulse_output *po,
const struct pa_cvolume *volume, GError **error_r)
{
pa_operation *o;
if (po->context == NULL || po->stream == NULL ||
pa_stream_get_state(po->stream) != PA_STREAM_READY) {
g_set_error(error_r, pulse_output_quark(), 0, "disconnected");
return false;
}
o = pa_context_set_sink_input_volume(po->context,
pa_stream_get_index(po->stream),
volume, NULL, NULL);
if (o == NULL) {
g_set_error(error_r, pulse_output_quark(), 0,
"failed to set PulseAudio volume: %s",
pa_strerror(pa_context_errno(po->context)));
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 != 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);
}
static 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:
if (po->mixer != NULL)
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 != NULL)
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)
{
struct pulse_output *po = userdata;
pa_subscription_event_type_t facility
= t & PA_SUBSCRIPTION_EVENT_FACILITY_MASK;
pa_subscription_event_type_t type
= t & PA_SUBSCRIPTION_EVENT_TYPE_MASK;
if (po->mixer != NULL &&
facility == PA_SUBSCRIPTION_EVENT_SINK_INPUT &&
po->stream != NULL &&
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(struct pulse_output *po, GError **error_r)
{
assert(po != NULL);
assert(po->context != NULL);
int error;
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;
}
return true;
}
/**
* Frees and clears the stream.
*/
static void
pulse_output_delete_stream(struct pulse_output *po)
{
assert(po != NULL);
assert(po->stream != NULL);
#if PA_CHECK_VERSION(0,9,8)
pa_stream_set_suspended_callback(po->stream, NULL, NULL);
#endif
pa_stream_set_state_callback(po->stream, NULL, NULL);
pa_stream_set_write_callback(po->stream, NULL, NULL);
pa_stream_disconnect(po->stream);
pa_stream_unref(po->stream);
po->stream = NULL;
}
/**
* Frees and clears the context.
*/
static void
pulse_output_delete_context(struct pulse_output *po)
{
assert(po != NULL);
assert(po->context != NULL);
pa_context_set_state_callback(po->context, NULL, NULL);
pa_context_set_subscribe_callback(po->context, NULL, NULL);
pa_context_disconnect(po->context);
pa_context_unref(po->context);
po->context = NULL;
}
/**
* 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)
{
assert(po != NULL);
assert(po->mainloop != NULL);
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);
pa_context_set_subscribe_callback(po->context,
pulse_output_subscribe_cb, po);
if (!pulse_output_connect(po, error_r)) {
pulse_output_delete_context(po);
return false;
}
return true;
}
static void *
pulse_output_init(G_GNUC_UNUSED const struct audio_format *audio_format,
const struct config_param *param,
G_GNUC_UNUSED GError **error_r)
{
struct pulse_output *po;
g_setenv("PULSE_PROP_media.role", "music", true);
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);
po->mixer = NULL;
po->mainloop = NULL;
po->context = NULL;
po->stream = NULL;
return po;
}
static void
pulse_output_finish(void *data)
{
struct pulse_output *po = data;
g_free(po);
}
static bool
pulse_output_enable(void *data, GError **error_r)
{
struct pulse_output *po = data;
assert(po->mainloop == NULL);
assert(po->context == NULL);
/* 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 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 = NULL;
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);
po->mainloop = NULL;
return false;
}
pa_threaded_mainloop_unlock(po->mainloop);
return true;
}
static void
pulse_output_disable(void *data)
{
struct pulse_output *po = data;
assert(po->mainloop != NULL);
pa_threaded_mainloop_stop(po->mainloop);
if (po->context != NULL)
pulse_output_delete_context(po);
pa_threaded_mainloop_free(po->mainloop);
po->mainloop = NULL;
}
/**
* 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)
{
assert(po->mainloop != NULL);
pa_context_state_t state;
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;
}
}
}
#if PA_CHECK_VERSION(0,9,8)
static void
pulse_output_stream_suspended_cb(G_GNUC_UNUSED pa_stream *stream, void *userdata)
{
struct pulse_output *po = userdata;
assert(stream == po->stream || po->stream == NULL);
assert(po->mainloop != NULL);
/* wake up the main loop to break out of the loop in
pulse_output_play() */
pa_threaded_mainloop_signal(po->mainloop, 0);
}
#endif
static void
pulse_output_stream_state_cb(pa_stream *stream, void *userdata)
{
struct pulse_output *po = userdata;
assert(stream == po->stream || po->stream == NULL);
assert(po->mainloop != NULL);
assert(po->context != NULL);
switch (pa_stream_get_state(stream)) {
case PA_STREAM_READY:
if (po->mixer != NULL)
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 != NULL)
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(G_GNUC_UNUSED pa_stream *stream, size_t nbytes,
void *userdata)
{
struct pulse_output *po = userdata;
assert(po->mainloop != NULL);
po->writable = nbytes;
pa_threaded_mainloop_signal(po->mainloop, 0);
}
static bool
pulse_output_open(void *data, struct audio_format *audio_format,
GError **error_r)
{
struct pulse_output *po = data;
pa_sample_spec ss;
int error;
assert(po->mainloop != NULL);
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->format = SAMPLE_FORMAT_S16;
ss.format = PA_SAMPLE_S16NE;
ss.rate = audio_format->sample_rate;
ss.channels = audio_format->channels;
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;
}
#if PA_CHECK_VERSION(0,9,8)
pa_stream_set_suspended_callback(po->stream,
pulse_output_stream_suspended_cb, po);
#endif
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) {
pulse_output_delete_stream(po);
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_output_close(void *data)
{
struct pulse_output *po = data;
pa_operation *o;
assert(po->mainloop != NULL);
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);
}
pulse_output_delete_stream(po);
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);
assert(po->mainloop != NULL);
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;
}
/**
* 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)
{
assert(po->stream != NULL);
#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->mainloop != NULL);
assert(po->context != NULL);
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_output_play(void *data, const void *chunk, size_t size, GError **error_r)
{
struct pulse_output *po = data;
int error;
assert(po->mainloop != NULL);
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)) {
pa_threaded_mainloop_unlock(po->mainloop);
return 0;
}
/* wait until the server allows us to write */
while (po->writable == 0) {
#if PA_CHECK_VERSION(0,9,8)
if (pa_stream_is_suspended(po->stream)) {
pa_threaded_mainloop_unlock(po->mainloop);
g_set_error(error_r, pulse_output_quark(), 0,
"suspended");
return 0;
}
#endif
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 0;
}
}
/* 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;
}
return size;
}
static void
pulse_output_cancel(void *data)
{
struct pulse_output *po = data;
pa_operation *o;
assert(po->mainloop != NULL);
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->mainloop != 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_output_test_default_device,
.init = pulse_output_init,
.finish = pulse_output_finish,
.enable = pulse_output_enable,
.disable = pulse_output_disable,
.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,
};