Added a wrapping library for pulseaudio

The wrapping library uses a global mainloop and context which
allows operations to share the connection. The global mainloop
is created and destroyed based on internal reference counting.

The capture code won't spawn a new thread for each input anymore
but instead just create the recording stream and rely on the
threaded mainloop to execute the read callback when data is available.

plugins/linux-pulseaudio/CMakeLists.txt

@@ -10,6 +10,7 @@ include_directories(SYSTEM "${CMAKE_SOURCE_DIR}/libobs")
 
 set(linux-pulseaudio_SOURCES
 	linux-pulseaudio.c
+	pulse-wrapper.c
 	pulse-input.c
 )
 

plugins/linux-pulseaudio/pulse-input.c

@@ -16,33 +16,15 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
 #include <util/bmem.h>
-#include <util/threading.h>
-#include <util/platform.h>
-
-#include <pulse/thread-mainloop.h>
-#include <pulse/mainloop.h>
-#include <pulse/context.h>
-#include <pulse/introspect.h>
-#include <pulse/stream.h>
-#include <pulse/error.h>
-
 #include <obs.h>
 
-#define PULSE_DATA(voidptr) struct pulse_data *data = voidptr;
+#include "pulse-wrapper.h"
 
-/*
- * delay in usecs before starting to record, this eliminates problems with
- * pulse audio sending weird data/timestamps when the stream is connected
- *
- * for more information see:
- * github.com/MaartenBaert/ssr/blob/master/src/AV/Input/PulseAudioInput.cpp
- */
-const uint64_t pulse_start_delay = 100000;
+#define PULSE_DATA(voidptr) struct pulse_data *data = voidptr;
 
 struct pulse_data {
-	pthread_t thread;
-	os_event_t event;
 	obs_source_t source;
+	char *device;
 
 	enum speaker_layout speakers;
 	pa_sample_format_t format;
@@ -51,21 +33,7 @@ struct pulse_data {
 
 	uint_fast32_t bytes_per_frame;
 
-	pa_mainloop *mainloop;
-	pa_context *context;
 	pa_stream *stream;
-	pa_proplist *props;
-};
-
-struct pulse_context_change {
-	pa_threaded_mainloop *mainloop;
-	pa_context_state_t state;
-};
-
-struct pulse_enumerate {
-	pa_threaded_mainloop *mainloop;
-	obs_property_t devices;
-	bool input;
 };
 
 /*
@@ -90,14 +58,6 @@ static enum audio_format pulse_to_obs_audio_format(
 	return AUDIO_FORMAT_UNKNOWN;
 }
 
-/*
- * get the number of frames from bytes and current format
- */
-static uint_fast32_t frames_to_bytes(struct pulse_data *data, size_t bytes)
-{
-	return (bytes / data->bytes_per_frame);
-}
-
 /*
  * get the buffer size needed for length msec with current settings
  */
@@ -122,136 +82,92 @@ static int pulse_get_stream_latency(pa_stream *stream, int64_t *latency)
 }
 
 /*
- * Iterate the mainloop
- *
- * The custom implementation gives better performance than the function
- * provided by pulse audio, maybe due to the timeout set in prepare ?
- */
-static void pulse_iterate(struct pulse_data *data)
-{
-	if (pa_mainloop_prepare(data->mainloop, 1000) < 0) {
-		blog(LOG_ERROR, "Unable to prepare main loop");
-		return;
-	}
-	if (pa_mainloop_poll(data->mainloop) < 0) {
-		blog(LOG_ERROR, "Unable to poll main loop");
-		return;
-	}
-	if (pa_mainloop_dispatch(data->mainloop) < 0)
-		blog(LOG_ERROR, "Unable to dispatch main loop");
-}
-
-/*
- * Server info callback, this is called from pa_mainloop_dispatch
- * TODO: how to free the server info struct ?
+ * Callback for pulse which gets executed when new audio data is available
  */
