diff --git a/meson.build b/meson.build index d2efa51c6..e8249ed45 100644 --- a/meson.build +++ b/meson.build @@ -427,6 +427,9 @@ need_alsa = get_option('pipewire-alsa').enabled() or 'media-session' in get_opti alsa_dep = dependency('alsa', version : '>=1.1.7', required: need_alsa) summary({'pipewire-alsa': alsa_dep.found()}, bool_yn: true) +openssl_lib = dependency('openssl', required: false) +summary({'openssl': openssl_lib.found()}, bool_yn: true) + installed_tests_metadir = pipewire_datadir / 'installed-tests' / pipewire_name installed_tests_execdir = pipewire_libexecdir / 'installed-tests' / pipewire_name installed_tests_enabled = not get_option('installed_tests').disabled() diff --git a/src/modules/meson.build b/src/modules/meson.build index 2fbf0116d..e9f0bae40 100644 --- a/src/modules/meson.build +++ b/src/modules/meson.build @@ -22,6 +22,7 @@ module_sources = [ 'module-pulse-tunnel.c', 'module-rt.c', 'module-rtkit.c', + 'module-raop-discover.c', 'module-session-manager.c', 'module-zeroconf-discover.c', 'module-roc-source.c', @@ -428,6 +429,34 @@ pipewire_module_zeroconf_discover = shared_library('pipewire-module-zeroconf-dis endif summary({'zeroconf-discover': build_module_zeroconf_discover}, bool_yn: true, section: 'Optional Modules') +build_module_raop_discover = avahi_dep.found() +if build_module_raop_discover +pipewire_module_raop_discover = shared_library('pipewire-module-raop-discover', + [ 'module-raop-discover.c', + 'module-zeroconf-discover/avahi-poll.c' ], + include_directories : [configinc, spa_inc], + install : true, + install_dir : modules_install_dir, + install_rpath: modules_install_dir, + dependencies : [mathlib, dl_lib, rt_lib, pipewire_dep, avahi_dep], +) +endif +summary({'raop-discover': build_module_raop_discover}, bool_yn: true, section: 'Optional Modules') + +build_module_raop = openssl_lib.found() +if build_module_raop +pipewire_module_raop_sink = shared_library('pipewire-module-raop-sink', + [ 'module-raop-sink.c', + 'module-raop/rtsp-client.c' ], + include_directories : [configinc, spa_inc], + install : true, + install_dir : modules_install_dir, + install_rpath: modules_install_dir, + dependencies : [mathlib, dl_lib, rt_lib, pipewire_dep, openssl_lib], +) +endif +summary({'raop-sink': build_module_raop}, bool_yn: true, section: 'Optional Modules') + build_module_roc = roc_lib.found() if build_module_roc pipewire_module_roc_sink = shared_library('pipewire-module-roc-sink', diff --git a/src/modules/module-raop-discover.c b/src/modules/module-raop-discover.c new file mode 100644 index 000000000..6ac8402f7 --- /dev/null +++ b/src/modules/module-raop-discover.c @@ -0,0 +1,541 @@ +/* PipeWire + * + * Copyright © 2021 Wim Taymans + * + * 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 "config.h" + +#include +#include +#include + +#include +#include +#include + +#include +#include +#include + +#include "module-protocol-pulse/format.h" +#include "module-zeroconf-discover/avahi-poll.h" + +/** \page page_module_zeroconf_discover PipeWire Module: Zeroconf Discover + */ + +#define NAME "zeroconf-discover" + +PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME); +#define PW_LOG_TOPIC_DEFAULT mod_topic + +#define MODULE_USAGE " " + +static const struct spa_dict_item module_props[] = { + { PW_KEY_MODULE_AUTHOR, "Wim Taymans " }, + { PW_KEY_MODULE_DESCRIPTION, "Discover remote streams" }, + { PW_KEY_MODULE_USAGE, MODULE_USAGE }, + { PW_KEY_MODULE_VERSION, PACKAGE_VERSION }, +}; + +#define SERVICE_TYPE_SINK "_raop._tcp" + +struct impl { + struct pw_context *context; + + struct pw_impl_module *module; + struct spa_hook module_listener; + struct pw_work_queue *work; + + struct pw_properties *properties; + + AvahiPoll *avahi_poll; + AvahiClient *client; + AvahiServiceBrowser *sink_browser; + + struct spa_list tunnel_list; + + unsigned int unloading:1; +}; + +struct tunnel_info { + AvahiIfIndex interface; + AvahiProtocol protocol; + const char *name; + const char *type; + const char *domain; +}; + +#define TUNNEL_INFO(...) (struct tunnel_info){ __VA_ARGS__ } + +struct tunnel { + struct spa_list link; + struct tunnel_info info; + struct pw_impl_module *module; + struct spa_hook module_listener; +}; + +static int start_client(struct impl *impl); + +static void do_unload_module(void *obj, void *data, int res, uint32_t id) +{ + struct impl *impl = data; + pw_impl_module_destroy(impl->module); +} + +static void unload_module(struct impl *impl) +{ + if (!impl->unloading) { + impl->unloading = true; + pw_work_queue_add(impl->work, impl, 0, do_unload_module, impl); + } +} + +static struct tunnel *make_tunnel(struct impl *impl, const struct tunnel_info *info) +{ + struct tunnel *t; + + t = calloc(1, sizeof(*t)); + if (t == NULL) + return NULL; + + t->info.interface = info->interface; + t->info.protocol = info->protocol; + t->info.name = strdup(info->name); + t->info.type = strdup(info->type); + t->info.domain = strdup(info->domain); + spa_list_append(&impl->tunnel_list, &t->link); + + return t; +} + +static struct tunnel *find_tunnel(struct impl *impl, const struct tunnel_info *info) +{ + struct tunnel *t; + spa_list_for_each(t, &impl->tunnel_list, link) { + if (t->info.interface == info->interface && + t->info.protocol == info->protocol && + spa_streq(t->info.name, info->name) && + spa_streq(t->info.type, info->type) && + spa_streq(t->info.domain, info->domain)) + return t; + } + return NULL; +} + +static void free_tunnel(struct tunnel *t) +{ + pw_impl_module_destroy(t->module); +} + +static void impl_free(struct impl *impl) +{ + struct tunnel *t; + + spa_list_consume(t, &impl->tunnel_list, link) + free_tunnel(t); + + if (impl->sink_browser) + avahi_service_browser_free(impl->sink_browser); + if (impl->client) + avahi_client_free(impl->client); + if (impl->avahi_poll) + pw_avahi_poll_free(impl->avahi_poll); + pw_properties_free(impl->properties); + if (impl->work) + pw_work_queue_cancel(impl->work, impl, SPA_ID_INVALID); + free(impl); +} + +static void module_destroy(void *data) +{ + struct impl *impl = data; + spa_hook_remove(&impl->module_listener); + impl_free(impl); +} + +static const struct pw_impl_module_events module_events = { + PW_VERSION_IMPL_MODULE_EVENTS, + .destroy = module_destroy, +}; + +static bool str_in_list(const char *haystack, const char *delimiters, const char *needle) +{ + const char *s, *state = NULL; + size_t len; + while ((s = pw_split_walk(haystack, delimiters, &len, &state))) { + if (spa_strneq(needle, s, len)) + return true; + } + return false; +} + +static void pw_properties_from_avahi_string(const char *key, const char *value, + struct pw_properties *props) +{ + if (spa_streq(key, "device")) { + pw_properties_set(props, "raop.device", value); + } + else if (spa_streq(key, "tp")) { + /* transport protocol, "UDP", "TCP", "UDP,TCP" */ + if (str_in_list(value, ",", "UDP")) + value = "udp"; + else if (str_in_list(value, ",", "TCP")) + value = "tcp"; + pw_properties_set(props, "raop.transport", value); + } else if (spa_streq(key, "et")) { + /* Supported encryption types: + * 0 = none, + * 1 = RSA, + * 2 = FairPlay, + * 3 = MFiSAP, + * 4 = FairPlay SAPv2.5. */ + if (str_in_list(value, ",", "1")) + value = "RSA"; + else + value = "none"; + pw_properties_set(props, "raop.encryption.type", value); + } else if (spa_streq(key, "cn")) { + /* Suported audio codecs: + * 0 = PCM, + * 1 = ALAC, + * 2 = AAC, + * 3 = AAC ELD. */ + if (str_in_list(value, ",", "0")) + value = "PCM"; + else if (str_in_list(value, ",", "1")) + value = "ALAC"; + else if (str_in_list(value, ",", "2")) + value = "AAC"; + else if (str_in_list(value, ",", "2")) + value = "AAC-ELD"; + else + value = "unknown"; + pw_properties_set(props, "raop.audio.codec", value); + } else if (spa_streq(key, "ch")) { + /* Number of channels */ + pw_properties_set(props, PW_KEY_AUDIO_CHANNELS, value); + } else if (spa_streq(key, "ss")) { + /* Sample size */ + if (spa_streq(value, "16")) + value = "S16"; + else if (spa_streq(value, "24")) + value = "S24"; + else if (spa_streq(value, "32")) + value = "S32"; + else + value = "UNKNOWN"; + pw_properties_set(props, PW_KEY_AUDIO_FORMAT, value); + } else if (spa_streq(key, "sr")) { + /* Sample rate */ + pw_properties_set(props, PW_KEY_AUDIO_RATE, value); + } else if (spa_streq(key, "am")) { + /* Device model */ + pw_properties_set(props, "device.model", value); + } +} + +static void submodule_destroy(void *data) +{ + struct tunnel *t = data; + + spa_list_remove(&t->link); + spa_hook_remove(&t->module_listener); + + free((char *) t->info.name); + free((char *) t->info.type); + free((char *) t->info.domain); + + free(t); +} + +static const struct pw_impl_module_events submodule_events = { + PW_VERSION_IMPL_MODULE_EVENTS, + .destroy = submodule_destroy, +}; + +static void resolver_cb(AvahiServiceResolver *r, AvahiIfIndex interface, AvahiProtocol protocol, + AvahiResolverEvent event, const char *name, const char *type, const char *domain, + const char *host_name, const AvahiAddress *a, uint16_t port, AvahiStringList *txt, + AvahiLookupResultFlags flags, void *userdata) +{ + struct impl *impl = userdata; + struct tunnel *t; + struct tunnel_info tinfo; + const char *str; + AvahiStringList *l; + FILE *f; + char *args; + size_t size; + struct pw_impl_module *mod; + struct pw_properties *props = NULL; + char at[AVAHI_ADDRESS_STR_MAX]; + + if (event != AVAHI_RESOLVER_FOUND) { + pw_log_error("Resolving of '%s' failed: %s", name, + avahi_strerror(avahi_client_errno(impl->client))); + goto done; + } + tinfo = TUNNEL_INFO(.interface = interface, + .protocol = protocol, + .name = name, + .type = type, + .domain = domain); + + props = pw_properties_new(NULL, NULL); + if (props == NULL) { + pw_log_error("Can't allocate properties: %m"); + goto done; + } + + avahi_address_snprint(at, sizeof(at), a); + + pw_properties_setf(props, "raop.hostname", "%s", at); + pw_properties_setf(props, "raop.port", "%u", port); + + if ((str = strstr(name, "@"))) { + str++; + if (strlen(str) > 0) + pw_properties_set(props, PW_KEY_NODE_DESCRIPTION, str); + else + pw_properties_setf(props, PW_KEY_NODE_DESCRIPTION, + "RAOP on %s", host_name); + } + + for (l = txt; l; l = l->next) { + char *key, *value; + + if (avahi_string_list_get_pair(l, &key, &value, NULL) != 0) + break; + + pw_properties_from_avahi_string(key, value, props); + avahi_free(key); + avahi_free(value); + } + + + f = open_memstream(&args, &size); + fprintf(f, "{"); + pw_properties_serialize_dict(f, &props->dict, 0); + fprintf(f, " stream.props = {"); + fprintf(f, " }"); + fprintf(f, "}"); + fclose(f); + + pw_properties_free(props); + + pw_log_info("loading module args:'%s'", args); + mod = pw_context_load_module(impl->context, + "libpipewire-module-raop-sink", + args, NULL); + free(args); + + if (mod == NULL) { + pw_log_error("Can't load module: %m"); + goto done; + } + + t = make_tunnel(impl, &tinfo); + if (t == NULL) { + pw_log_error("Can't make tunnel: %m"); + pw_impl_module_destroy(mod); + goto done; + } + + pw_impl_module_add_listener(mod, &t->module_listener, &submodule_events, t); + + t->module = mod; + +done: + avahi_service_resolver_free(r); +} + + +static void browser_cb(AvahiServiceBrowser *b, AvahiIfIndex interface, AvahiProtocol protocol, + AvahiBrowserEvent event, const char *name, const char *type, const char *domain, + AvahiLookupResultFlags flags, void *userdata) +{ + struct impl *impl = userdata; + struct tunnel_info info; + struct tunnel *t; + + if (flags & AVAHI_LOOKUP_RESULT_LOCAL) + return; + + info = TUNNEL_INFO(.interface = interface, + .protocol = protocol, + .name = name, + .type = type, + .domain = domain); + + t = find_tunnel(impl, &info); + + switch (event) { + case AVAHI_BROWSER_NEW: + if (t != NULL) + return; + if (!(avahi_service_resolver_new(impl->client, + interface, protocol, + name, type, domain, + AVAHI_PROTO_UNSPEC, 0, + resolver_cb, impl))) + pw_log_error("can't make service resolver: %s", + avahi_strerror(avahi_client_errno(impl->client))); + break; + case AVAHI_BROWSER_REMOVE: + if (t == NULL) + return; + free_tunnel(t); + break; + default: + break; + } +} + + +static struct AvahiServiceBrowser *make_browser(struct impl *impl, const char *service_type) +{ + struct AvahiServiceBrowser *s; + + s = avahi_service_browser_new(impl->client, + AVAHI_IF_UNSPEC, AVAHI_PROTO_UNSPEC, + service_type, NULL, 0, + browser_cb, impl); + if (s == NULL) { + pw_log_error("can't make browser for %s: %s", service_type, + avahi_strerror(avahi_client_errno(impl->client))); + } + return s; +} + +static void client_callback(AvahiClient *c, AvahiClientState state, void *userdata) +{ + struct impl *impl = userdata; + + impl->client = c; + + switch (state) { + case AVAHI_CLIENT_S_REGISTERING: + case AVAHI_CLIENT_S_RUNNING: + case AVAHI_CLIENT_S_COLLISION: + if (impl->sink_browser == NULL) + impl->sink_browser = make_browser(impl, SERVICE_TYPE_SINK); + if (impl->sink_browser == NULL) + goto error; + break; + case AVAHI_CLIENT_FAILURE: + if (avahi_client_errno(c) == AVAHI_ERR_DISCONNECTED) + start_client(impl); + + SPA_FALLTHROUGH; + case AVAHI_CLIENT_CONNECTING: + if (impl->sink_browser) { + avahi_service_browser_free(impl->sink_browser); + impl->sink_browser = NULL; + } + break; + default: + break; + } + return; +error: + unload_module(impl); +} + +static int start_client(struct impl *impl) +{ + int res; + if ((impl->client = avahi_client_new(impl->avahi_poll, + AVAHI_CLIENT_NO_FAIL, + client_callback, impl, + &res)) == NULL) { + pw_log_error("can't create client: %s", avahi_strerror(res)); + unload_module(impl); + return -EIO; + } + return 0; +} + +static int start_avahi(struct impl *impl) +{ + struct pw_loop *loop; + + loop = pw_context_get_main_loop(impl->context); + impl->avahi_poll = pw_avahi_poll_new(loop); + + return start_client(impl); +} + +SPA_EXPORT +int pipewire__module_init(struct pw_impl_module *module, const char *args) +{ + struct pw_context *context = pw_impl_module_get_context(module); + struct pw_properties *props; + struct impl *impl; + int res; + + PW_LOG_TOPIC_INIT(mod_topic); + + impl = calloc(1, sizeof(struct impl)); + if (impl == NULL) + goto error_errno; + + pw_log_debug("module %p: new %s", impl, args); + + if (args == NULL) + args = ""; + + props = pw_properties_new_string(args); + if (props == NULL) + goto error_errno; + + spa_list_init(&impl->tunnel_list); + + impl->module = module; + impl->context = context; + impl->properties = props; + + impl->work = pw_context_get_work_queue(context); + if (impl->work == NULL) + goto error_errno; + + pw_impl_module_add_listener(module, &impl->module_listener, &module_events, impl); + + pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_props)); + + start_avahi(impl); + + return 0; + +error_errno: + res = -errno; + if (impl) + impl_free(impl); + return res; +} diff --git a/src/modules/module-raop-sink.c b/src/modules/module-raop-sink.c new file mode 100644 index 000000000..f414994fe --- /dev/null +++ b/src/modules/module-raop-sink.c @@ -0,0 +1,1063 @@ +/* PipeWire + * + * Copyright © 2021 Wim Taymans + * + * 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 +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include "config.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include "module-raop/rtsp-client.h" + +/** \page page_module_raop_sink PipeWire Module: AirPlay Sink + */ + +#define NAME "raop-sink" + +PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME); +#define PW_LOG_TOPIC_DEFAULT mod_topic + +#define FRAMES_PER_TCP_PACKET 4096 +#define FRAMES_PER_UDP_PACKET 352 + +#define DEFAULT_TCP_AUDIO_PORT 6000 +#define DEFAULT_UDP_AUDIO_PORT 6000 +#define DEFAULT_UDP_CONTROL_PORT 6001 +#define DEFAULT_UDP_TIMING_PORT 6002 + +#define AES_CHUNK_SIZE 16 + +#define MAX_PORT_RETRY 16 + +#define DEFAULT_FORMAT "S16" +#define DEFAULT_RATE 48000 +#define DEFAULT_CHANNELS "2" +#define DEFAULT_POSITION "[ FL FR ]" + +#define MODULE_USAGE "[ node.latency= ] " \ + "[ node.name= ] " \ + "[ node.description= ] " \ + "[ audio.format= ] " \ + "[ audio.rate= ] " \ + "[ audio.channels= ] " \ + "[ audio.position= ] " \ + "[ stream.props= ] " + + +static const struct spa_dict_item module_props[] = { + { PW_KEY_MODULE_AUTHOR, "Wim Taymans " }, + { PW_KEY_MODULE_DESCRIPTION, "An RAOP audio sink" }, + { PW_KEY_MODULE_USAGE, MODULE_USAGE }, + { PW_KEY_MODULE_VERSION, PACKAGE_VERSION }, +}; + +struct impl { + struct pw_context *context; + + struct pw_properties *props; + + struct pw_impl_module *module; + struct pw_loop *loop; + struct pw_work_queue *work; + + struct spa_hook module_listener; + + int protocol; + + struct pw_core *core; + struct spa_hook core_proxy_listener; + struct spa_hook core_listener; + + struct pw_properties *stream_props; + struct pw_stream *stream; + struct spa_hook stream_listener; + struct spa_audio_info_raw info; + uint32_t frame_size; + + struct pw_rtsp_client *rtsp; + struct spa_hook rtsp_listener; + struct pw_properties *headers; + + char *session_id; + + unsigned int do_disconnect:1; + unsigned int unloading:1; + + uint8_t key[AES_CHUNK_SIZE]; /* Key for aes-cbc */ + uint8_t iv[AES_CHUNK_SIZE]; /* Initialization vector for cbc */ + AES_KEY aes; /* AES encryption */ + + uint16_t control_port; + int control_fd; + uint16_t timing_port; + int timing_fd; + uint16_t server_port; + int server_fd; + + uint16_t seq; + uint32_t rtptime; + uint32_t ssrc; +}; + +static void do_unload_module(void *obj, void *data, int res, uint32_t id) +{ + struct impl *impl = data; + pw_impl_module_destroy(impl->module); +} + +static void unload_module(struct impl *impl) +{ + if (!impl->unloading) { + impl->unloading = true; + pw_work_queue_add(impl->work, impl, 0, do_unload_module, impl); + } +} + +static void stream_destroy(void *d) +{ + struct impl *impl = d; + spa_hook_remove(&impl->stream_listener); + impl->stream = NULL; +} + +static void stream_state_changed(void *d, enum pw_stream_state old, + enum pw_stream_state state, const char *error) +{ + struct impl *impl = d; + switch (state) { + case PW_STREAM_STATE_ERROR: + case PW_STREAM_STATE_UNCONNECTED: + unload_module(impl); + break; + case PW_STREAM_STATE_PAUSED: + case PW_STREAM_STATE_STREAMING: + break; + default: + break; + } +} + +static inline void bit_writer(uint8_t **p, int *pos, uint8_t data, int len) +{ + int lb, rb; + + lb = 7 - *pos; + rb = lb - len + 1; + + if (rb >= 0) { + **p = (*pos ? **p : 0) | (data << rb); + *pos += len; + } else { + **p |= (data >> -rb); + (*p)++; + **p = data << (8+rb); + *pos = -rb; + } +} + +static int aes_encrypt(struct impl *impl, uint8_t *data, int len) +{ + uint8_t nv[AES_CHUNK_SIZE]; + uint8_t *buffer; + int i, j; + + memcpy(nv, impl->iv, AES_CHUNK_SIZE); + for (i = 0; i + AES_CHUNK_SIZE <= len; i += AES_CHUNK_SIZE) { + buffer = data + i; + for (j = 0; j < AES_CHUNK_SIZE; j++) + buffer[j] ^= nv[j]; + + AES_encrypt(buffer, buffer, &impl->aes); + + memcpy(nv, buffer, AES_CHUNK_SIZE); + } + return i; +} + + +static int add_to_packet(struct impl *impl, uint8_t *data, size_t size) +{ + const size_t max = 12 + 8 + (FRAMES_PER_UDP_PACKET * 4); + uint8_t *pkt, *bp; + int bpos = 0, res; + + pkt = alloca(max); + pkt[0] = 0x80; /* RTP v2: 0x80 */ + pkt[1] = 0x60; /* Payload type: 0x60 */ + pkt[2] = impl->seq >> 8; /* Sequence number: 0x0000 */ + pkt[3] = impl->seq; + pkt[4] = impl->rtptime >> 24; /* Timestamp */ + pkt[5] = impl->rtptime >> 16; + pkt[6] = impl->rtptime >> 8; + pkt[7] = impl->rtptime; + pkt[8] = impl->ssrc >> 24; + pkt[9] = impl->ssrc >> 16; + pkt[10] = impl->ssrc >> 8; + pkt[11] = impl->ssrc; + + bp = &pkt[12]; + + size = SPA_MIN(size, FRAMES_PER_UDP_PACKET * 4u); + + bit_writer(&bp, &bpos, 1, 3); /* channel=1, stereo */ + bit_writer(&bp, &bpos, 0, 4); /* Unknown */ + bit_writer(&bp, &bpos, 0, 8); /* Unknown */ + bit_writer(&bp, &bpos, 0, 4); /* Unknown */ + bit_writer(&bp, &bpos, 1, 1); /* Hassize */ + bit_writer(&bp, &bpos, 0, 2); /* Unused */ + bit_writer(&bp, &bpos, 1, 1); /* Is-not-compressed */ + /* Size of data, integer, big endian. */ + bit_writer(&bp, &bpos, (size >> 24) & 0xff, 8); + bit_writer(&bp, &bpos, (size >> 16) & 0xff, 8); + bit_writer(&bp, &bpos, (size >> 8) & 0xff, 8); + bit_writer(&bp, &bpos, (size) & 0xff, 8); + + impl->rtptime += size / 4; + + while (size > 4) { + /* Byte swap stereo data. */ + bit_writer(&bp, &bpos, *(data + 1), 8); + bit_writer(&bp, &bpos, *(data + 0), 8); + bit_writer(&bp, &bpos, *(data + 3), 8); + bit_writer(&bp, &bpos, *(data + 2), 8); + data += 4; + size -= 4; + } + + impl->seq = (impl->seq + 1) & 0xffff; + + aes_encrypt(impl, pkt + 12, max - 12); + + pw_log_info("send %zu", max); + res = write(impl->server_fd, pkt, max); + + return 0; +} + +static void playback_stream_process(void *d) +{ + struct impl *impl = d; + struct pw_buffer *buf; + struct spa_data *bd; + void *data; + uint32_t size; + + + if ((buf = pw_stream_dequeue_buffer(impl->stream)) == NULL) { + pw_log_debug("out of buffers: %m"); + return; + } + + bd = &buf->buffer->datas[0]; + data = SPA_PTROFF(bd->data, bd->chunk->offset, void); + size = bd->chunk->size; + + if (impl->server_fd > 0) + add_to_packet(impl, data, size); + + pw_stream_queue_buffer(impl->stream, buf); +} + +static const struct pw_stream_events playback_stream_events = { + PW_VERSION_STREAM_EVENTS, + .destroy = stream_destroy, + .state_changed = stream_state_changed, + .process = playback_stream_process +}; + +static int create_udp_socket(struct impl *impl, uint16_t *port) +{ + int res, ip_version, fd, val, i, af; + struct sockaddr_in sa4; + struct sockaddr_in6 sa6; + + if ((res = pw_rtsp_client_get_local_ip(impl->rtsp, + &ip_version, NULL, 0)) < 0) + return res; + + if (ip_version == 4) { + sa4.sin_family = af = AF_INET; + sa4.sin_addr.s_addr = INADDR_ANY; + } else { + sa6.sin6_family = af = AF_INET6; + sa6.sin6_addr = in6addr_any; + } + + if ((fd = socket(af, SOCK_DGRAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0) { + pw_log_error("socket failed: %m"); + return -errno; + } + +#ifdef SO_TIMESTAMP + val = 1; + if (setsockopt(fd, SOL_SOCKET, SO_TIMESTAMP, &val, sizeof(val)) < 0) { + res = -errno; + pw_log_error("setsockopt failed: %m"); + goto error; + } +#endif + val = 1; + if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val)) < 0) { + res = -errno; + pw_log_error("setsockopt failed: %m"); + goto error; + } + + for (i = 0; i < MAX_PORT_RETRY; i++) { + int ret; + + if (ip_version == 4) { + sa4.sin_port = htons(*port); + ret = bind(fd, &sa4, sizeof(sa4)); + } else { + sa6.sin6_port = htons(*port); + ret = bind(fd, &sa6, sizeof(sa6)); + } + if (ret == 0) + break; + if (ret < 0 && errno != EADDRINUSE) { + res = -errno; + pw_log_error("bind failed: %m"); + goto error; + } + (*port)++; + } + return fd; +error: + close(fd); + return res; +} + +static int connect_udp_socket(struct impl *impl, int fd, uint16_t port) +{ + const char *host; + struct sockaddr_in sa4; + struct sockaddr_in6 sa6; + struct sockaddr *sa; + size_t salen; + int res, af; + + host = pw_properties_get(impl->props, "raop.hostname"); + if (host == NULL) + return -EINVAL; + + if (inet_pton(AF_INET, host, &sa4.sin_addr) > 0) { + sa4.sin_family = af = AF_INET; + sa4.sin_port = htons(port); + sa = (struct sockaddr *) &sa4; + salen = sizeof(sa4); + } else if (inet_pton(AF_INET6, host, &sa6.sin6_addr) > 0) { + sa6.sin6_family = af = AF_INET6; + sa6.sin6_port = htons(port); + sa = (struct sockaddr *) &sa6; + salen = sizeof(sa6); + } else { + pw_log_error("Invalid host '%s'", host); + return -EINVAL; + } + + if (fd < 0 && + (fd = socket(af, SOCK_DGRAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0) { + pw_log_error("socket failed: %m"); + return -errno; + } + + if (connect(fd, sa, salen) < 0) { + res = -errno; + pw_log_error("connect failed: %m"); + goto error; + } + pw_log_info("Connected to host:%s port:%d", host, port); + return fd; + +error: + if (fd >= 0) + close(fd); + return res; +} + +static void rtsp_record_reply(void *data, int status, const struct spa_dict *headers) +{ + pw_log_info("reply %d", status); +} + +static void rtsp_do_record(struct impl *impl) +{ + pw_getrandom(&impl->seq, sizeof(impl->seq), 0); + pw_getrandom(&impl->rtptime, sizeof(impl->rtptime), 0); + + pw_properties_set(impl->headers, "Range", "npt=0-"); + pw_properties_setf(impl->headers, "RTP-Info", + "seq=%u;rtptime=%u", impl->seq, impl->rtptime); + + pw_rtsp_client_send(impl->rtsp, "RECORD", &impl->headers->dict, + NULL, NULL, rtsp_record_reply, impl); + + pw_properties_set(impl->headers, "Range", NULL); + pw_properties_set(impl->headers, "RTP-Info", NULL); +} + +static void rtsp_setup_reply(void *data, int status, const struct spa_dict *headers) +{ + struct impl *impl = data; + const char *str, *state = NULL, *s; + size_t len; + uint16_t control_port, timing_port; + + pw_log_info("reply %d", status); + + if ((str = spa_dict_lookup(headers, "Session")) == NULL) { + pw_log_error("missing Session header"); + return; + } + pw_properties_set(impl->headers, "Session", str); + + if ((str = spa_dict_lookup(headers, "Transport")) == NULL) { + pw_log_error("missing Transport header"); + return; + } + + impl->server_port = control_port = timing_port = 0; + while ((s = pw_split_walk(str, ";", &len, &state)) != NULL) { + if (spa_strstartswith(s, "server_port=")) { + impl->server_port = atoi(s + 12); + } + else if (spa_strstartswith(s, "control_port=")) { + control_port = atoi(s + 13); + } + else if (spa_strstartswith(s, "timing_port=")) { + timing_port = atoi(s + 12); + } + } + if (impl->server_port == 0 || control_port == 0 || timing_port == 0) { + pw_log_error("missing ports in Transport"); + return; + } + pw_log_info("server port:%u control:%u timing:%u", + impl->server_port, control_port, timing_port); + + if ((impl->server_fd = connect_udp_socket(impl, -1, impl->server_port)) <= 0) + return; + if ((impl->control_fd = connect_udp_socket(impl, impl->control_fd, control_port)) <= 0) + return; + if ((impl->timing_fd = connect_udp_socket(impl, impl->timing_fd, timing_port)) <= 0) + return; + + rtsp_do_record(impl); +} + +static void rtsp_do_setup(struct impl *impl) +{ + if (impl->protocol == 1) { + pw_properties_set(impl->headers, "Transport", + "RTP/AVP/TCP;unicast;interleaved=0-1;mode=record"); + } else { + impl->control_port = DEFAULT_UDP_CONTROL_PORT; + impl->timing_port = DEFAULT_UDP_TIMING_PORT; + + impl->control_fd = create_udp_socket(impl, &impl->control_port); + impl->timing_fd = create_udp_socket(impl, &impl->timing_port); + if (impl->control_fd < 0 || impl->timing_fd < 0) + goto error; + + pw_properties_setf(impl->headers, "Transport", + "RTP/AVP/UDP;unicast;interleaved=0-1;mode=record;" + "control_port=%u;timing_port=%u", + impl->control_port, impl->timing_port); + } + + pw_rtsp_client_send(impl->rtsp, "SETUP", &impl->headers->dict, + NULL, NULL, rtsp_setup_reply, impl); + + pw_properties_set(impl->headers, "Transport", NULL); + + return; + +error: + if (impl->control_fd > 0) + close(impl->control_fd); + impl->control_fd = -1; + if (impl->timing_fd > 0) + close(impl->timing_fd); + impl->timing_fd = -1; +} + +static void rtsp_announce_reply(void *data, int status, const struct spa_dict *headers) +{ + struct impl *impl = data; + + pw_log_info("reply %d", status); + + pw_properties_set(impl->headers, "Apple-Challenge", NULL); + + rtsp_do_setup(impl); +} + +static void base64_encode(const uint8_t *data, size_t len, char *enc, char pad) +{ + static const char tab[] = + "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + size_t i; + for (i = 0; i < len; i += 3) { + uint32_t v; + v = data[i+0] << 16; + v |= (i+1 < len ? data[i+1] : 0) << 8; + v |= (i+2 < len ? data[i+2] : 0); + *enc++ = tab[(v >> (3*6)) & 0x3f]; + *enc++ = tab[(v >> (2*6)) & 0x3f]; + *enc++ = i+1 < len ? tab[(v >> (1*6)) & 0x3f] : pad; + *enc++ = i+2 < len ? tab[(v >> (0*6)) & 0x3f] : pad; + } + *enc = '\0'; +} + +static size_t base64_decode(const char *data, size_t len, uint8_t *dec) +{ + uint8_t tab[] = { + 62, -1, -1, -1, 63, 52, 53, 54, 55, 56, + 57, 58, 59, 60, 61, -1, -1, -1, -1, -1, + -1, -1, 0, 1, 2, 3, 4, 5, 6, 7, + 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, + 18, 19, 20, 21, 22, 23, 24, 25, -1, -1, + -1, -1, -1, -1, 26, 27, 28, 29, 30, 31, + 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, + 42, 43, 44, 45, 46, 47, 48, 49, 50, 51 }; + size_t i, j; + for (i = 0, j = 0; i < len; i += 4) { + uint32_t v; + v = tab[data[i+0]-43] << (3*6); + v |= tab[data[i+1]-43] << (2*6); + v |= (data[i+2] == '=' ? 0 : tab[data[i+2]-43]) << (1*6); + v |= (data[i+3] == '=' ? 0 : tab[data[i+3]-43]); + dec[j++] = (v >> 16) & 0xff; + if (data[i+2] != '=') dec[j++] = (v >> 8) & 0xff; + if (data[i+3] != '=') dec[j++] = v & 0xff; + } + return j; +} + +static int rsa_encrypt(uint8_t *data, int len, uint8_t *res) +{ + RSA *rsa; + uint8_t modulus[256]; + uint8_t exponent[8]; + size_t size; + BIGNUM *n_bn = NULL; + BIGNUM *e_bn = NULL; + char n[] = + "59dE8qLieItsH1WgjrcFRKj6eUWqi+bGLOX1HL3U3GhC/j0Qg90u3sG/1CUtwC" + "5vOYvfDmFI6oSFXi5ELabWJmT2dKHzBJKa3k9ok+8t9ucRqMd6DZHJ2YCCLlDR" + "KSKv6kDqnw4UwPdpOMXziC/AMj3Z/lUVX1G7WSHCAWKf1zNS1eLvqr+boEjXuB" + "OitnZ/bDzPHrTOZz0Dew0uowxf/+sG+NCK3eQJVxqcaJ/vEHKIVd2M+5qL71yJ" + "Q+87X6oV3eaYvt3zWZYD6z5vYTcrtij2VZ9Zmni/UAaHqn9JdsBWLUEpVviYnh" + "imNVvYFZeCXg/IdTQ+x4IRdiXNv5hEew=="; + char e[] = "AQAB"; + + rsa = RSA_new(); + + size = base64_decode(n, strlen(n), modulus); + n_bn = BN_bin2bn(modulus, size, NULL); + + size = base64_decode(e, strlen(e), exponent); + e_bn = BN_bin2bn(exponent, size, NULL); + + RSA_set0_key(rsa, n_bn, e_bn, NULL); + + size = RSA_public_encrypt(len, data, res, rsa, RSA_PKCS1_OAEP_PADDING); + RSA_free(rsa); + return size; +} + +static void rtsp_do_announce(struct impl *impl) +{ + const char *host; + uint8_t rsakey[512]; + char key[512*2]; + char iv[16*2]; + int frames, i, ip_version; + char *sdp; + char local_ip[256]; + + host = pw_properties_get(impl->props, "raop.hostname"); + + if (impl->protocol == 0) + frames = FRAMES_PER_TCP_PACKET; + else + frames = FRAMES_PER_UDP_PACKET; + + pw_getrandom(impl->key, sizeof(impl->key), 0); + AES_set_encrypt_key(impl->key, 128, &impl->aes); + pw_getrandom(impl->iv, sizeof(impl->iv), 0); + + pw_log_info("aes %p", &impl->aes); + + i = rsa_encrypt(impl->key, 16, rsakey); + base64_encode(rsakey, i, key, '='); + base64_encode(impl->iv, 16, iv, '='); + + pw_rtsp_client_get_local_ip(impl->rtsp, &ip_version, + local_ip, sizeof(local_ip)); + +#if 0 + asprintf(&sdp, "v=0\r\n" + "o=iTunes %s 0 IN IP%d %s\r\n" + "s=iTunes\r\n" + "c=IN IP%d %s\r\n" + "t=0 0\r\n" + "m=audio 0 RTP/AVP 96\r\n" + "a=rtpmap:96 AppleLossless\r\n" + "a=fmtp:96 %d 0 16 40 10 14 2 255 0 0 44100\r\n", + impl->session_id, ip_version, local_ip, + ip_version, host, frames); +#else + asprintf(&sdp, "v=0\r\n" + "o=iTunes %s 0 IN IP%d %s\r\n" + "s=iTunes\r\n" + "c=IN IP%d %s\r\n" + "t=0 0\r\n" + "m=audio 0 RTP/AVP 96\r\n" + "a=rtpmap:96 AppleLossless\r\n" + "a=fmtp:96 %d 0 16 40 10 14 2 255 0 0 44100\r\n" + "a=rsaaeskey:%s\r\n" + "a=aesiv:%s\r\n", + impl->session_id, ip_version, local_ip, + ip_version, host, frames, key, iv); +#endif + + pw_rtsp_client_send(impl->rtsp, "ANNOUNCE", &impl->headers->dict, + "application/sdp", sdp, rtsp_announce_reply, impl); +} + + +static void rtsp_options_reply(void *data, int status, const struct spa_dict *headers) +{ + struct impl *impl = data; + pw_log_info("options"); + + rtsp_do_announce(impl); +} + +static void rtsp_connected(void *data) +{ + struct impl *impl = data; + uint32_t sci[2]; + uint8_t rac[16]; + char sac[16*4]; + + pw_log_info("connected"); + + pw_getrandom(sci, sizeof(sci), 0); + pw_properties_setf(impl->headers, "Client-Instance", + "%08x%08x", sci[0], sci[1]); + + pw_getrandom(rac, sizeof(rac), 0); + base64_encode(rac, sizeof(rac), sac, '\0'); + pw_properties_set(impl->headers, "Apple-Challenge", sac); + + pw_rtsp_client_send(impl->rtsp, "OPTIONS", &impl->headers->dict, + NULL, NULL, rtsp_options_reply, impl); +} + +static void rtsp_disconnected(void *data) +{ + pw_log_info("disconnected"); +} + +static void rtsp_error(void *data, int res) +{ + pw_log_info("error %d", res); +} + +static void rtsp_message(void *data, int status, int state, + const struct spa_dict *headers) +{ + const struct spa_dict_item *it; + pw_log_info("message %d %d", status, state); + spa_dict_for_each(it, headers) + pw_log_info(" %s: %s", it->key, it->value); + +} + +static const struct pw_rtsp_client_events rtsp_events = { + PW_VERSION_RTSP_CLIENT_EVENTS, + .connected = rtsp_connected, + .error = rtsp_error, + .disconnected = rtsp_disconnected, + .message = rtsp_message, +}; + +static int create_stream(struct impl *impl) +{ + int res; + uint32_t n_params; + const struct spa_pod *params[1]; + uint8_t buffer[1024]; + struct spa_pod_builder b; + const char *hostname, *port; + uint32_t session_id; + + impl->stream = pw_stream_new(impl->core, "example sink", impl->stream_props); + impl->stream_props = NULL; + + if (impl->stream == NULL) + return -errno; + + pw_stream_add_listener(impl->stream, + &impl->stream_listener, + &playback_stream_events, impl); + + n_params = 0; + spa_pod_builder_init(&b, buffer, sizeof(buffer)); + params[n_params++] = spa_format_audio_raw_build(&b, + SPA_PARAM_EnumFormat, &impl->info); + + if ((res = pw_stream_connect(impl->stream, + 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; + + impl->headers = pw_properties_new(NULL, NULL); + + impl->rtsp = pw_rtsp_client_new(impl->loop, NULL, 0); + if (impl->rtsp == NULL) + return -errno; + + pw_rtsp_client_add_listener(impl->rtsp, &impl->rtsp_listener, + &rtsp_events, impl); + + hostname = pw_properties_get(impl->props, "raop.hostname"); + port = pw_properties_get(impl->props, "raop.port"); + if (hostname == NULL || port == NULL) + return -EINVAL; + + pw_getrandom(&session_id, sizeof(session_id), 0); + asprintf(&impl->session_id, "%u", session_id); + + pw_rtsp_client_connect(impl->rtsp, hostname, atoi(port), impl->session_id); + + return 0; +} + +static void core_error(void *data, uint32_t id, int seq, int res, const char *message) +{ + struct impl *impl = 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) + unload_module(impl); +} + +static const struct pw_core_events core_events = { + PW_VERSION_CORE_EVENTS, + .error = core_error, +}; + +static void core_destroy(void *d) +{ + struct impl *impl = d; + spa_hook_remove(&impl->core_listener); + impl->core = NULL; + unload_module(impl); +} + +static const struct pw_proxy_events core_proxy_events = { + .destroy = core_destroy, +}; + +static void impl_destroy(struct impl *impl) +{ + if (impl->stream) + pw_stream_destroy(impl->stream); + if (impl->core && impl->do_disconnect) + pw_core_disconnect(impl->core); + + pw_properties_free(impl->stream_props); + pw_properties_free(impl->props); + + if (impl->work) + pw_work_queue_cancel(impl->work, impl, SPA_ID_INVALID); + free(impl); +} + +static void module_destroy(void *data) +{ + struct impl *impl = data; + impl->unloading = true; + spa_hook_remove(&impl->module_listener); + impl_destroy(impl); +} + +static const struct pw_impl_module_events module_events = { + PW_VERSION_IMPL_MODULE_EVENTS, + .destroy = module_destroy, +}; + +static inline uint32_t format_from_name(const char *name, size_t len) +{ + int i; + for (i = 0; spa_type_audio_format[i].name; i++) { + if (strncmp(name, spa_debug_type_short_name(spa_type_audio_format[i].name), len) == 0) + return spa_type_audio_format[i].type; + } + return SPA_AUDIO_FORMAT_UNKNOWN; +} + +static uint32_t channel_from_name(const char *name) +{ + int i; + for (i = 0; spa_type_audio_channel[i].name; i++) { + if (spa_streq(name, spa_debug_type_short_name(spa_type_audio_channel[i].name))) + return spa_type_audio_channel[i].type; + } + return SPA_AUDIO_CHANNEL_UNKNOWN; +} + +static void parse_position(struct spa_audio_info_raw *info, const char *val, size_t len) +{ + struct spa_json it[2]; + char v[256]; + + spa_json_init(&it[0], val, len); + if (spa_json_enter_array(&it[0], &it[1]) <= 0) + spa_json_init(&it[1], val, len); + + info->channels = 0; + while (spa_json_get_string(&it[1], v, sizeof(v)) > 0 && + info->channels < SPA_AUDIO_MAX_CHANNELS) { + info->position[info->channels++] = channel_from_name(v); + } +} + +static int parse_audio_info(struct impl *impl) +{ + struct pw_properties *props = impl->stream_props; + struct spa_audio_info_raw *info = &impl->info; + const char *str; + + spa_zero(*info); + + if ((str = pw_properties_get(props, PW_KEY_AUDIO_FORMAT)) == NULL) + str = DEFAULT_FORMAT; + info->format = format_from_name(str, strlen(str)); + switch (info->format) { + case SPA_AUDIO_FORMAT_S8: + case SPA_AUDIO_FORMAT_U8: + impl->frame_size = 1; + break; + case SPA_AUDIO_FORMAT_S16: + impl->frame_size = 2; + break; + case SPA_AUDIO_FORMAT_S24: + impl->frame_size = 3; + break; + case SPA_AUDIO_FORMAT_S24_32: + case SPA_AUDIO_FORMAT_S32: + case SPA_AUDIO_FORMAT_F32: + impl->frame_size = 4; + break; + case SPA_AUDIO_FORMAT_F64: + impl->frame_size = 8; + break; + default: + pw_log_error("unsupported format '%s'", str); + return -EINVAL; + } + info->rate = pw_properties_get_uint32(props, PW_KEY_AUDIO_RATE, DEFAULT_RATE); + if (info->rate == 0) { + pw_log_error("invalid rate '%s'", str); + return -EINVAL; + } + if ((str = pw_properties_get(props, PW_KEY_AUDIO_CHANNELS)) == NULL) + str = DEFAULT_CHANNELS; + info->channels = atoi(str); + if ((str = pw_properties_get(props, SPA_KEY_AUDIO_POSITION)) == NULL) + str = DEFAULT_POSITION; + parse_position(info, str, strlen(str)); + if (info->channels == 0) { + pw_log_error("invalid channels '%s'", str); + return -EINVAL; + } + impl->frame_size *= info->channels; + + return 0; +} + +static void copy_props(struct impl *impl, struct pw_properties *props, const char *key) +{ + const char *str; + if ((str = pw_properties_get(props, key)) != NULL) { + if (pw_properties_get(impl->stream_props, key) == NULL) + pw_properties_set(impl->stream_props, key, str); + } +} + +SPA_EXPORT +int pipewire__module_init(struct pw_impl_module *module, const char *args) +{ + struct pw_context *context = pw_impl_module_get_context(module); + struct pw_properties *props = NULL; + uint32_t id = pw_global_get_id(pw_impl_module_get_global(module)); + struct impl *impl; + const char *str; + int res; + + PW_LOG_TOPIC_INIT(mod_topic); + + impl = calloc(1, sizeof(struct impl)); + if (impl == NULL) + return -errno; + + pw_log_debug("module %p: new %s", impl, args); + + if (args == NULL) + args = ""; + + props = pw_properties_new_string(args); + if (props == NULL) { + res = -errno; + pw_log_error( "can't create properties: %m"); + goto error; + } + impl->props = props; + + impl->stream_props = pw_properties_new(NULL, NULL); + if (impl->stream_props == NULL) { + res = -errno; + pw_log_error( "can't create properties: %m"); + goto error; + } + + impl->module = module; + impl->context = context; + impl->loop = pw_context_get_main_loop(context); + impl->work = pw_context_get_work_queue(context); + if (impl->work == NULL) { + res = -errno; + pw_log_error( "can't get work queue: %m"); + goto error; + } + + if (pw_properties_get(props, PW_KEY_NODE_GROUP) == NULL) + pw_properties_set(props, PW_KEY_NODE_GROUP, "pipewire.dummy"); + if (pw_properties_get(props, PW_KEY_NODE_VIRTUAL) == NULL) + pw_properties_set(props, PW_KEY_NODE_VIRTUAL, "true"); + + if (pw_properties_get(props, PW_KEY_MEDIA_CLASS) == NULL) + pw_properties_set(props, PW_KEY_MEDIA_CLASS, "Audio/Sink"); + + if (pw_properties_get(props, PW_KEY_NODE_NAME) == NULL) + pw_properties_setf(props, PW_KEY_NODE_NAME, "raop-sink-%u", id); + if (pw_properties_get(props, PW_KEY_NODE_DESCRIPTION) == NULL) + pw_properties_set(props, PW_KEY_NODE_DESCRIPTION, + pw_properties_get(props, PW_KEY_NODE_NAME)); + + if ((str = pw_properties_get(props, "stream.props")) != NULL) + pw_properties_update_string(impl->stream_props, str, strlen(str)); + + copy_props(impl, props, PW_KEY_AUDIO_FORMAT); + copy_props(impl, props, PW_KEY_AUDIO_RATE); + copy_props(impl, props, PW_KEY_AUDIO_CHANNELS); + copy_props(impl, props, SPA_KEY_AUDIO_POSITION); + copy_props(impl, props, PW_KEY_NODE_NAME); + copy_props(impl, props, PW_KEY_NODE_DESCRIPTION); + copy_props(impl, props, PW_KEY_NODE_GROUP); + copy_props(impl, props, PW_KEY_NODE_LATENCY); + copy_props(impl, props, PW_KEY_NODE_VIRTUAL); + copy_props(impl, props, PW_KEY_MEDIA_CLASS); + + if ((res = parse_audio_info(impl)) < 0) { + pw_log_error( "can't parse audio format"); + goto error; + } + + impl->core = pw_context_get_object(impl->context, PW_TYPE_INTERFACE_Core); + if (impl->core == NULL) { + str = pw_properties_get(props, PW_KEY_REMOTE_NAME); + impl->core = pw_context_connect(impl->context, + pw_properties_new( + PW_KEY_REMOTE_NAME, str, + NULL), + 0); + impl->do_disconnect = true; + } + if (impl->core == NULL) { + res = -errno; + pw_log_error("can't connect: %m"); + goto error; + } + + pw_proxy_add_listener((struct pw_proxy*)impl->core, + &impl->core_proxy_listener, + &core_proxy_events, impl); + pw_core_add_listener(impl->core, + &impl->core_listener, + &core_events, impl); + + if ((res = create_stream(impl)) < 0) + goto error; + + pw_impl_module_add_listener(module, &impl->module_listener, &module_events, impl); + + pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_props)); + + return 0; + +error: + impl_destroy(impl); + return res; +} diff --git a/src/modules/module-raop/rtsp-client.c b/src/modules/module-raop/rtsp-client.c new file mode 100644 index 000000000..97860428f --- /dev/null +++ b/src/modules/module-raop/rtsp-client.c @@ -0,0 +1,511 @@ +/* PipeWire + * + * Copyright © 2021 Wim Taymans + * + * 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 "rtsp-client.h" + +enum { + STATE_INIT = 0, + STATE_CONNECTING, + STATE_CONNECTED, +}; + +#define pw_rtsp_client_emit(o,m,v,...) spa_hook_list_call(&o->listener_list, struct pw_rtsp_client_events, m, v, ##__VA_ARGS__) +#define pw_rtsp_client_emit_destroy(c) pw_rtsp_client_emit(c, destroy, 0) +#define pw_rtsp_client_emit_connected(c) pw_rtsp_client_emit(c, connected, 0) +#define pw_rtsp_client_emit_disconnected(c) pw_rtsp_client_emit(c, disconnected, 0) +#define pw_rtsp_client_emit_error(c,r) pw_rtsp_client_emit(c, error, 0, r) +#define pw_rtsp_client_emit_message(c,...) pw_rtsp_client_emit(c, message, 0, __VA_ARGS__) + +struct message { + struct spa_list link; + void *data; + size_t len; + size_t offset; + int cseq; + void (*reply) (void *user_data, int status, const struct spa_dict *headers); + void *user_data; +}; + +struct pw_rtsp_client { + struct pw_loop *loop; + struct pw_properties *props; + + struct spa_hook_list listener_list; + + char *session_id; + char *url; + + union { + struct sockaddr sa; + struct sockaddr_in in; + struct sockaddr_in6 in6; + } local_addr; + + int state; + struct spa_source *source; + unsigned int need_flush:1; + unsigned int wait_status:1; + + int status; + char line_buf[1024]; + size_t line_pos; + struct pw_properties *headers; + + char *session; + int cseq; + + struct spa_list messages; + struct spa_list pending; + + void *user_data; +}; + +struct pw_rtsp_client *pw_rtsp_client_new(struct pw_loop *main_loop, + struct pw_properties *props, + size_t user_data_size) +{ + struct pw_rtsp_client *client; + + client = calloc(1, sizeof(*client) + user_data_size); + if (client == NULL) + return NULL; + + client->loop = main_loop; + client->props = props; + if (user_data_size > 0) + client->user_data = SPA_PTROFF(client, sizeof(*client), void); + client->state = STATE_INIT; + + spa_list_init(&client->messages); + spa_list_init(&client->pending); + spa_hook_list_init(&client->listener_list); + client->headers = pw_properties_new(NULL, NULL); + + pw_log_info("new client %p", client); + + return client; +} + +void pw_rtsp_client_destroy(struct pw_rtsp_client *client) +{ + spa_hook_list_clean(&client->listener_list); + free(client); +} + +void *pw_rtsp_client_get_user_data(struct pw_rtsp_client *client) +{ + return client->user_data; +} + +void pw_rtsp_client_add_listener(struct pw_rtsp_client *client, + struct spa_hook *listener, + const struct pw_rtsp_client_events *events, void *data) +{ + spa_hook_list_append(&client->listener_list, listener, events, data); +} + +const struct pw_properties *pw_rtsp_client_get_properties(struct pw_rtsp_client *client) +{ + return client->props; +} + +int pw_rtsp_client_get_local_ip(struct pw_rtsp_client *client, + int *version, char *ip, size_t len) +{ + if (client->local_addr.sa.sa_family == AF_INET) { + *version = 4; + if (ip) + inet_ntop(client->local_addr.sa.sa_family, + &client->local_addr.in.sin_addr, ip, len); + } else if (client->local_addr.sa.sa_family == AF_INET6) { + *version = 6; + if (ip) + inet_ntop(client->local_addr.sa.sa_family, + &client->local_addr.in6.sin6_addr, + ip, len); + } else + return -EIO; + return 0; +} + +static int handle_connect(struct pw_rtsp_client *client, int fd) +{ + int res, ip_version; + socklen_t len; + char local_ip[INET6_ADDRSTRLEN]; + + len = sizeof(res); + if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &res, &len) < 0) { + pw_log_error("getsockopt: %m"); + return -errno; + } + if (res != 0) + return -res; + + len = sizeof(client->local_addr.sa); + if (getsockname(fd, &client->local_addr.sa, &len) < 0) + return -errno; + + if ((res = pw_rtsp_client_get_local_ip(client, &ip_version, + local_ip, sizeof(local_ip))) < 0) + return res; + + if (ip_version == 4) + asprintf(&client->url, "rtsp://%s/%s", local_ip, client->session_id); + else + asprintf(&client->url, "rtsp://[%s]/%s", local_ip, client->session_id); + + pw_log_info("connected local ip %s", local_ip); + + client->state = STATE_CONNECTED; + client->wait_status = true; + pw_rtsp_client_emit_connected(client); + + return 0; +} + +static int read_line(struct pw_rtsp_client *client, char **buf) +{ + int res; + + while (true) { + uint8_t c; + + res = read(client->source->fd, &c, 1); + if (res == 0) + return -EPIPE; + if (res < 0) { + if (res == EAGAIN) + return 0; + return -errno; + } + if (c == '\n') { + client->line_buf[client->line_pos] = '\0'; + client->line_pos = 0; + if (buf) + *buf = client->line_buf; + return 1; + } + if (c == '\r') + continue; + if (client->line_pos < sizeof(client->line_buf) - 1) + client->line_buf[client->line_pos++] = c; + client->line_buf[client->line_pos] = '\0'; + } + return 0; +} + +static struct message *find_pending(struct pw_rtsp_client *client, int cseq) +{ + struct message *msg; + spa_list_for_each(msg, &client->pending, link) { + if (msg->cseq == cseq) + return msg; + } + return NULL; +} + +static int process_input(struct pw_rtsp_client *client) +{ + char *buf = NULL; + int res; + + if ((res = read_line(client, &buf)) <= 0) + return res; + + pw_log_debug("%s", buf); + + if (client->wait_status) { + const char *state = NULL, *s; + size_t len; + + pw_log_info("status: %s", buf); + + s = pw_split_walk(buf, " ", &len, &state); + if (!spa_strstartswith(s, "RTSP/")) + goto error; + + s = pw_split_walk(buf, " ", &len, &state); + if (s == NULL) + goto error; + + client->status = atoi(s); + + s = pw_split_walk(buf, " ", &len, &state); + if (s == NULL) + goto error; + + client->wait_status = false; + pw_properties_clear(client->headers); + } else { + if (strlen(buf) == 0) { + int cseq; + struct message *msg; + const struct spa_dict_item *it; + + spa_dict_for_each(it, &client->headers->dict) + pw_log_info(" %s: %s", it->key, it->value); + + cseq = pw_properties_get_int32(client->headers, "CSeq", 0); + + if ((msg = find_pending(client, cseq)) != NULL) { + msg->reply(msg->user_data, client->status, &client->headers->dict); + spa_list_remove(&msg->link); + free(msg); + } else { + pw_rtsp_client_emit_message(client, client->status, + client->state, &client->headers->dict); + } + client->wait_status = true; + } else { + char *key, *value; + + key = buf; + value = strstr(buf, ":"); + if (value == NULL) + goto error; + *value++ = '\0'; + pw_properties_set(client->headers, key, value); + } + } + return 0; +error: + return -EPROTO; +} + +static int flush_output(struct pw_rtsp_client *client) +{ + int res; + + client->need_flush = false; + + while (true) { + struct message *msg; + void *data; + size_t size; + + if (spa_list_is_empty(&client->messages)) + break; + + msg = spa_list_first(&client->messages, struct message, link); + + if (msg->offset < msg->len) { + data = SPA_PTROFF(msg->data, msg->offset, void); + size = msg->len - msg->offset; + } else { + pw_log_info("sent: %s", (char *)msg->data); + spa_list_remove(&msg->link); + if (msg->reply != NULL) + spa_list_append(&client->pending, &msg->link); + else + free(msg); + continue; + } + + while (true) { + res = send(client->source->fd, data, size, MSG_NOSIGNAL | MSG_DONTWAIT); + if (res < 0) { + res = -errno; + if (res == -EINTR) + continue; + if (res != -EAGAIN && res != -EWOULDBLOCK) + pw_log_warn("client %p: send %zu, error %d: %m", + client, size, res); + return res; + } + msg->offset += res; + break; + } + } + return 0; +} + +static void +on_source_io(void *data, int fd, uint32_t mask) +{ + struct pw_rtsp_client *client = data; + int res; + + if (mask & (SPA_IO_ERR | SPA_IO_HUP)) { + res = -EPIPE; + goto error; + } + if (mask & SPA_IO_IN) { + if ((res = process_input(client)) < 0) + goto error; + } + if (mask & SPA_IO_OUT || client->need_flush) { + if (client->state == STATE_CONNECTING) { + if ((res = handle_connect(client, fd)) < 0) + goto error; + } + res = flush_output(client); + if (res >= 0) { + pw_loop_update_io(client->loop, client->source, + client->source->mask & ~SPA_IO_OUT); + } else if (res != -EAGAIN) + goto error; + } +done: + return; +error: + pw_log_error("%p: got connection error %d (%s)", client, res, spa_strerror(res)); + pw_rtsp_client_emit_error(client, res); + pw_rtsp_client_disconnect(client); + goto done; +} + +int pw_rtsp_client_connect(struct pw_rtsp_client *client, + const char *hostname, uint16_t port, const char *session_id) +{ + struct sockaddr_in my_addr, dest_addr; + struct hostent *h; + int fd; + + if (client->source != NULL) + pw_rtsp_client_disconnect(client); + + pw_log_info("%p: connect %s:%u", client, hostname, port); + fd = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0); + if (fd < 0) + return -errno; + + spa_zero(my_addr); + my_addr.sin_addr.s_addr = htonl(INADDR_ANY); + my_addr.sin_family = AF_INET; + my_addr.sin_port = 0; + + if (bind(fd, (struct sockaddr *) &my_addr, sizeof(my_addr)) < 0) { + pw_log_error("%p: bind failed: %m", client); + close(fd); + return -errno; + } + + h = gethostbyname(hostname); + if (h != NULL) { + dest_addr.sin_family = h->h_addrtype; + memcpy((char*) &dest_addr.sin_addr.s_addr, h->h_addr_list[0], h->h_length); + } else { + dest_addr.sin_family = AF_INET; + if ((dest_addr.sin_addr.s_addr = inet_addr(hostname)) == 0xFFFFFFFF) + return -1; + } + dest_addr.sin_port = htons(port); + + if (connect(fd, (struct sockaddr *)&dest_addr, sizeof(struct sockaddr)) < 0) { + if (errno != EINPROGRESS) { + pw_log_error("%p: connect failed: %m", client); + close(fd); + return -errno; + } + } + + client->source = pw_loop_add_io(client->loop, fd, + SPA_IO_IN | SPA_IO_OUT | SPA_IO_HUP | SPA_IO_ERR, + true, on_source_io, client); + if (client->source == NULL) { + pw_log_error("%p: source create failed: %m", client); + close(fd); + return -errno; + } + client->state = STATE_CONNECTING; + free(client->session_id); + client->session_id = strdup(session_id); + pw_log_info("%p: connecting", client); + + return 0; +} + +int pw_rtsp_client_disconnect(struct pw_rtsp_client *client) +{ + if (client->source == NULL) + return 0; + + pw_loop_destroy_source(client->loop, client->source); + client->source = NULL; + client->state = STATE_INIT; + pw_rtsp_client_emit_disconnected(client); + return 0; +} + +int pw_rtsp_client_send(struct pw_rtsp_client *client, + const char *cmd, const struct spa_dict *headers, + const char *content_type, const char *content, + void (*reply) (void *user_data, int status, const struct spa_dict *headers), + void *user_data) +{ + FILE *f; + size_t len; + const struct spa_dict_item *it; + struct message *msg; + int cseq; + + f = open_memstream((char**)&msg, &len); + + fseek(f, sizeof(*msg), SEEK_SET); + + cseq = ++client->cseq; + + fprintf(f, "%s %s RTSP/1.0\r\n", cmd, client->url); + fprintf(f, "CSeq: %d\r\n", cseq); + + if (headers != NULL) { + spa_dict_for_each(it, headers) + fprintf(f, "%s: %s\r\n", it->key, it->value); + } + if (content_type != NULL && content != NULL) { + fprintf(f, "Content-Type: %s\r\nContent-Length: %d\r\n", + content_type, (int)strlen(content)); + } + fprintf(f, "\r\n"); + + if (content_type && content) + fprintf(f, "%s", content); + + fclose(f); + + msg->data = SPA_PTROFF(msg, sizeof(*msg), void); + msg->len = len - sizeof(*msg); + msg->offset = 0; + msg->reply = reply; + msg->user_data = user_data; + msg->cseq = cseq; + + spa_list_append(&client->messages, &msg->link); + + client->need_flush = true; + if (client->source && !(client->source->mask & SPA_IO_OUT)) { + pw_loop_update_io(client->loop, client->source, + client->source->mask | SPA_IO_OUT); + } + return 0; +} diff --git a/src/modules/module-raop/rtsp-client.h b/src/modules/module-raop/rtsp-client.h new file mode 100644 index 000000000..bf62ba2df --- /dev/null +++ b/src/modules/module-raop/rtsp-client.h @@ -0,0 +1,88 @@ +/* PipeWire + * + * Copyright © 2021 Wim Taymans + * + * 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. + */ + +#ifndef PIPEWIRE_RTSP_CLIENT_H +#define PIPEWIRE_RTSP_CLIENT_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +#include + +struct pw_rtsp_client; + +struct pw_rtsp_client_events { +#define PW_VERSION_RTSP_CLIENT_EVENTS 0 + uint32_t version; + + void (*destroy) (void *data); + + void (*connected) (void *data); + void (*error) (void *data, int res); + void (*disconnected) (void *data); + + void (*message) (void *data, int status, int state, + const struct spa_dict *headers); + +}; + +struct pw_rtsp_client * pw_rtsp_client_new(struct pw_loop *main_loop, + struct pw_properties *props, + size_t user_data_size); + +void pw_rtsp_client_destroy(struct pw_rtsp_client *client); + +void *pw_rtsp_client_get_user_data(struct pw_rtsp_client *client); + +void pw_rtsp_client_add_listener(struct pw_rtsp_client *client, + struct spa_hook *listener, + const struct pw_rtsp_client_events *events, void *data); + +const struct pw_properties *pw_rtsp_client_get_properties(struct pw_rtsp_client *client); + +void *pw_rtsp_client_get_user_data(struct pw_rtsp_client *client); + + +int pw_rtsp_client_connect(struct pw_rtsp_client *client, + const char *hostname, uint16_t port, const char *session_id); +int pw_rtsp_client_disconnect(struct pw_rtsp_client *client); + +int pw_rtsp_client_get_local_ip(struct pw_rtsp_client *client, + int *version, char *ip, size_t len); + +int pw_rtsp_client_send(struct pw_rtsp_client *client, + const char *cmd, const struct spa_dict *headers, + const char *content_type, const char *content, + void (*reply) (void *user_data, int status, const struct spa_dict *headers), + void *user_data); + + +#ifdef __cplusplus +} +#endif + +#endif /* PIPEWIRE_RTSP_CLIENT_H */