Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 95 additions & 2 deletions src/audio/pipewire.c
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@

#include <spa/param/audio/format-utils.h>
#include <spa/utils/ringbuffer.h>
#include <spa/utils/result.h>
#include <spa/param/props.h>

#include <spd_audio_plugin.h>
Expand Down Expand Up @@ -61,6 +62,33 @@ do { \
} while (0)
#define error(arg, ...) message(0, "pipewire: fatal: " arg, ##__VA_ARGS__)

// Convert between pipewire and speech-dispatcher log levels
static enum spa_log_level spd_to_pw_log_level(int level)
{
if (level >= 5)
return SPA_LOG_LEVEL_DEBUG;
if (level == 4)
return SPA_LOG_LEVEL_INFO;
if (level >= 2)
return SPA_LOG_LEVEL_WARN;
return SPA_LOG_LEVEL_ERROR;
}

static int pw_to_spd_log_level(enum spa_log_level level)
{
switch (level)
{
case SPA_LOG_LEVEL_ERROR:
return 1;
case SPA_LOG_LEVEL_WARN:
return 2;
case SPA_LOG_LEVEL_INFO:
return 4;
default:
return 5;
}
}

// state of the backend, where all components are gathered. This will be allocated on the heap
typedef struct
{
Expand All @@ -76,6 +104,47 @@ typedef struct
int32_t eventfd_number; // used in on_process to signal the thread where pipewire_play is running that the ringbuffer has been drained to the point where playback either finished or is in progress, but for sure to the point where we could push more audio, because our side of the buffer at least is drained
} module_state;

static void forward_logtv(void *object, enum spa_log_level level, const struct spa_log_topic *topic, const char *file, int line, const char *func, const char *fmt, va_list args)
{
char *text;
vasprintf(&text, fmt, args);
message(pw_to_spd_log_level(level), "%s:%d(%s): %s", file, line, func, text);
}

static void forward_logt(void *object, enum spa_log_level level, const struct spa_log_topic *topic, const char *file, int line, const char *func, const char *fmt, ...)
{
va_list args;
va_start(args, fmt);
forward_logtv(object, level, topic, file, line, func, fmt, args);
va_end(args);
}

static void forward_logv(void *object, enum spa_log_level level, const char *file, int line, const char *func, const char *fmt, va_list args)
{
forward_logtv(object, level, NULL, file, line, func, fmt, args);
}

static void forward_log(void *object, enum spa_log_level level, const char *file, int line, const char *func, const char *fmt, ...)
{
va_list args;
va_start(args, fmt);
forward_logtv(object, level, NULL, file, line, func, fmt, args);
va_end(args);
}

static const struct spa_log_methods forward_log_methods = {
SPA_VERSION_LOG_METHODS,
.log = forward_log,
.logv = forward_logv,
.logt = forward_logt,
.logtv = forward_logtv,
};

static struct spa_log forward_log_interface = {
.iface = SPA_INTERFACE_INIT(SPA_TYPE_INTERFACE_Log, SPA_VERSION_LOG, &forward_log_methods, NULL),
.level = SPA_LOG_LEVEL_ERROR,
};

// pipewire on_process callback
// this function would be called each time the server requests data from us
static void on_process(void *userdata)
Expand Down Expand Up @@ -162,10 +231,19 @@ static void on_process(void *userdata)
spa_system_eventfd_write(state->inner_loop->system, state->eventfd_number, 1);
}

static void on_state_changed(void *userdata, enum pw_stream_state old, enum pw_stream_state new_state, const char *error_message)
{
message(4, "stream state changed from %s to %s", pw_stream_state_as_string(old), pw_stream_state_as_string(new_state));
if (new_state == PW_STREAM_STATE_ERROR)
error("stream error: %s", error_message ? error_message : "unknown error");
else if (new_state == PW_STREAM_STATE_UNCONNECTED && old != PW_STREAM_STATE_UNCONNECTED)
message(3, "stream got disconnected from the pipewire daemon");
}

// pipewire internal: structure describing what kind of events we subscribe to
// For now, this is only on_process
static const struct pw_stream_events stream_events = {
PW_VERSION_STREAM_EVENTS,
.state_changed = on_state_changed,
.process = on_process,
};

Expand All @@ -175,6 +253,9 @@ static AudioID *pipewire_open(void **pars)
{
module_state *state = (module_state *)malloc(sizeof(module_state));
message(3, "initialising pipewire output");
// forward all pipewire messages to log
pw_log_set(&forward_log_interface);
pw_log_set_level(SPA_LOG_LEVEL_WARN);
pw_init(0, NULL);
// initialise the main loop, and then get the inner loop from it, so that locks are kept from the most often contended path as much as possible
state->loop = pw_thread_loop_new("pipewire audio thread", NULL);
Expand Down Expand Up @@ -268,7 +349,18 @@ static int pipewire_begin(AudioID *id, AudioTrack track)
params = spa_format_audio_raw_build(&b, SPA_PARAM_EnumFormat, &SPA_AUDIO_INFO_RAW_INIT(.format = format, .channels = track.num_channels, .rate = state->playback_sample_rate));
message(4, "connecting pipewire stream");
pw_thread_loop_lock(state->loop);
pw_stream_connect(state->stream, PW_DIRECTION_OUTPUT, PW_ID_ANY, PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS, &params, 1);
int res = pw_stream_connect(state->stream, PW_DIRECTION_OUTPUT, PW_ID_ANY, PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS, &params, 1);
if (res < 0)
{
const char *stream_error = NULL;
pw_stream_get_state(state->stream, &stream_error);
// the error string belongs to the stream, so print it before releasing the lock
error("cannot connect stream: %s (%s)", spa_strerror(res), stream_error ? stream_error : "no details from pipewire");
pw_thread_loop_unlock(state->loop);
// forget the sample rate, so that the next call tries to connect again
state->playback_sample_rate = 0;
return -1;
}

// The default quality 4 makes espeak-ng's output notably lame
int quality = 10;
Expand Down Expand Up @@ -434,6 +526,7 @@ static void pipewire_set_log_level(int level)
if (level != 0)
{
pipewire_log_level = level;
pw_log_set_level(spd_to_pw_log_level(pipewire_log_level));
}
}

Expand Down
Loading