-static void pulse_get_server_info_cb(pa_context *c, const pa_server_info *i,
-	void *userdata)
+static void pulse_stream_read(pa_stream *p, size_t nbytes, void *userdata)
 {
-	UNUSED_PARAMETER(c);
+	UNUSED_PARAMETER(p);
+	UNUSED_PARAMETER(nbytes);
 	PULSE_DATA(userdata);
 
-	const pa_sample_spec *spec = &i->sample_spec;
-	data->format = spec->format;
-	data->samples_per_sec = spec->rate;
-
-	blog(LOG_DEBUG, "pulse-input: Default format: %s, %u Hz, %u channels",
-		pa_sample_format_to_string(spec->format),
-		spec->rate,
-		spec->channels);
-}
+	const void *frames;
+	size_t bytes;
+	uint64_t pa_time;
+	int64_t pa_latency;
 
-/*
- * Request pulse audio server info
- * TODO: handle failures ?
- */
-static int pulse_get_server_info(struct pulse_data *data)
-{
-	pa_server_info_cb_t cb = pulse_get_server_info_cb;
-	pa_operation *op = pa_context_get_server_info(data->context, cb, data);
-
-	for(;;) {
-		pulse_iterate(data);
-		pa_operation_state_t state = pa_operation_get_state(op);
-		if (state == PA_OPERATION_DONE) {
-			pa_operation_unref(op);
-			break;
-		}
-	}
+	pa_stream_peek(data->stream, &frames, &bytes);
 
-	return 0;
-}
+	// check if we got data
+	if (!bytes)
+		goto exit;
 
