mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-03 09:01:54 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

pulse-server: implement pipe sink/source with module

Browse source
This commit is contained in:
Wim Taymans 2022-06-03 11:51:29 +02:00
parent 87172fde06
commit 6276253c30
3 changed files with 119 additions and 425 deletions
Download patch file Download diff file Expand all files Collapse all files

8
src/modules/module-pipe-tunnel.c
View file

@ -393,8 +393,12 @@ static int create_fifo(struct impl *impl)
pw_log_error("'%s' is not a FIFO.", filename);
goto error;
}
pw_log_info("opened fifo '%s' for %s", filename,
impl->direction == PW_DIRECTION_INPUT ? "writing" : "reading");
pw_log_info("%s fifo '%s' with format:%s channels:%d rate:%d",
impl->direction == PW_DIRECTION_OUTPUT ? "reading from" : "writing to",
filename,
spa_debug_type_find_name(spa_type_audio_format, impl->info.format),
impl->info.channels, impl->info.rate);
impl->filename = strdup(filename);
impl->unlink_fifo = do_unlink_fifo;
impl->fd = fd;

238
src/modules/module-protocol-pulse/modules/module-pipe-sink.c
View file

@ -40,149 +40,78 @@
PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
#define PW_LOG_TOPIC_DEFAULT mod_topic
#define DEFAULT_FILE_NAME "/tmp/music.output"
struct module_pipesink_data {
struct module *module;
struct pw_core *core;
struct pw_stream *capture;
struct spa_hook core_listener;
struct spa_hook capture_listener;
struct spa_hook mod_listener;
struct pw_impl_module *mod;
struct pw_properties *capture_props;
struct spa_audio_info_raw info;
char *filename;
int fd;
bool do_unlink_fifo;
};
static void capture_process(void *data)
{
struct module_pipesink_data *impl = data;
struct pw_buffer *in;
struct spa_data *d;
uint32_t i, size, offset;
int written;
if ((in = pw_stream_dequeue_buffer(impl->capture)) == NULL) {
pw_log_warn("Out of capture buffers: %m");
return;
}
for (i = 0; i < in->buffer->n_datas; i++) {
d = &in->buffer->datas[i];
size = d->chunk->size;
offset = d->chunk->offset;
while (size > 0) {
written = write(impl->fd, SPA_MEMBER(d->data, offset, void), size);
if (written < 0) {
if (errno == EINTR) {
/* retry if interrupted */
continue;
} else if (errno == EAGAIN || errno == EWOULDBLOCK) {
/* Don't continue writing */
break;
} else {
pw_log_warn("Failed to write to pipe sink");
}
}
offset += written;
size -= written;
}
}
pw_stream_queue_buffer(impl->capture, in);
}
static void on_core_error(void *data, uint32_t id, int seq, int res, const char *message)
static void module_destroy(void *data)
{
struct module_pipesink_data *d = data;
pw_log_error("error id:%u seq:%d res:%d (%s): %s",
id, seq, res, spa_strerror(res), message);
if (id == PW_ID_CORE && res == -EPIPE)
spa_hook_remove(&d->mod_listener);
d->mod = NULL;
module_schedule_unload(d->module);
}
static const struct pw_core_events core_events = {
PW_VERSION_CORE_EVENTS,
.error = on_core_error,
};
static void on_stream_state_changed(void *data, enum pw_stream_state old,
enum pw_stream_state state, const char *error)
{
struct module_pipesink_data *d = data;
switch (state) {
case PW_STREAM_STATE_UNCONNECTED:
pw_log_info("stream disconnected, unloading");
module_schedule_unload(d->module);
break;
case PW_STREAM_STATE_ERROR:
pw_log_error("stream error: %s", error);
break;
default:
break;
}
}
static const struct pw_stream_events in_stream_events = {
PW_VERSION_STREAM_EVENTS,
.state_changed = on_stream_state_changed,
.process = capture_process
static const struct pw_impl_module_events module_events = {
PW_VERSION_IMPL_MODULE_EVENTS,
.destroy = module_destroy
};
static int module_pipe_sink_load(struct client *client, struct module *module)
{
struct module_pipesink_data *data = module->user_data;
int res;
uint32_t n_params;
const struct spa_pod *params[1];
uint8_t buffer[1024];
struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer));
FILE *f;
char *args;
size_t size;
uint32_t i;
data->core = pw_context_connect(module->impl->context,
pw_properties_copy(client->props),
0);
if (data->core == NULL)
pw_properties_setf(data->capture_props, "pulse.module.id",
"%u", module->index);
if ((f = open_memstream(&args, &size)) == NULL)
return -errno;
pw_core_add_listener(data->core,
&data->core_listener,
&core_events, data);
fprintf(f, "{");
fprintf(f, " \"tunnel.mode\" = \"sink\" ");
if (data->filename != NULL)
fprintf(f, " \"pipe.filename\": \"%s\"", data->filename);
if (data->info.format != 0)
fprintf(f, " \"audio.format\": \"%s\"", format_id2name(data->info.format));
if (data->info.rate != 0)
fprintf(f, " \"audio.rate\": %u,", data->info.rate);
if (data->info.channels != 0) {
fprintf(f, " \"audio.channels\": %u,", data->info.channels);
if (!(data->info.flags & SPA_AUDIO_FLAG_UNPOSITIONED)) {
fprintf(f, " \"audio.position\": [ ");
for (i = 0; i < data->info.channels; i++)
fprintf(f, "%s\"%s\"", i == 0 ? "" : ",",
channel_id2name(data->info.position[i]));
fprintf(f, " ],");
}
}
pw_properties_serialize_dict(f, &data->capture_props->dict, 0);
fprintf(f, " }");
fclose(f);
pw_properties_setf(data->capture_props, "pulse.module.id", "%u", module->index);
data->mod = pw_context_load_module(module->impl->context,
"libpipewire-module-pipe-tunnel",
args, NULL);
data->capture = pw_stream_new(data->core,
"pipesink capture", data->capture_props);
data->capture_props = NULL;
if (data->capture == NULL)
free(args);
if (data->mod == NULL)
return -errno;
pw_stream_add_listener(data->capture,
&data->capture_listener,
&in_stream_events, data);
n_params = 0;
params[n_params++] = spa_format_audio_raw_build(&b, SPA_PARAM_EnumFormat,
&data->info);
if ((res = pw_stream_connect(data->capture,
PW_DIRECTION_INPUT,
PW_ID_ANY,
PW_STREAM_FLAG_AUTOCONNECT |
PW_STREAM_FLAG_MAP_BUFFERS |
PW_STREAM_FLAG_RT_PROCESS,
params, n_params)) < 0)
return res;
pw_impl_module_add_listener(data->mod,
&data->mod_listener,
&module_events, data);
return 0;
}
@ -190,19 +119,13 @@ static int module_pipe_sink_unload(struct module *module)
{
struct module_pipesink_data *d = module->user_data;
pw_properties_free(d->capture_props);
if (d->capture != NULL)
pw_stream_destroy(d->capture);
if (d->core != NULL)
pw_core_disconnect(d->core);
if (d->filename) {
if (d->do_unlink_fifo)
unlink(d->filename);
free(d->filename);
if (d->mod) {
spa_hook_remove(&d->mod_listener);
pw_impl_module_destroy(d->mod);
d->mod = NULL;
}
if (d->fd >= 0)
close(d->fd);
pw_properties_free(d->capture_props);
free(d->filename);
return 0;
}
@ -211,6 +134,7 @@ static const struct spa_dict_item module_pipe_sink_info[] = {
{ PW_KEY_MODULE_DESCRIPTION, "Pipe sink" },
{ PW_KEY_MODULE_USAGE, "file=<name of the FIFO special file to use> "
"sink_name=<name for the sink> "
"sink_properties=<sink properties> "
"format=<sample format> "
"rate=<sample rate> "
"channels=<number of channels> "
@ -224,12 +148,9 @@ static int module_pipe_sink_prepare(struct module * const module)
struct pw_properties * const props = module->props;
struct pw_properties *capture_props = NULL;
struct spa_audio_info_raw info = { 0 };
struct stat st;
const char *str;
char *filename = NULL;
bool do_unlink_fifo;
int res = 0;
int fd = -1;
PW_LOG_TOPIC_INIT(mod_topic);
@ -244,81 +165,34 @@ static int module_pipe_sink_prepare(struct module * const module)
goto out;
}
info.format = SPA_AUDIO_FORMAT_S16;
if ((str = pw_properties_get(props, "format")) != NULL) {
info.format = format_paname2id(str, strlen(str));
pw_properties_set(props, "format", NULL);
}
if ((str = pw_properties_get(props, "sink_name")) != NULL) {
pw_properties_set(capture_props, PW_KEY_NODE_NAME, str);
pw_properties_set(props, "sink_name", NULL);
}
if ((str = pw_properties_get(props, "sink_properties")) != NULL)
module_args_add_props(capture_props, str);
if ((str = pw_properties_get(props, "file")) != NULL) {
filename = strdup(str);
pw_properties_set(props, "file", NULL);
} else {
filename = strdup(DEFAULT_FILE_NAME);
}
do_unlink_fifo = false;
if (mkfifo(filename, 0666) < 0) {
if (errno != EEXIST) {
res = -errno;
pw_log_error("mkfifo('%s'): %s", filename, spa_strerror(res));
goto out;
}
} else {
do_unlink_fifo = true;
/*
* Our umask is 077, so the pipe won't be created with the
* requested permissions. Let's fix the permissions with chmod().
*/
if (chmod(filename, 0666) < 0)
pw_log_warn("chmod('%s'): %s", filename, spa_strerror(-errno));
}
if ((fd = open(filename, O_RDWR | O_CLOEXEC | O_NONBLOCK, 0)) <= 0) {
res = -errno;
pw_log_error("open('%s'): %s", filename, spa_strerror(res));
goto out;
}
if (fstat(fd, &st) < 0) {
res = -errno;
pw_log_error("fstat('%s'): %s", filename, spa_strerror(res));
goto out;
}
if (!S_ISFIFO(st.st_mode)) {
pw_log_error("'%s' is not a FIFO.", filename);
goto out;
}
if (pw_properties_get(capture_props, PW_KEY_NODE_VIRTUAL) == NULL)
pw_properties_set(capture_props, PW_KEY_NODE_VIRTUAL, "true");
pw_properties_set(capture_props, PW_KEY_MEDIA_CLASS, "Audio/Sink");
d->module = module;
d->capture_props = capture_props;
d->info = info;
d->fd = fd;
d->filename = filename;
d->do_unlink_fifo = do_unlink_fifo;
pw_log_info("Successfully loaded module-pipe-sink");
return 0;
out:
pw_properties_free(capture_props);
if (filename) {
if (do_unlink_fifo)
unlink(filename);
free(filename);
}
if (fd >= 0)
close(fd);
return res;
}

292
src/modules/module-protocol-pulse/modules/module-pipe-source.c
View file

@ -40,171 +40,78 @@
PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
#define PW_LOG_TOPIC_DEFAULT mod_topic
#define DEFAULT_FILE_NAME "/tmp/music.input"
struct module_pipesrc_data {
struct module *module;
struct pw_core *core;
struct pw_stream *playback;
struct spa_hook core_listener;
struct spa_hook playback_listener;
struct spa_hook mod_listener;
struct pw_impl_module *mod;
struct pw_properties *playback_props;
struct spa_audio_info_raw info;
bool do_unlink;
char *filename;
int fd;
uint32_t stride;
uint32_t leftover_count;
uint8_t leftover[]; /* `stride` bytes for storing a partial sample */
};
static void playback_process(void *data)
{
struct module_pipesrc_data *impl = data;
struct spa_chunk *chunk;
struct pw_buffer *buffer;
struct spa_data *d;
uint32_t left, leftover;
ssize_t bytes_read;
if ((buffer = pw_stream_dequeue_buffer(impl->playback)) == NULL) {
pw_log_warn("Out of playback buffers: %m");
return;
}
d = &buffer->buffer->datas[0];
if (d->data == NULL)
return;
left = d->maxsize;
spa_assert(left >= impl->leftover_count);
chunk = d->chunk;
chunk->offset = 0;
chunk->stride = impl->stride;
chunk->size = impl->leftover_count;
memcpy(d->data, impl->leftover, impl->leftover_count);
left -= impl->leftover_count;
bytes_read = read(impl->fd, SPA_PTROFF(d->data, chunk->size, void), left);
if (bytes_read < 0) {
const bool important = !(errno == EINTR
|| errno == EAGAIN
|| errno == EWOULDBLOCK);
if (important)
pw_log_warn("failed to read from pipe (%s): %s",
impl->filename, strerror(errno));
}
else {
chunk->size += bytes_read;
}
leftover = chunk->size % impl->stride;
chunk->size -= leftover;
memcpy(impl->leftover, SPA_PTROFF(d->data, chunk->size, void), leftover);
impl->leftover_count = leftover;
pw_stream_queue_buffer(impl->playback, buffer);
}
static void on_core_error(void *data, uint32_t id, int seq, int res, const char *message)
static void module_destroy(void *data)
{
struct module_pipesrc_data *d = data;
pw_log_error("error id:%u seq:%d res:%d (%s): %s",
id, seq, res, spa_strerror(res), message);
if (id == PW_ID_CORE && res == -EPIPE)
spa_hook_remove(&d->mod_listener);
d->mod = NULL;
module_schedule_unload(d->module);
}
static const struct pw_core_events core_events = {
PW_VERSION_CORE_EVENTS,
.error = on_core_error,
};
static void on_stream_state_changed(void *data, enum pw_stream_state old,
enum pw_stream_state state, const char *error)
{
struct module_pipesrc_data *d = data;
switch (state) {
case PW_STREAM_STATE_UNCONNECTED:
pw_log_info("stream disconnected, unloading");
module_schedule_unload(d->module);
break;
case PW_STREAM_STATE_ERROR:
pw_log_error("stream error: %s", error);
break;
default:
break;
}
}
static const struct pw_stream_events out_stream_events = {
PW_VERSION_STREAM_EVENTS,
.state_changed = on_stream_state_changed,
.process = playback_process
static const struct pw_impl_module_events module_events = {
PW_VERSION_IMPL_MODULE_EVENTS,
.destroy = module_destroy
};
static int module_pipe_source_load(struct client *client, struct module *module)
{
struct module_pipesrc_data *data = module->user_data;
int res;
uint32_t n_params;
const struct spa_pod *params[1];
uint8_t buffer[1024];
struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer));
data->core = pw_context_connect(module->impl->context,
pw_properties_copy(client->props),
0);
if (data->core == NULL)
return -errno;
pw_core_add_listener(data->core,
&data->core_listener,
&core_events, data);
FILE *f;
char *args;
size_t size;
uint32_t i;
pw_properties_setf(data->playback_props, "pulse.module.id",
"%u", module->index);
data->playback = pw_stream_new(data->core,
"pipesource playback", data->playback_props);
data->playback_props = NULL;
if (data->playback == NULL)
if ((f = open_memstream(&args, &size)) == NULL)
return -errno;
pw_stream_add_listener(data->playback,
&data->playback_listener,
&out_stream_events, data);
fprintf(f, "{");
fprintf(f, " \"tunnel.mode\" = \"source\" ");
if (data->filename != NULL)
fprintf(f, " \"pipe.filename\": \"%s\"", data->filename);
if (data->info.format != 0)
fprintf(f, " \"audio.format\": \"%s\"", format_id2name(data->info.format));
if (data->info.rate != 0)
fprintf(f, " \"audio.rate\": %u,", data->info.rate);
if (data->info.channels != 0) {
fprintf(f, " \"audio.channels\": %u,", data->info.channels);
if (!(data->info.flags & SPA_AUDIO_FLAG_UNPOSITIONED)) {
fprintf(f, " \"audio.position\": [ ");
for (i = 0; i < data->info.channels; i++)
fprintf(f, "%s\"%s\"", i == 0 ? "" : ",",
channel_id2name(data->info.position[i]));
fprintf(f, " ],");
}
}
pw_properties_serialize_dict(f, &data->playback_props->dict, 0);
fprintf(f, " }");
fclose(f);
n_params = 0;
params[n_params++] = spa_format_audio_raw_build(&b, SPA_PARAM_EnumFormat,
&data->info);
data->mod = pw_context_load_module(module->impl->context,
"libpipewire-module-pipe-tunnel",
args, NULL);
if ((res = pw_stream_connect(data->playback,
PW_DIRECTION_OUTPUT,
PW_ID_ANY,
PW_STREAM_FLAG_AUTOCONNECT |
PW_STREAM_FLAG_MAP_BUFFERS |
PW_STREAM_FLAG_RT_PROCESS,
params, n_params)) < 0)
return res;
free(args);
if (data->mod == NULL)
return -errno;
pw_impl_module_add_listener(data->mod,
&data->mod_listener,
&module_events, data);
return 0;
}
@ -212,17 +119,13 @@ static int module_pipe_source_unload(struct module *module)
{
struct module_pipesrc_data *d = module->user_data;
if (d->mod) {
spa_hook_remove(&d->mod_listener);
pw_impl_module_destroy(d->mod);
d->mod = NULL;
}
pw_properties_free(d->playback_props);
if (d->playback != NULL)
pw_stream_destroy(d->playback);
if (d->core != NULL)
pw_core_disconnect(d->core);
if (d->do_unlink)
unlink(d->filename);
free(d->filename);
if (d->fd >= 0)
close(d->fd);
return 0;
}
@ -231,6 +134,7 @@ static const struct spa_dict_item module_pipe_source_info[] = {
{ PW_KEY_MODULE_DESCRIPTION, "Pipe source" },
{ PW_KEY_MODULE_USAGE, "file=<name of the FIFO special file to use> "
"source_name=<name for the source> "
"source_properties=<source properties> "
"format=<sample format> "
"rate=<sample rate> "
"channels=<number of channels> "
@ -244,12 +148,9 @@ static int module_pipe_source_prepare(struct module * const module)
struct pw_properties * const props = module->props;
struct pw_properties *playback_props = NULL;
struct spa_audio_info_raw info = { 0 };
struct stat st;
const char *str;
bool do_unlink = false;
char *filename = NULL;
int stride, res = 0;
int fd = -1;
int res = 0;
PW_LOG_TOPIC_INIT(mod_topic);
@ -264,119 +165,34 @@ static int module_pipe_source_prepare(struct module * const module)
goto out;
}
info.format = SPA_AUDIO_FORMAT_S16;
if ((str = pw_properties_get(props, "format")) != NULL) {
info.format = format_paname2id(str, strlen(str));
pw_properties_set(props, "format", NULL);
}
switch (info.format) {
case SPA_AUDIO_FORMAT_U8:
stride = info.channels;
break;
case SPA_AUDIO_FORMAT_S16_LE:
case SPA_AUDIO_FORMAT_S16_BE:
case SPA_AUDIO_FORMAT_S16P:
stride = 2 * info.channels;
break;
case SPA_AUDIO_FORMAT_S24_LE:
case SPA_AUDIO_FORMAT_S24_BE:
case SPA_AUDIO_FORMAT_S24P:
stride = 3 * info.channels;
break;
case SPA_AUDIO_FORMAT_F32_LE:
case SPA_AUDIO_FORMAT_F32_BE:
case SPA_AUDIO_FORMAT_F32P:
case SPA_AUDIO_FORMAT_S32_LE:
case SPA_AUDIO_FORMAT_S32_BE:
case SPA_AUDIO_FORMAT_S32P:
case SPA_AUDIO_FORMAT_S24_32_LE:
case SPA_AUDIO_FORMAT_S24_32_BE:
case SPA_AUDIO_FORMAT_S24_32P:
stride = 4 * info.channels;
break;
default:
pw_log_error("Invalid audio format");
res = -EINVAL;
goto out;
}
if ((str = pw_properties_get(props, "source_name")) != NULL) {
pw_properties_set(playback_props, PW_KEY_NODE_NAME, str);
pw_properties_set(props, "source_name", NULL);
}
if ((str = pw_properties_get(props, "source_properties")) != NULL)
module_args_add_props(playback_props, str);
if ((str = pw_properties_get(props, "file")) != NULL) {
filename = strdup(str);
pw_properties_set(props, "file", NULL);
} else {
filename = strdup(DEFAULT_FILE_NAME);
}
if (filename == NULL) {
res = -ENOMEM;
goto out;
}
if (mkfifo(filename, 0666) < 0) {
if (errno != EEXIST) {
res = -errno;
pw_log_error("mkfifo('%s'): %s", filename, spa_strerror(res));
goto out;
}
do_unlink = false;
} else {
/*
* Our umask is 077, so the pipe won't be created with the
* requested permissions. Let's fix the permissions with chmod().
*/
if (chmod(filename, 0666) < 0)
pw_log_warn("chmod('%s'): %s", filename, spa_strerror(-errno));
do_unlink = true;
}
if ((fd = open(filename, O_RDONLY | O_CLOEXEC | O_NONBLOCK, 0)) <= 0) {
res = -errno;
pw_log_error("open('%s'): %s", filename, spa_strerror(res));
goto out;
}
if (fstat(fd, &st) < 0) {
res = -errno;
pw_log_error("fstat('%s'): %s", filename, spa_strerror(res));
goto out;
}
if (!S_ISFIFO(st.st_mode)) {
res = -EEXIST;
pw_log_error("'%s' is not a FIFO", filename);
goto out;
}
if (pw_properties_get(playback_props, PW_KEY_NODE_VIRTUAL) == NULL)
pw_properties_set(playback_props, PW_KEY_NODE_VIRTUAL, "true");
pw_properties_set(playback_props, PW_KEY_MEDIA_CLASS, "Audio/Source");
d->module = module;
d->playback_props = playback_props;
d->info = info;
d->fd = fd;
d->filename = filename;
d->do_unlink = do_unlink;
d->stride = stride;
pw_log_info("Successfully loaded module-pipe-source");
return 0;
out:
pw_properties_free(playback_props);
if (do_unlink)
unlink(filename);
free(filename);
if (fd >= 0)
close(fd);
return res;
}