mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-10-29 05:40:27 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

raop: add beginnings of RAOP protocol

Browse source
This commit is contained in:
Wim Taymans 2021-11-11 11:42:16 +01:00
parent c21536eec8
commit 9223fc2885
6 changed files with 2235 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

3
meson.build
View file

@ -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()

29
src/modules/meson.build
View file

@ -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',

541
src/modules/module-raop-discover.c Normal file
View file

@ -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 <string.h>
#include <stdio.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include "config.h"
#include <spa/utils/result.h>
#include <spa/utils/string.h>
#include <spa/utils/json.h>
#include <pipewire/impl.h>
#include <pipewire/private.h>
#include <pipewire/i18n.h>
#include <avahi-client/lookup.h>
#include <avahi-common/error.h>
#include <avahi-common/malloc.h>
#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 <wim.taymans@gmail.com>" },
{ 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;
}

1063
src/modules/module-raop-sink.c Normal file
View file

File diff suppressed because it is too large Load diff

511
src/modules/module-raop/rtsp-client.c Normal file
View file

@ -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 <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <spa/utils/result.h>
#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;
}

88
src/modules/module-raop/rtsp-client.h Normal file
View file

@ -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 <stdarg.h>
#include <pipewire/pipewire.h>
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 */