-/*
- * Create a new pulse audio main loop and connect to the server
- *
- * Returns a negative value on error
- */
-static int pulse_connect(struct pulse_data *data)
-{
-	data->mainloop = pa_mainloop_new();
-	if (!data->mainloop) {
-		blog(LOG_ERROR, "pulse-input: Unable to create main loop");
-		return -1;
+	if (!frames) {
+		blog(LOG_DEBUG,
+			"pulse-input: Got audio hole of %u bytes",
+			(unsigned int) bytes);
+		pa_stream_drop(data->stream);
+		goto exit;
 	}
 
-	data->context = pa_context_new_with_proplist(
-		pa_mainloop_get_api(data->mainloop), "OBS Studio", data->props);
-	if (!data->context) {
-		blog(LOG_ERROR, "pulse-input: Unable to create context");
-		return -1;
+	if (pa_stream_get_time(data->stream, &pa_time) < 0) {
+		blog(LOG_ERROR,
+			"pulse-input: Failed to get timing info !");
+		pa_stream_drop(data->stream);
+		goto exit;
 	}
 
-	int status = pa_context_connect(
-		data->context, NULL, PA_CONTEXT_NOAUTOSPAWN, NULL);
-	if (status < 0) {
-		blog(LOG_ERROR, "pulse-input: Unable to connect! Status: %d",
-		     status);
-		return -1;
-	}
+	pulse_get_stream_latency(data->stream, &pa_latency);
 
-	// wait until connected
-	for (;;) {
-		pulse_iterate(data);
-		pa_context_state_t state = pa_context_get_state(data->context);
-		if (state == PA_CONTEXT_READY) {
-			blog(LOG_DEBUG, "pulse-input: Context ready");
-			break;
-		}
-		if (!PA_CONTEXT_IS_GOOD(state)) {
-			blog(LOG_ERROR, "pulse-input: Context connect failed");
-			return -1;
-		}
-	}
+	struct source_audio out;
+	out.speakers        = data->speakers;
+	out.samples_per_sec = data->samples_per_sec;
+	out.format          = pulse_to_obs_audio_format(data->format);
+	out.data[0]         = (uint8_t *) frames;
+	out.frames          = bytes / data->bytes_per_frame;
+	out.timestamp       = (pa_time - pa_latency) * 1000;
+	obs_source_output_audio(data->source, &out);
 
-	return 0;
+	pa_stream_drop(data->stream);
+
+exit:
+	pulse_signal(0);
 }
 
 /*
- * Disconnect from the pulse audio server and destroy the main loop
+ * Server info callback
  */
-static void pulse_disconnect(struct pulse_data *data)
+static void pulse_server_info(pa_context *c, const pa_server_info *i,
+	void *userdata)
 {
-	if (data->context) {
-		pa_context_disconnect(data->context);
-		pa_context_unref(data->context);
-	}
+	UNUSED_PARAMETER(c);
+	PULSE_DATA(userdata);
+
+	data->format          = i->sample_spec.format;
+	data->samples_per_sec = i->sample_spec.rate;
+	data->channels        = i->sample_spec.channels;
 
-	if (data->mainloop)
-		pa_mainloop_free(data->mainloop);
+	blog(LOG_DEBUG, "pulse-input: Default format: %s, %u Hz, %u channels",
+		pa_sample_format_to_string(i->sample_spec.format),
+		i->sample_spec.rate,
+		i->sample_spec.channels);
+
+	pulse_signal(0);
 }
 
 /*
- * Create a new pulse audio stream and connect to it
- *
- * Return a negative value on error
+ * start recording
  */
-static int pulse_connect_stream(struct pulse_data *data)
+static int_fast32_t pulse_start_recording(struct pulse_data *data)
 {
+	if (pulse_get_server_info(pulse_server_info, (void *) data) < 0) {
+		blog(LOG_ERROR, "pulse-input: Unable to get server info !");
+		return -1;
+	}
+
 	pa_sample_spec spec;
-	spec.format = data->format;
-	spec.rate = data->samples_per_sec;
-	spec.channels = get_audio_channels(data->speakers);
+	spec.format   = data->format;
+	spec.rate     = data->samples_per_sec;
+	spec.channels = data->channels;
 
 	if (!pa_sample_spec_valid(&spec)) {
 		blog(LOG_ERROR, "pulse-input: Sample spec is not valid");
@@ -262,290 +178,88 @@ static int pulse_connect_stream(struct pulse_data *data)
 	blog(LOG_DEBUG, "pulse-input: %u bytes per frame",
 	     (unsigned int) data->bytes_per_frame);
 
-	pa_buffer_attr attr;
-	attr.fragsize = get_buffer_size(data, 250);
-	attr.maxlength = (uint32_t) -1;
-	attr.minreq = (uint32_t) -1;
-	attr.prebuf = (uint32_t) -1;
-	attr.tlength = (uint32_t) -1;
-
-	data->stream = pa_stream_new_with_proplist(data->context,
-		obs_source_getname(data->source), &spec, NULL, data->props);
+	data->stream = pulse_stream_new(obs_source_getname(data->source),
+		&spec, NULL);
 	if (!data->stream) {
 		blog(LOG_ERROR, "pulse-input: Unable to create stream");
 		return -1;
 	}
+
+	pulse_lock();
+	pa_stream_set_read_callback(data->stream, pulse_stream_read,
+		(void *) data);
+	pulse_unlock();
+
+	pa_buffer_attr attr;
+	attr.fragsize  = get_buffer_size(data, 250);
+	attr.maxlength = (uint32_t) -1;
+	attr.minreq    = (uint32_t) -1;
+	attr.prebuf    = (uint32_t) -1;
+	attr.tlength   = (uint32_t) -1;
+
 	pa_stream_flags_t flags =
 		PA_STREAM_INTERPOLATE_TIMING
 		| PA_STREAM_AUTO_TIMING_UPDATE
 		| PA_STREAM_ADJUST_LATENCY;
-	if (pa_stream_connect_record(data->stream, NULL, &attr, flags) < 0) {
+
+	pulse_lock();
+	int_fast32_t ret = pa_stream_connect_record(data->stream, data->device,
+		&attr, flags);
+	pulse_unlock();
+	if (ret < 0) {
 		blog(LOG_ERROR, "pulse-input: Unable to connect to stream");
 		return -1;
 	}
 
-	for (;;) {
-		pulse_iterate(data);
-		pa_stream_state_t state = pa_stream_get_state(data->stream);
-		if (state == PA_STREAM_READY) {
-			blog(LOG_DEBUG, "pulse-input: Stream ready");
-			break;
-		}
-		if (!PA_STREAM_IS_GOOD(state)) {
-			blog(LOG_ERROR, "pulse-input: Stream connect failed");
-			return -1;
-		}
-	}
-
+	blog(LOG_DEBUG, "pulse-input: Recording started");
 	return 0;
 }
 
 /*
- * Disconnect from the pulse audio stream
+ * stop recording
  */
-static void pulse_diconnect_stream(struct pulse_data *data)
+static void pulse_stop_recording(struct pulse_data *data)
 {
 	if (data->stream) {
+		pulse_lock();
 		pa_stream_disconnect(data->stream);
 		pa_stream_unref(data->stream);
+		pulse_unlock();
 	}
 }
 
 /*
- * Loop to skip the first few samples of a stream
- */
-static int pulse_skip(struct pulse_data *data)
-{
-	uint64_t skip = 1;
-	const void *frames;
-	size_t bytes;
-	uint64_t pa_time;
-
-	while (os_event_try(data->event) == EAGAIN) {
-		pulse_iterate(data);
-		pa_stream_peek(data->stream, &frames, &bytes);
-
-		if (!bytes)
-			continue;
-		if (!frames || pa_stream_get_time(data->stream, &pa_time) < 0) {
-			pa_stream_drop(data->stream);
-			continue;
-		}
-
-		if (skip == 1 && pa_time)
-			skip = pa_time;
-		if (skip + pulse_start_delay < pa_time)
-			return 0;
-
-		pa_stream_drop(data->stream);
-	}
-
-	return -1;
-}
-
-/*
- * Worker thread to get audio data
- *
- * Will run until signaled
- */
-static void *pulse_thread(void *vptr)
-{
-	PULSE_DATA(vptr);
-
-	if (pulse_connect(data) < 0)
-		return NULL;
-	if (pulse_get_server_info(data) < 0)
-		return NULL;
-	if (pulse_connect_stream(data) < 0)
-		return NULL;
-
-	if (pulse_skip(data) < 0)
-		return NULL;
-
-	blog(LOG_DEBUG, "pulse-input: Start recording");
-
-	const void *frames;
-	size_t bytes;
-	uint64_t pa_time;
-	int64_t pa_latency;
-
-	struct source_audio out;
-	out.speakers = data->speakers;
-	out.samples_per_sec = data->samples_per_sec;
-	out.format = pulse_to_obs_audio_format(data->format);
-
-	while (os_event_try(data->event) == EAGAIN) {
-		pulse_iterate(data);
-
-		pa_stream_peek(data->stream, &frames, &bytes);
-
-		// check if we got data
-		if (!bytes)
-			continue;
-		if (!frames) {
-			blog(LOG_DEBUG,
-				"pulse-input: Got audio hole of %u bytes",
-				(unsigned int) bytes);
-			pa_stream_drop(data->stream);
-			continue;
-		}
-
-		if (pa_stream_get_time(data->stream, &pa_time) < 0) {
-			blog(LOG_ERROR,
-				"pulse-input: Failed to get timing info !");
-			pa_stream_drop(data->stream);
-			continue;
-		}
-
-		pulse_get_stream_latency(data->stream, &pa_latency);
-
-		out.data[0] = (uint8_t *) frames;
-		out.frames = frames_to_bytes(data, bytes);
-		out.timestamp = (pa_time - pa_latency) * 1000;
-		obs_source_output_audio(data->source, &out);
-
-		pa_stream_drop(data->stream);
-	}
-
-	pulse_diconnect_stream(data);
-	pulse_disconnect(data);
-
-	return NULL;
-}
-
-/*
- * Create a new pulseaudio context
+ * input info callback
  */
-static pa_context *pulse_context_create(pa_threaded_mainloop *m)
+static void pulse_input_info(pa_context *c, const pa_source_info *i, int eol,
+	void *userdata)
 {
-	pa_context *c;
-	pa_proplist *p;
-
-	p = pa_proplist_new();
-	pa_proplist_sets(p, PA_PROP_APPLICATION_NAME, "OBS Studio");
-	pa_proplist_sets(p, PA_PROP_APPLICATION_ICON_NAME, "application-exit");
-	pa_proplist_sets(p, PA_PROP_MEDIA_ROLE, "production");
-
-	pa_threaded_mainloop_lock(m);
-	c = pa_context_new_with_proplist(pa_threaded_mainloop_get_api(m),
-		"OBS Studio", p);
-	pa_threaded_mainloop_unlock(m);
+	UNUSED_PARAMETER(c);
+	if (eol != 0 || i->monitor_of_sink != PA_INVALID_INDEX)
+		goto skip;
 
-	pa_proplist_free(p);
+	obs_property_list_add_string((obs_property_t) userdata,
+		i->description, i->name);
 
-	return c;
+skip:
+	pulse_signal(0);
 }
 
 /**
- * Context state callback
- */
-static void pulse_context_state_changed(pa_context *c, void *userdata)
-{
-	struct pulse_context_change *ctx =
-		(struct pulse_context_change *) userdata;
-	ctx->state = pa_context_get_state(c);
-
-	pa_threaded_mainloop_signal(ctx->mainloop, 0);
-}
-
-/*
- * Connect context
+ * output info callback
  */
-static int pulse_context_connect(pa_threaded_mainloop *m, pa_context *c)
-{
-	int status = 0;
-	struct pulse_context_change ctx;
-	ctx.mainloop = m;
-	ctx.state = PA_CONTEXT_UNCONNECTED;
-
-	pa_threaded_mainloop_lock(m);
-	pa_context_set_state_callback(c, pulse_context_state_changed,
-		(void *) &ctx);
-
-	status = pa_context_connect(c, NULL, PA_CONTEXT_NOAUTOSPAWN, NULL);
-	if (status < 0) {
-		blog(LOG_ERROR, "pulse-input: Unable to connect! Status: %d",
-		     status);
-	}
-	else {
-		for (;;) {
-			if (ctx.state == PA_CONTEXT_READY) {
-				blog(LOG_DEBUG, "pulse-input: Context Ready");
-				break;
-			}
-			if (!PA_CONTEXT_IS_GOOD(ctx.state)) {
-				blog(LOG_ERROR,
-				     "pulse-input: Context connect failed !");
-				status = -1;
-				break;
-			}
-			pa_threaded_mainloop_wait(m);
-		}
-	}
-
-	pa_threaded_mainloop_unlock(m);
-	return status;
-}
-
-/*
- * Source properties callback
- */
-static void pulse_source_info(pa_context *c, const pa_source_info *i, int eol,
+static void pulse_output_info(pa_context *c, const pa_source_info *i, int eol,
 	void *userdata)
 {
 	UNUSED_PARAMETER(c);
+	if (eol != 0 || i->monitor_of_sink == PA_INVALID_INDEX)
+		goto skip;
 
-	if (eol != 0)
-		return;
-
-	struct pulse_enumerate *e = (struct pulse_enumerate *) userdata;
-
-	if ((e->input) ^ (i->monitor_of_sink == PA_INVALID_INDEX))
-		return;
-
-	blog(LOG_DEBUG, "pulse-input: Got source #%u '%s'",
-	     i->index, i->description);
-
-	obs_property_list_add_string(e->devices, i->description, i->name);
+	obs_property_list_add_string((obs_property_t) userdata,
+		i->description, i->name);
 
-	pa_threaded_mainloop_signal(e->mainloop, 0);
-}
-
-/*
- * enumerate input/output devices
- */
-static void pulse_enumerate_devices(obs_properties_t props, bool input)
-{
-	pa_context *c;
-	pa_operation *op;
-	pa_threaded_mainloop *m = pa_threaded_mainloop_new();
-	struct pulse_enumerate e;
-
-	e.mainloop = m;
-	e.devices = obs_properties_add_list(props, "device_id", "Device",
-		OBS_COMBO_TYPE_LIST, OBS_COMBO_FORMAT_STRING);
-	e.input = input;
-
-	pa_threaded_mainloop_start(m);
-	c = pulse_context_create(m);
-
-	if (pulse_context_connect(m, c) < 0)
-		goto fail;
-
-	pa_threaded_mainloop_lock(m);
-
-	op = pa_context_get_source_info_list(c, pulse_source_info, (void *) &e);
-	while (pa_operation_get_state(op) == PA_OPERATION_RUNNING)
-		pa_threaded_mainloop_wait(m);
-	pa_operation_unref(op);
-
-	pa_threaded_mainloop_unlock(m);
-
-	pa_context_disconnect(c);
-fail:
-	pa_context_unref(c);
-	pa_threaded_mainloop_stop(m);
-	pa_threaded_mainloop_free(m);
+skip:
+	pulse_signal(0);
 }
 
 /*
@@ -553,11 +267,14 @@ fail:
  */
 static obs_properties_t pulse_properties(const char *locale, bool input)
 {
-	blog(LOG_DEBUG, "pulse-input: properties requested !");
-
 	obs_properties_t props = obs_properties_create(locale);
+	obs_property_t devices = obs_properties_add_list(props, "device_id",
+		"Device", OBS_COMBO_TYPE_LIST, OBS_COMBO_FORMAT_STRING);
 
-	pulse_enumerate_devices(props, input);
+	pulse_init();
+	pa_source_info_cb_t cb = (input) ? pulse_input_info : pulse_output_info;
+	pulse_get_source_info_list(cb, (void *) devices);
+	pulse_unref();
 
 	return props;
 }
@@ -605,19 +322,14 @@ static void pulse_destroy(void *vptr)
 	if (!data)
 		return;
 
-	if (data->thread) {
-		void *ret;
-		os_event_signal(data->event);
-		pthread_join(data->thread, &ret);
-	}
-
-	os_event_destroy(data->event);
+	pulse_stop_recording(data);
+	pulse_unref();
 
-	pa_proplist_free(data->props);
+	if (data->device)
+		bfree(data->device);
+	bfree(data);
 
 	blog(LOG_DEBUG, "pulse-input: Input destroyed");
-
-	bfree(data);
 }
 
 /*
@@ -625,38 +337,28 @@ static void pulse_destroy(void *vptr)
  */
 static void *pulse_create(obs_data_t settings, obs_source_t source)
 {
-	UNUSED_PARAMETER(settings);
-
-	struct pulse_data *data = bmalloc(sizeof(struct pulse_data));
-	memset(data, 0, sizeof(struct pulse_data));
+	struct pulse_data *data = bzalloc(sizeof(struct pulse_data));
 
-	data->source = source;
+	data->source   = source;
 	data->speakers = SPEAKERS_STEREO;
+	data->device   = bstrdup(obs_data_getstring(settings, "device_id"));
 
-	blog(LOG_DEBUG, "pulse-input: obs wants '%s'",
-		obs_data_getstring(settings, "device_id"));
-
-	/* TODO: use obs-studio icon */
-	data->props = pa_proplist_new();
-	pa_proplist_sets(data->props, PA_PROP_APPLICATION_NAME,
-		"OBS Studio");
-	pa_proplist_sets(data->props, PA_PROP_APPLICATION_ICON_NAME,
-		"application-exit");
-	pa_proplist_sets(data->props, PA_PROP_MEDIA_ROLE,
-		"production");
-
-	if (os_event_init(&data->event, OS_EVENT_TYPE_MANUAL) != 0)
-		goto fail;
-	if (pthread_create(&data->thread, NULL, pulse_thread, data) != 0)
+	pulse_init();
+	if (pulse_start_recording(data) < 0)
 		goto fail;
 
 	return data;
-
 fail:
 	pulse_destroy(data);
 	return NULL;
 }
 
+static void pulse_update(void *vptr, obs_data_t settings)
+{
+	UNUSED_PARAMETER(vptr);
+	UNUSED_PARAMETER(settings);
+}
+
 struct obs_source_info pulse_input_capture = {
 	.id           = "pulse_input_capture",
 	.type         = OBS_SOURCE_TYPE_INPUT,
@@ -664,6 +366,7 @@ struct obs_source_info pulse_input_capture = {
 	.getname      = pulse_input_getname,
 	.create       = pulse_create,
 	.destroy      = pulse_destroy,
+	.update       = pulse_update,
 	.defaults     = pulse_defaults,
 	.properties   = pulse_input_properties
 };
@@ -675,6 +378,7 @@ struct obs_source_info pulse_output_capture = {
 	.getname      = pulse_output_getname,
 	.create       = pulse_create,
 	.destroy      = pulse_destroy,
+	.update       = pulse_update,
 	.defaults     = pulse_defaults,
 	.properties   = pulse_output_properties
 };

plugins/linux-pulseaudio/pulse-wrapper.c

@@ -0,0 +1,225 @@
+/*
+Copyright (C) 2014 by Leonhard Oelke <[email protected]>
+
+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, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <pthread.h>
+
+#include <pulse/thread-mainloop.h>
+
+#include <util/base.h>
+#include <obs.h>
+
+#include "pulse-wrapper.h"
+
+/* global data */
+static uint_fast32_t pulse_refs = 0;
+static pthread_mutex_t pulse_mutex = PTHREAD_MUTEX_INITIALIZER;
+static pa_threaded_mainloop *pulse_mainloop = NULL;
+static pa_context *pulse_context = NULL;
+
+/**
+ * context status change callback
+ *
+ * @todo this is currently a noop, we want to reconnect here if the connection
+ *       is lost ...
+ */
+static void pulse_context_state_changed(pa_context *c, void *userdata)
+{
+	UNUSED_PARAMETER(userdata);
+	UNUSED_PARAMETER(c);
+
+	blog(LOG_DEBUG, "pulse: context state changed");
+
+	pulse_signal(0);
+}
+
+/**
+ * get the default properties
+ */
+static pa_proplist *pulse_properties()
+{
+	pa_proplist *p = pa_proplist_new();
+
+	pa_proplist_sets(p, PA_PROP_APPLICATION_NAME, "OBS Studio");
+	pa_proplist_sets(p, PA_PROP_APPLICATION_ICON_NAME, "application-exit");
+	pa_proplist_sets(p, PA_PROP_MEDIA_ROLE, "production");
+
+	return p;
+}
+
+/**
+ * Initialize the pulse audio context with properties and callback
+ */
+static void pulse_init_context()
+{
+	pulse_lock();
+
+	pa_proplist *p = pulse_properties();
+	pulse_context = pa_context_new_with_proplist(
+		pa_threaded_mainloop_get_api(pulse_mainloop), "OBS Studio", p);
+
+	pa_context_set_state_callback(pulse_context,
+		pulse_context_state_changed, NULL);
+
+	pa_context_connect(pulse_context, NULL, PA_CONTEXT_NOAUTOSPAWN, NULL);
+	pa_proplist_free(p);
+
+	pulse_unlock();
+}
+
+/**
+ * wait for context to be ready
+ */
+static int_fast32_t pulse_context_ready()
+{
+	pulse_lock();
+	
+	if (!PA_CONTEXT_IS_GOOD(pa_context_get_state(pulse_context))) {
+		pulse_unlock();
+		return -1;
+	}
+
+	while (pa_context_get_state(pulse_context) != PA_CONTEXT_READY)
+		pulse_wait();
+
+	pulse_unlock();
+	return 0;
+}
+
+int_fast32_t pulse_init()
+{
+	pthread_mutex_lock(&pulse_mutex);
+
+	if (pulse_refs == 0) {
+		pulse_mainloop = pa_threaded_mainloop_new();
+		pa_threaded_mainloop_start(pulse_mainloop);
+
+		pulse_init_context();
+	}
+
+	pulse_refs++;
+	blog(LOG_DEBUG, "pulse: Reference count increased to %"PRIuFAST32,
+	     pulse_refs);
+
+	pthread_mutex_unlock(&pulse_mutex);
+
+	return 0;
+}
+
+void pulse_unref()
+{
+	pthread_mutex_lock(&pulse_mutex);
+
+	pulse_refs--;
+	blog(LOG_DEBUG, "pulse: Reference count decreased to %"PRIuFAST32,
+	     pulse_refs);
+
+	if (pulse_refs == 0) {
+		pulse_lock();
+		if (pulse_context != NULL) {
+			pa_context_disconnect(pulse_context);
+			pa_context_unref(pulse_context);
+			pulse_context = NULL;
+		}
+		pulse_unlock();
+
+		if (pulse_mainloop != NULL) {
+			pa_threaded_mainloop_stop(pulse_mainloop);
+			pa_threaded_mainloop_free(pulse_mainloop);
+			pulse_mainloop = NULL;
+		}
+	}
+
+	pthread_mutex_unlock(&pulse_mutex);
+}
+
+void pulse_lock()
+{
+	pa_threaded_mainloop_lock(pulse_mainloop);
+}
+
+void pulse_unlock()
+{
+	pa_threaded_mainloop_unlock(pulse_mainloop);
+}
+
+void pulse_wait()
+{
+	pa_threaded_mainloop_wait(pulse_mainloop);
+}
+
+void pulse_signal(int wait_for_accept)
+{
+	pa_threaded_mainloop_signal(pulse_mainloop, wait_for_accept);
+}
+
+void pulse_accept()
+{
+	pa_threaded_mainloop_accept(pulse_mainloop);
+}
+
+int_fast32_t pulse_get_source_info_list(pa_source_info_cb_t cb, void* userdata)
+{
+	if (pulse_context_ready() < 0)
+		return -1;
+
+	pulse_lock();
+
+	pa_operation *op = pa_context_get_source_info_list(
+		pulse_context, cb, userdata);
+	while (pa_operation_get_state(op) == PA_OPERATION_RUNNING)
+		pulse_wait();
+	pa_operation_unref(op);
+
+	pulse_unlock();
+
+	return 0;
+}
+
+int_fast32_t pulse_get_server_info(pa_server_info_cb_t cb, void* userdata)
+{
+	if (pulse_context_ready() < 0)
+		return -1;
+
+	pulse_lock();
+
+	pa_operation *op = pa_context_get_server_info(
+		pulse_context, cb, userdata);
+	while (pa_operation_get_state(op) == PA_OPERATION_RUNNING)
+		pulse_wait();
+	pa_operation_unref(op);
+
+	pulse_unlock();
+	return 0;
+}
+
+pa_stream* pulse_stream_new(const char* name, const pa_sample_spec* ss,
+	const pa_channel_map* map)
+{
+	if (pulse_context_ready() < 0)
+		return NULL;
+
+	pulse_lock();
+
+	pa_proplist *p = pulse_properties();
+	pa_stream *s = pa_stream_new_with_proplist(
+		pulse_context, name, ss, map, p);
+	pa_proplist_free(p);
+
+	pulse_unlock();
+	return s;
+}
+

plugins/linux-pulseaudio/pulse-wrapper.h

@@ -0,0 +1,118 @@
+/*
+Copyright (C) 2014 by Leonhard Oelke <[email protected]>
+
+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, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <inttypes.h>
+#include <pulse/stream.h>
+#include <pulse/context.h>
+#include <pulse/introspect.h>
+
+#pragma once
+
+/**
+ * Initialize the pulseaudio mainloop and increase the reference count
+ */
+int_fast32_t pulse_init();
+
+/**
+ * Unreference the pulseaudio mainloop, when the reference count reaches
+ * zero the mainloop will automatically be destroyed
+ */
+void pulse_unref();
+
+/**
+ * Lock the mainloop
+ *
+ * In order to allow for multiple threads to use the same mainloop pulseaudio
+ * provides it's own locking mechanism. This function should be called before
+ * using any pulseaudio function that is in any way related to the mainloop or
+ * context.
+ *
+ * @note use of this function may cause deadlocks
+ *
+ * @warning do not use with pulse_ wrapper functions
+ */
+void pulse_lock();
+
+/**
+ * Unlock the mainloop
+ *
+ * @see pulse_lock()
+ */
+void pulse_unlock();
+
+/**
+ * Wait for events to happen
+ *
+ * This function should be called when waiting for an event to happen.
+ */
+void pulse_wait();
+
+/**
+ * Wait for accept signal from calling thread
+ *
+ * This function tells the pulseaudio mainloop wheter the data provided to
+ * the callback should be retained until the calling thread executes
+ * pulse_accept()
+ *
+ * If wait_for_accept is 0 the function returns and the data is freed.
+ */
+void pulse_signal(int wait_for_accept);
+
+/**
+ * Signal the waiting callback to return
+ *
+ * This function is used in conjunction with pulse_signal()
+ */
+void pulse_accept();
+
+/**
+ * Request source information
+ *
+ * The function will block until the operation was executed and the mainloop
+ * called the provided callback functions.
+ *
+ * @return negative on error
+ *
+ * @note The function will block until the server context is ready.
+ *
+ * @warning call without active locks
+ */
+int_fast32_t pulse_get_source_info_list(pa_source_info_cb_t cb, void *userdata);
+
+/**
+ * Request server information
+ *
+ * The function will block until the operation was executed and the mainloop
+ * called the provided callback functions
+ *
+ * @return negative on error
+ *
+ * @note The function will block until the server context is ready.
+ *
+ * @warning call without active locks
+ */
+int_fast32_t pulse_get_server_info(pa_server_info_cb_t cb, void *userdata);
+
+/**
+ * Create a new stream with the default properties
+ *
+ * @note The function will block until the server context is ready.
+ *
+ * @warning call without active locks
+ */
+pa_stream *pulse_stream_new(const char *name, const pa_sample_spec *ss,
+	const pa_channel_map *map);