From 96c77e1f2f91073b63b3952af950c9e8af456a29 Mon Sep 17 00:00:00 2001 From: Sanchayan Maity Date: Sat, 12 Jun 2021 14:41:27 +0530 Subject: [PATCH] pulse-server: Implement module-pipe-source --- src/modules/meson.build | 1 + src/modules/module-protocol-pulse/module.c | 1 + .../modules/module-pipe-source.c | 367 ++++++++++++++++++ .../module-protocol-pulse/modules/registry.h | 1 + 4 files changed, 370 insertions(+) create mode 100644 src/modules/module-protocol-pulse/modules/module-pipe-source.c diff --git a/src/modules/meson.build b/src/modules/meson.build index ce03d6a4a..1cdf1ee8c 100644 --- a/src/modules/meson.build +++ b/src/modules/meson.build @@ -142,6 +142,7 @@ pipewire_module_protocol_pulse_sources = [ 'module-protocol-pulse/modules/module-loopback.c', 'module-protocol-pulse/modules/module-native-protocol-tcp.c', 'module-protocol-pulse/modules/module-null-sink.c', + 'module-protocol-pulse/modules/module-pipe-source.c', 'module-protocol-pulse/modules/module-pipe-sink.c', 'module-protocol-pulse/modules/module-remap-sink.c', 'module-protocol-pulse/modules/module-remap-source.c', diff --git a/src/modules/module-protocol-pulse/module.c b/src/modules/module-protocol-pulse/module.c index d4e694748..2387c9c90 100644 --- a/src/modules/module-protocol-pulse/module.c +++ b/src/modules/module-protocol-pulse/module.c @@ -225,6 +225,7 @@ static const struct module_info module_list[] = { { "module-loopback", create_module_loopback, }, { "module-null-sink", create_module_null_sink, }, { "module-native-protocol-tcp", create_module_native_protocol_tcp, }, + { "module-pipe-source", create_module_pipe_source, }, { "module-pipe-sink", create_module_pipe_sink, }, { "module-remap-sink", create_module_remap_sink, }, { "module-remap-source", create_module_remap_source, }, diff --git a/src/modules/module-protocol-pulse/modules/module-pipe-source.c b/src/modules/module-protocol-pulse/modules/module-pipe-source.c new file mode 100644 index 000000000..cbd037eab --- /dev/null +++ b/src/modules/module-protocol-pulse/modules/module-pipe-source.c @@ -0,0 +1,367 @@ +/* PipeWire + * + * Copyright © 2021 Wim Taymans + * Copyright © 2021 Sanchayan Maity + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice (including the next + * paragraph) shall be included in all copies or substantial portions of the + * Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + */ + +#include +#include +#include + +#include +#include +#include +#include + +#include "../defs.h" +#include "../module.h" +#include "registry.h" + +#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 pw_properties *playback_props; + + struct spa_audio_info_raw info; + + char *filename; + int fd; + int stride; +}; + +static void playback_process(void *data) +{ + struct module_pipesrc_data *impl = data; + struct pw_buffer *b; + struct spa_buffer *buf; + int bytes_read; + uint8_t *dst; + + if ((b = pw_stream_dequeue_buffer(impl->playback)) == NULL) { + pw_log_warn("Out of playback buffers: %m"); + return; + } + + buf = b->buffer; + if ((dst = buf->datas[0].data) == NULL) + return; + + buf->datas[0].chunk->offset = 0; + buf->datas[0].chunk->stride = impl->stride; + buf->datas[0].chunk->size = 0; + + /* + * Read from the pipe in a multiple of stride. If the number of bytes + * read is not a multiple of stride, we end up with incomplete samples + * in the buffer. This is especially noticeable when using S24LE. + * + * Trying to read (buf->datas[0].maxsize / stride) * stride does not + * necessarily result in bytes_read being a multiple of stride + * resulting in us being left with incomplete samples in the buffer + * again. + */ + bytes_read = read(impl->fd, dst, impl->stride * 4096); + if (bytes_read < 0) { + if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) { + pw_log_debug("Error in reading from pipe source: %s", + spa_strerror(-errno)); + } else { + pw_log_error("Failed to read from pipe source: %s", + spa_strerror(-errno)); + } + } else { + buf->datas[0].chunk->size = bytes_read; + } + + pw_stream_queue_buffer(impl->playback, b); +} + +static void on_core_error(void *data, uint32_t id, int seq, int res, const char *message) +{ + 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) + 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 int module_pipesource_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); + + data->playback = pw_stream_new(data->core, + "pipesource playback", data->playback_props); + data->playback_props = NULL; + if (data->playback == NULL) + return -errno; + + pw_stream_add_listener(data->playback, + &data->playback_listener, + &out_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->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; + + return 0; +} + +static int module_pipesource_unload(struct client *client, struct module *module) +{ + struct module_pipesrc_data *d = module->user_data; + + 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->filename) { + unlink(d->filename); + free(d->filename); + } + if (d->fd >= 0) + close(d->fd); + + return 0; +} + +static const struct module_methods module_pipesource_methods = { + VERSION_MODULE_METHODS, + .load = module_pipesource_load, + .unload = module_pipesource_unload, +}; + +static const struct spa_dict_item module_pipesource_info[] = { + { PW_KEY_MODULE_AUTHOR, "Sanchayan Maity " }, + { PW_KEY_MODULE_DESCRIPTION, "Pipe source" }, + { PW_KEY_MODULE_USAGE, "file= " + "source_name= " + "format= " + "rate= " + "channels= " + "channel_map= " }, + { PW_KEY_MODULE_VERSION, PACKAGE_VERSION }, +}; + +struct module *create_module_pipe_source(struct impl *impl, const char *argument) +{ + struct module *module; + struct module_pipesrc_data *d; + struct pw_properties *props = NULL, *playback_props = NULL; + struct spa_audio_info_raw info = { 0 }; + struct stat st; + const char *str; + char *filename = NULL; + int stride, res = 0; + int fd = -1; + + props = pw_properties_new_dict(&SPA_DICT_INIT_ARRAY(module_pipesource_info)); + playback_props = pw_properties_new(NULL, NULL); + if (!props || !playback_props) { + res = -errno; + goto out; + } + if (argument) + module_args_add_props(props, argument); + + if (module_args_to_audioinfo(impl, props, &info) < 0) { + res = -EINVAL; + goto out; + } + + 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, "file")) != NULL) { + filename = strdup(str); + pw_properties_set(props, "file", NULL); + } else { + filename = strdup(DEFAULT_FILE_NAME); + } + + if (mkfifo(filename, 0666) < 0) { + if (errno != EEXIST) { + res = -errno; + pw_log_error("mkfifo('%s'): %s", filename, spa_strerror(res)); + goto out; + } + } 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)); + } + + 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)) { + pw_log_error("'%s' is not a FIFO.", filename); + goto out; + } + + if ((str = pw_properties_get(props, PW_KEY_MEDIA_CLASS)) == NULL) + pw_properties_set(props, PW_KEY_MEDIA_CLASS, "Audio/Source"); + + module = module_new(impl, &module_pipesource_methods, sizeof(*d)); + if (module == NULL) { + res = -errno; + goto out; + } + + module->props = props; + d = module->user_data; + d->module = module; + d->playback_props = playback_props; + d->info = info; + d->fd = fd; + d->filename = filename; + d->stride = stride; + + pw_log_info("Successfully loaded module-pipe-source"); + + return module; +out: + pw_properties_free(props); + pw_properties_free(playback_props); + if (filename) { + unlink(filename); + free(filename); + } + if (fd >= 0) + close(fd); + errno = -res; + + return NULL; +} diff --git a/src/modules/module-protocol-pulse/modules/registry.h b/src/modules/module-protocol-pulse/modules/registry.h index 18c82d570..0614c3990 100644 --- a/src/modules/module-protocol-pulse/modules/registry.h +++ b/src/modules/module-protocol-pulse/modules/registry.h @@ -40,6 +40,7 @@ struct module *create_module_remap_source(struct impl *impl, const char *argumen struct module *create_module_tunnel_sink(struct impl *impl, const char *argument); struct module *create_module_tunnel_source(struct impl *impl, const char *argument); struct module *create_module_simple_protocol_tcp(struct impl *impl, const char *argument); +struct module *create_module_pipe_source(struct impl *impl, const char *argument); struct module *create_module_pipe_sink(struct impl *impl, const char *argument); struct module *create_module_zeroconf_discover(struct impl *impl, const char *argument); struct module *create_module_zeroconf_publish(struct impl *impl, const char *argument);