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

modules-vban: add vban sender and receiver

Browse source 
Fixes #3380
This commit is contained in:
Wim Taymans 2023-07-25 18:36:48 +02:00
parent 938bbc3039
commit ce8f2c5d10
8 changed files with 2004 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

2
doc/pipewire-modules.dox
View file

@ -86,6 +86,8 @@ List of known modules:
- \subpage page_module_rtp_session
- \subpage page_module_rt
- \subpage page_module_session_manager
- \subpage page_module_vban_recv
- \subpage page_module_vban_send
- \subpage page_module_x11_bell
- \subpage page_module_zeroconf_discover

22
src/modules/meson.build
View file

@ -38,6 +38,8 @@ module_sources = [
'module-rtp-session.c',
'module-rtp-source.c',
'module-rtp-sink.c',
'module-vban-recv.c',
'module-vban-send.c',
'module-session-manager.c',
'module-zeroconf-discover.c',
'module-roc-source.c',
@ -659,6 +661,26 @@ pipewire_module_rtp_sap = shared_library('pipewire-module-rtp-sap',
dependencies : [mathlib, dl_lib, rt_lib, pipewire_dep],
)
pipewire_module_vban_send = shared_library('pipewire-module-vban-send',
[ 'module-vban-send.c',
'module-vban/stream.c' ],
include_directories : [configinc],
install : true,
install_dir : modules_install_dir,
install_rpath: modules_install_dir,
dependencies : [mathlib, dl_lib, rt_lib, pipewire_dep],
)
pipewire_module_vban_recv = shared_library('pipewire-module-vban-recv',
[ 'module-vban-recv.c',
'module-vban/stream.c' ],
include_directories : [configinc],
install : true,
install_dir : modules_install_dir,
install_rpath: modules_install_dir,
dependencies : [mathlib, dl_lib, rt_lib, pipewire_dep],
)
build_module_roc = roc_dep.found()
if build_module_roc
pipewire_module_roc_sink = shared_library('pipewire-module-roc-sink',

572
src/modules/module-vban-recv.c Normal file
View file

@ -0,0 +1,572 @@
/* PipeWire */
/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
/* SPDX-License-Identifier: MIT */
#include "config.h"
#include <limits.h>
#include <string.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <net/if.h>
#include <ctype.h>
#include <spa/utils/hook.h>
#include <spa/utils/result.h>
#include <spa/utils/ringbuffer.h>
#include <spa/utils/defs.h>
#include <spa/utils/dll.h>
#include <spa/utils/json.h>
#include <spa/param/audio/format-utils.h>
#include <spa/control/control.h>
#include <spa/debug/types.h>
#include <spa/debug/mem.h>
#include <pipewire/pipewire.h>
#include <pipewire/impl.h>
#include <module-vban/stream.h>
#ifdef __FreeBSD__
#define ifr_ifindex ifr_index
#endif
/** \page page_module_vban_recv PipeWire Module: VBAN receiver
*
* The `vban-recv` module creates a PipeWire source that receives audio
* and midi [VBAN](https://vb-audio.com) packets.
*
* ## Module Options
*
* Options specific to the behavior of this module
*
* - `local.ifname = <str>`: interface name to use
* - `source.ip = <str>`: the source ip address, default 127.0.0.1
* - `source.port = <int>`: the source port
* - `node.always-process = <bool>`: true to receive even when not running
* - `sess.latency.msec = <str>`: target network latency in milliseconds, default 100
* - `sess.ignore-ssrc = <bool>`: ignore SSRC, default false
* - `sess.media = <string>`: the media type audio|midi|opus, default audio
* - `stream.props = {}`: properties to be passed to the stream
*
* ## General options
*
* Options with well-known behavior:
*
* - \ref PW_KEY_REMOTE_NAME
* - \ref PW_KEY_AUDIO_FORMAT
* - \ref PW_KEY_AUDIO_RATE
* - \ref PW_KEY_AUDIO_CHANNELS
* - \ref SPA_KEY_AUDIO_POSITION
* - \ref PW_KEY_MEDIA_NAME
* - \ref PW_KEY_MEDIA_CLASS
* - \ref PW_KEY_NODE_NAME
* - \ref PW_KEY_NODE_DESCRIPTION
* - \ref PW_KEY_NODE_GROUP
* - \ref PW_KEY_NODE_LATENCY
* - \ref PW_KEY_NODE_VIRTUAL
*
* ## Example configuration
*\code{.unparsed}
* context.modules = [
* { name = libpipewire-module-vban-recv
* args = {
* #local.ifname = eth0
* #source.ip = 127.0.0.1
* #source.port = 6980
* sess.latency.msec = 100
* #sess.ignore-ssrc = false
* #node.always-process = false
* #sess.media = "audio"
* #audio.format = "S16LE"
* #audio.rate = 44100
* #audio.channels = 2
* #audio.position = [ FL FR ]
* stream.props = {
* #media.class = "Audio/Source"
* node.name = "vban-receiver"
* }
* }
* }
* ]
*\endcode
*
* \since 0.3.76
*/
#define NAME "vban-recv"
PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
#define PW_LOG_TOPIC_DEFAULT mod_topic
#define DEFAULT_CLEANUP_SEC 60
#define DEFAULT_SOURCE_IP "127.0.0.1"
#define DEFAULT_SOURCE_PORT 6980
#define USAGE "( local.ifname=<local interface name to use> ) " \
"( source.ip=<source IP address, default:"DEFAULT_SOURCE_IP"> ) " \
"( source.port=<int, source port, default:"SPA_STRINGIFY(DEFAULT_SOURCE_PORT)"> " \
"( sess.latency.msec=<target network latency, default "SPA_STRINGIFY(DEFAULT_SESS_LATENCY)"> ) "\
"( sess.media=<string, the media type audio|midi, default audio> ) " \
"( audio.format=<format, default:"DEFAULT_FORMAT"> ) " \
"( audio.rate=<sample rate, default:"SPA_STRINGIFY(DEFAULT_RATE)"> ) " \
"( audio.channels=<number of channels, default:"SPA_STRINGIFY(DEFAULT_CHANNELS)"> ) " \
"( audio.position=<channel map, default:"DEFAULT_POSITION"> ) " \
"( stream.props= { key=value ... } ) "
static const struct spa_dict_item module_info[] = {
{ PW_KEY_MODULE_AUTHOR, "Wim Taymans <wim.taymans@gmail.com>" },
{ PW_KEY_MODULE_DESCRIPTION, "VBAN Receiver" },
{ PW_KEY_MODULE_USAGE, USAGE },
{ PW_KEY_MODULE_VERSION, PACKAGE_VERSION },
};
struct impl {
struct pw_impl_module *module;
struct spa_hook module_listener;
struct pw_properties *props;
struct pw_context *context;
struct pw_loop *loop;
struct pw_loop *data_loop;
struct pw_core *core;
struct spa_hook core_listener;
struct spa_hook core_proxy_listener;
unsigned int do_disconnect:1;
char *ifname;
bool always_process;
uint32_t cleanup_interval;
struct spa_source *timer;
struct pw_properties *stream_props;
struct vban_stream *stream;
uint16_t src_port;
struct sockaddr_storage src_addr;
socklen_t src_len;
struct spa_source *source;
unsigned receiving:1;
};
static void
on_vban_io(void *data, int fd, uint32_t mask)
{
struct impl *impl = data;
ssize_t len;
uint8_t buffer[2048];
if (mask & SPA_IO_IN) {
if ((len = recv(fd, buffer, sizeof(buffer), 0)) < 0)
goto receive_error;
if (len < 12)
goto short_packet;
if (SPA_LIKELY(impl->stream))
vban_stream_receive_packet(impl->stream, buffer, len);
impl->receiving = true;
}
return;
receive_error:
pw_log_warn("recv error: %m");
return;
short_packet:
pw_log_warn("short packet received");
return;
}
static int parse_address(const char *address, uint16_t port,
struct sockaddr_storage *addr, socklen_t *len)
{
struct sockaddr_in *sa4 = (struct sockaddr_in*)addr;
struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)addr;
if (inet_pton(AF_INET, address, &sa4->sin_addr) > 0) {
sa4->sin_family = AF_INET;
sa4->sin_port = htons(port);
*len = sizeof(*sa4);
} else if (inet_pton(AF_INET6, address, &sa6->sin6_addr) > 0) {
sa6->sin6_family = AF_INET6;
sa6->sin6_port = htons(port);
*len = sizeof(*sa6);
} else
return -EINVAL;
return 0;
}
static int make_socket(const struct sockaddr* sa, socklen_t salen, char *ifname)
{
int af, fd, val, res;
struct ifreq req;
af = sa->sa_family;
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;
}
spa_zero(req);
if (ifname) {
snprintf(req.ifr_name, sizeof(req.ifr_name), "%s", ifname);
res = ioctl(fd, SIOCGIFINDEX, &req);
if (res < 0)
pw_log_warn("SIOCGIFINDEX %s failed: %m", ifname);
}
res = 0;
if (af == AF_INET) {
static const uint32_t ipv4_mcast_mask = 0xe0000000;
struct sockaddr_in *sa4 = (struct sockaddr_in*)sa;
if ((ntohl(sa4->sin_addr.s_addr) & ipv4_mcast_mask) == ipv4_mcast_mask) {
struct ip_mreqn mr4;
memset(&mr4, 0, sizeof(mr4));
mr4.imr_multiaddr = sa4->sin_addr;
mr4.imr_ifindex = req.ifr_ifindex;
res = setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mr4, sizeof(mr4));
} else {
sa4->sin_addr.s_addr = INADDR_ANY;
}
} else if (af == AF_INET6) {
struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)sa;
if (sa6->sin6_addr.s6_addr[0] == 0xff) {
struct ipv6_mreq mr6;
memset(&mr6, 0, sizeof(mr6));
mr6.ipv6mr_multiaddr = sa6->sin6_addr;
mr6.ipv6mr_interface = req.ifr_ifindex;
res = setsockopt(fd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mr6, sizeof(mr6));
} else {
sa6->sin6_addr = in6addr_any;
}
} else {
res = -EINVAL;
goto error;
}
if (res < 0) {
res = -errno;
pw_log_error("join mcast failed: %m");
goto error;
}
if (bind(fd, sa, salen) < 0) {
res = -errno;
pw_log_error("bind() failed: %m");
goto error;
}
return fd;
error:
close(fd);
return res;
}
static int stream_start(struct impl *impl)
{
int fd;
if (impl->source != NULL)
return 0;
pw_log_info("starting VBAN listener");
if ((fd = make_socket((const struct sockaddr *)&impl->src_addr,
impl->src_len, impl->ifname)) < 0) {
pw_log_error("failed to create socket: %m");
return fd;
}
impl->source = pw_loop_add_io(impl->data_loop, fd,
SPA_IO_IN, true, on_vban_io, impl);
if (impl->source == NULL) {
pw_log_error("can't create io source: %m");
close(fd);
return -errno;
}
return 0;
}
static void stream_stop(struct impl *impl)
{
if (!impl->source)
return;
pw_log_info("stopping VBAN listener");
pw_loop_destroy_source(impl->data_loop, impl->source);
impl->source = NULL;
}
static void stream_destroy(void *d)
{
struct impl *impl = d;
impl->stream = NULL;
}
static void stream_state_changed(void *data, bool started, const char *error)
{
struct impl *impl = data;
if (error) {
pw_log_error("stream error: %s", error);
pw_impl_module_schedule_destroy(impl->module);
} else if (started) {
if ((errno = -stream_start(impl)) < 0)
pw_log_error("failed to start VBAN stream: %m");
} else {
if (!impl->always_process)
stream_stop(impl);
}
}
static const struct vban_stream_events stream_events = {
VBAN_VERSION_STREAM_EVENTS,
.destroy = stream_destroy,
.state_changed = stream_state_changed,
};
static void on_timer_event(void *data, uint64_t expirations)
{
struct impl *impl = data;
if (!impl->receiving) {
pw_log_info("timeout, inactive VBAN source");
//pw_impl_module_schedule_destroy(impl->module);
} else {
pw_log_debug("timeout, keeping active VBAN source");
}
impl->receiving = false;
}
static void core_destroy(void *d)
{
struct impl *impl = d;
spa_hook_remove(&impl->core_listener);
impl->core = NULL;
pw_impl_module_schedule_destroy(impl->module);
}
static const struct pw_proxy_events core_proxy_events = {
.destroy = core_destroy,
};
static void impl_destroy(struct impl *impl)
{
if (impl->stream)
vban_stream_destroy(impl->stream);
if (impl->source)
pw_loop_destroy_source(impl->data_loop, impl->source);
if (impl->core && impl->do_disconnect)
pw_core_disconnect(impl->core);
if (impl->timer)
pw_loop_destroy_source(impl->loop, impl->timer);
pw_properties_free(impl->stream_props);
pw_properties_free(impl->props);
free(impl->ifname);
free(impl);
}
static void module_destroy(void *d)
{
struct impl *impl = d;
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 void on_core_error(void *d, uint32_t id, int seq, int res, const char *message)
{
struct impl *impl = d;
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)
pw_impl_module_schedule_destroy(impl->module);
}
static const struct pw_core_events core_events = {
PW_VERSION_CORE_EVENTS,
.error = on_core_error,
};
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 impl *impl;
const char *str, *sess_name;
struct timespec value, interval;
struct pw_properties *props, *stream_props;
int res = 0;
PW_LOG_TOPIC_INIT(mod_topic);
impl = calloc(1, sizeof(struct impl));
if (impl == NULL)
return -errno;
if (args == NULL)
args = "";
props = impl->props = pw_properties_new_string(args);
stream_props = impl->stream_props = pw_properties_new(NULL, NULL);
if (props == NULL || stream_props == NULL) {
res = -errno;
pw_log_error( "can't create properties: %m");
goto out;
}
impl->module = module;
impl->context = context;
impl->loop = pw_context_get_main_loop(context);
impl->data_loop = pw_data_loop_get_loop(pw_context_get_data_loop(context));
if ((sess_name = pw_properties_get(props, "sess.name")) == NULL)
sess_name = pw_get_host_name();
if (pw_properties_get(props, PW_KEY_NODE_NAME) == NULL)
pw_properties_setf(props, PW_KEY_NODE_NAME, "vban_session.%s", sess_name);
if (pw_properties_get(props, PW_KEY_NODE_DESCRIPTION) == NULL)
pw_properties_setf(props, PW_KEY_NODE_DESCRIPTION, "%s", sess_name);
if (pw_properties_get(props, PW_KEY_MEDIA_NAME) == NULL)
pw_properties_setf(props, PW_KEY_MEDIA_NAME, "VBAN Session with %s",
sess_name);
if ((str = pw_properties_get(props, "stream.props")) != NULL)
pw_properties_update_string(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_NODE_CHANNELNAMES);
copy_props(impl, props, PW_KEY_MEDIA_NAME);
copy_props(impl, props, PW_KEY_MEDIA_CLASS);
copy_props(impl, props, "net.mtu");
copy_props(impl, props, "sess.media");
copy_props(impl, props, "sess.name");
copy_props(impl, props, "sess.min-ptime");
copy_props(impl, props, "sess.max-ptime");
copy_props(impl, props, "sess.latency.msec");
str = pw_properties_get(props, "local.ifname");
impl->ifname = str ? strdup(str) : NULL;
impl->src_port = pw_properties_get_uint32(props, "source.port", DEFAULT_SOURCE_PORT);
if (impl->src_port == 0) {
pw_log_error("invalid source.port");
goto out;
}
if ((str = pw_properties_get(props, "source.ip")) == NULL)
str = DEFAULT_SOURCE_IP;
if ((res = parse_address(str, impl->src_port, &impl->src_addr, &impl->src_len)) < 0) {
pw_log_error("invalid source.ip %s: %s", str, spa_strerror(res));
goto out;
}
impl->always_process = pw_properties_get_bool(stream_props,
PW_KEY_NODE_ALWAYS_PROCESS, true);
impl->cleanup_interval = pw_properties_get_uint32(props,
"cleanup.sec", DEFAULT_CLEANUP_SEC);
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 out;
}
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);
impl->timer = pw_loop_add_timer(impl->loop, on_timer_event, impl);
if (impl->timer == NULL) {
res = -errno;
pw_log_error("can't create timer source: %m");
goto out;
}
value.tv_sec = impl->cleanup_interval;
value.tv_nsec = 0;
interval.tv_sec = impl->cleanup_interval;
interval.tv_nsec = 0;
pw_loop_update_timer(impl->loop, impl->timer, &value, &interval, false);
impl->stream = vban_stream_new(impl->core,
PW_DIRECTION_OUTPUT, pw_properties_copy(stream_props),
&stream_events, impl);
if (impl->stream == NULL) {
res = -errno;
pw_log_error("can't create stream: %m");
goto out;
}
pw_impl_module_add_listener(module, &impl->module_listener, &module_events, impl);
pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_info));
pw_log_info("Successfully loaded module-vban-recv");
return 0;
out:
impl_destroy(impl);
return res;
}

534
src/modules/module-vban-send.c Normal file
View file

@ -0,0 +1,534 @@
/* PipeWire */
/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
/* SPDX-License-Identifier: MIT */
#include "config.h"
#include <limits.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <arpa/inet.h>
#include <netinet/ip.h>
#include <netinet/in.h>
#include <net/if.h>
#include <ctype.h>
#include <spa/utils/hook.h>
#include <spa/utils/result.h>
#include <spa/utils/ringbuffer.h>
#include <spa/utils/json.h>
#include <spa/param/audio/format-utils.h>
#include <spa/debug/types.h>
#include <pipewire/pipewire.h>
#include <pipewire/impl.h>
#include <module-vban/stream.h>
#ifndef IPTOS_DSCP
#define IPTOS_DSCP_MASK 0xfc
#define IPTOS_DSCP(x) ((x) & IPTOS_DSCP_MASK)
#endif
/** \page page_module_vban_send PipeWire Module: VBAN sender
*
* The `vban-send` module creates a PipeWire sink that sends
* audio and midi [VBAN](https://vb-audio.com) packets.
*
* ## Module Options
*
* Options specific to the behavior of this module
*
* - `source.ip =<str>`: source IP address, default "0.0.0.0"
* - `destination.ip =<str>`: destination IP address, default "127.0.0.1"
* - `destination.port =<int>`: destination port, default 6980
* - `local.ifname = <str>`: interface name to use
* - `net.mtu = <int>`: MTU to use, default 1500
* - `net.ttl = <int>`: TTL to use, default 1
* - `net.loop = <bool>`: loopback multicast, default false
* - `sess.min-ptime = <int>`: minimum packet time in milliseconds, default 2
* - `sess.max-ptime = <int>`: maximum packet time in milliseconds, default 20
* - `sess.name = <str>`: a session name
* - `sess.media = <string>`: the media type audio|midi, default audio
* - `stream.props = {}`: properties to be passed to the stream
*
* ## General options
*
* Options with well-known behavior:
*
* - \ref PW_KEY_REMOTE_NAME
* - \ref PW_KEY_AUDIO_FORMAT
* - \ref PW_KEY_AUDIO_RATE
* - \ref PW_KEY_AUDIO_CHANNELS
* - \ref SPA_KEY_AUDIO_POSITION
* - \ref PW_KEY_NODE_NAME
* - \ref PW_KEY_NODE_DESCRIPTION
* - \ref PW_KEY_MEDIA_NAME
* - \ref PW_KEY_NODE_GROUP
* - \ref PW_KEY_NODE_LATENCY
* - \ref PW_KEY_NODE_VIRTUAL
* - \ref PW_KEY_MEDIA_CLASS
*
* ## Example configuration
*\code{.unparsed}
* context.modules = [
* { name = libpipewire-module-vban-send
* args = {
* #local.ifname = "eth0"
* #source.ip = "0.0.0.0"
* #destination.ip = "127.0.0.1"
* #destination.port = 6980
* #net.mtu = 1500
* #net.ttl = 1
* #net.loop = false
* #sess.min-ptime = 2
* #sess.max-ptime = 20
* #sess.name = "PipeWire VBAN stream"
* #sess.media = "audio"
* #audio.format = "S16LE"
* #audio.rate = 44100
* #audio.channels = 2
* #audio.position = [ FL FR ]
* stream.props = {
* node.name = "vban-sender"
* }
* }
*}
*]
*\endcode
*
* \since 0.3.76
*/
#define NAME "vban-send"
PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
#define PW_LOG_TOPIC_DEFAULT mod_topic
#define DEFAULT_PORT 6980
#define DEFAULT_SOURCE_IP "0.0.0.0"
#define DEFAULT_DESTINATION_IP "127.0.0.1"
#define DEFAULT_TTL 1
#define DEFAULT_LOOP false
#define DEFAULT_DSCP 34 /* Default to AES-67 AF41 (34) */
#define USAGE "( source.ip=<source IP address, default:"DEFAULT_SOURCE_IP"> ) " \
"( destination.ip=<destination IP address, default:"DEFAULT_DESTINATION_IP"> ) " \
"( destination.port=<int, default:"SPA_STRINGIFY(DEFAULT_PORT)"> ) " \
"( local.ifname=<local interface name to use> ) " \
"( net.mtu=<desired MTU, default:"SPA_STRINGIFY(DEFAULT_MTU)"> ) " \
"( net.ttl=<desired TTL, default:"SPA_STRINGIFY(DEFAULT_TTL)"> ) " \
"( net.loop=<desired loopback, default:"SPA_STRINGIFY(DEFAULT_LOOP)"> ) " \
"( net.dscp=<desired DSCP, default:"SPA_STRINGIFY(DEFAULT_DSCP)"> ) " \
"( sess.name=<a name for the session> ) " \
"( sess.min-ptime=<minimum packet time in milliseconds, default:2> ) " \
"( sess.max-ptime=<maximum packet time in milliseconds, default:20> ) " \
"( sess.media=<string, the media type audio|midi, default audio> ) " \
"( audio.format=<format, default:"DEFAULT_FORMAT"> ) " \
"( audio.rate=<sample rate, default:"SPA_STRINGIFY(DEFAULT_RATE)"> ) " \
"( audio.channels=<number of channels, default:"SPA_STRINGIFY(DEFAULT_CHANNELS)"> ) " \
"( audio.position=<channel map, default:"DEFAULT_POSITION"> ) " \
"( stream.props= { key=value ... } ) "
static const struct spa_dict_item module_info[] = {
{ PW_KEY_MODULE_AUTHOR, "Wim Taymans <wim.taymans@gmail.com>" },
{ PW_KEY_MODULE_DESCRIPTION, "VBAN Sender" },
{ PW_KEY_MODULE_USAGE, USAGE },
{ PW_KEY_MODULE_VERSION, PACKAGE_VERSION },
};
struct impl {
struct pw_context *context;
struct pw_impl_module *module;
struct spa_hook module_listener;
struct pw_properties *props;
struct pw_loop *loop;
struct pw_core *core;
struct spa_hook core_listener;
struct spa_hook core_proxy_listener;
struct pw_properties *stream_props;
struct vban_stream *stream;
unsigned int do_disconnect:1;
char *ifname;
char *session_name;
uint32_t ttl;
bool mcast_loop;
uint32_t dscp;
struct sockaddr_storage src_addr;
socklen_t src_len;
uint16_t dst_port;
struct sockaddr_storage dst_addr;
socklen_t dst_len;
int vban_fd;
};
static void stream_destroy(void *d)
{
struct impl *impl = d;
impl->stream = NULL;
}
static void stream_send_packet(void *data, struct iovec *iov, size_t iovlen)
{
struct impl *impl = data;
struct msghdr msg;
ssize_t n;
spa_zero(msg);
msg.msg_iov = iov;
msg.msg_iovlen = iovlen;
msg.msg_control = NULL;
msg.msg_controllen = 0;
msg.msg_flags = 0;
n = sendmsg(impl->vban_fd, &msg, MSG_NOSIGNAL);
if (n < 0)
pw_log_debug("sendmsg() failed: %m");
}
static void stream_state_changed(void *data, bool started, const char *error)
{
struct impl *impl = data;
if (error) {
pw_log_error("stream error: %s", error);
pw_impl_module_schedule_destroy(impl->module);
}
}
static const struct vban_stream_events stream_events = {
VBAN_VERSION_STREAM_EVENTS,
.destroy = stream_destroy,
.state_changed = stream_state_changed,
.send_packet = stream_send_packet,
};
static int parse_address(const char *address, uint16_t port,
struct sockaddr_storage *addr, socklen_t *len)
{
struct sockaddr_in *sa4 = (struct sockaddr_in*)addr;
struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)addr;
if (inet_pton(AF_INET, address, &sa4->sin_addr) > 0) {
sa4->sin_family = AF_INET;
sa4->sin_port = htons(port);
*len = sizeof(*sa4);
} else if (inet_pton(AF_INET6, address, &sa6->sin6_addr) > 0) {
sa6->sin6_family = AF_INET6;
sa6->sin6_port = htons(port);
*len = sizeof(*sa6);
} else
return -EINVAL;
return 0;
}
static bool is_multicast(struct sockaddr *sa, socklen_t salen)
{
if (sa->sa_family == AF_INET) {
static const uint32_t ipv4_mcast_mask = 0xe0000000;
struct sockaddr_in *sa4 = (struct sockaddr_in*)sa;
return (ntohl(sa4->sin_addr.s_addr) & ipv4_mcast_mask) == ipv4_mcast_mask;
} else if (sa->sa_family == AF_INET6) {
struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)sa;
return sa6->sin6_addr.s6_addr[0] == 0xff;
}
return false;
}
static int make_socket(struct sockaddr_storage *src, socklen_t src_len,
struct sockaddr_storage *dst, socklen_t dst_len,
bool loop, int ttl, int dscp)
{
int af, fd, val, res;
af = src->ss_family;
if ((fd = socket(af, SOCK_DGRAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0) {
pw_log_error("socket failed: %m");
return -errno;
}
if (bind(fd, (struct sockaddr*)src, src_len) < 0) {
res = -errno;
pw_log_error("bind() failed: %m");
goto error;
}
if (connect(fd, (struct sockaddr*)dst, dst_len) < 0) {
res = -errno;
pw_log_error("connect() failed: %m");
goto error;
}
if (is_multicast((struct sockaddr*)dst, dst_len)) {
val = loop;
if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_LOOP, &val, sizeof(val)) < 0)
pw_log_warn("setsockopt(IP_MULTICAST_LOOP) failed: %m");
val = ttl;
if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_TTL, &val, sizeof(val)) < 0)
pw_log_warn("setsockopt(IP_MULTICAST_TTL) failed: %m");
}
#ifdef SO_PRIORITY
val = 6;
if (setsockopt(fd, SOL_SOCKET, SO_PRIORITY, &val, sizeof(val)) < 0)
pw_log_warn("setsockopt(SO_PRIORITY) failed: %m");
#endif
if (dscp > 0) {
val = IPTOS_DSCP(dscp << 2);
if (setsockopt(fd, IPPROTO_IP, IP_TOS, &val, sizeof(val)) < 0)
pw_log_warn("setsockopt(IP_TOS) failed: %m");
}
return fd;
error:
close(fd);
return res;
}
static int get_ip(const struct sockaddr_storage *sa, char *ip, size_t len)
{
if (sa->ss_family == AF_INET) {
struct sockaddr_in *in = (struct sockaddr_in*)sa;
inet_ntop(sa->ss_family, &in->sin_addr, ip, len);
} else if (sa->ss_family == AF_INET6) {
struct sockaddr_in6 *in = (struct sockaddr_in6*)sa;
inet_ntop(sa->ss_family, &in->sin6_addr, ip, len);
} else
return -EIO;
return 0;
}
static void core_destroy(void *d)
{
struct impl *impl = d;
spa_hook_remove(&impl->core_listener);
impl->core = NULL;
pw_impl_module_schedule_destroy(impl->module);
}
static const struct pw_proxy_events core_proxy_events = {
.destroy = core_destroy,
};
static void impl_destroy(struct impl *impl)
{
if (impl->stream)
vban_stream_destroy(impl->stream);
if (impl->core && impl->do_disconnect)
pw_core_disconnect(impl->core);
if (impl->vban_fd != -1)
close(impl->vban_fd);
pw_properties_free(impl->stream_props);
pw_properties_free(impl->props);
free(impl->ifname);
free(impl->session_name);
free(impl);
}
static void module_destroy(void *d)
{
struct impl *impl = d;
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 void on_core_error(void *d, uint32_t id, int seq, int res, const char *message)
{
struct impl *impl = d;
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)
pw_impl_module_schedule_destroy(impl->module);
}
static const struct pw_core_events core_events = {
PW_VERSION_CORE_EVENTS,
.error = on_core_error,
};
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 impl *impl;
struct pw_properties *props = NULL, *stream_props = NULL;
char addr[64];
const char *str, *sess_name;
int res = 0;
PW_LOG_TOPIC_INIT(mod_topic);
impl = calloc(1, sizeof(struct impl));
if (impl == NULL)
return -errno;
impl->vban_fd = -1;
if (args == NULL)
args = "";
props = pw_properties_new_string(args);
if (props == NULL) {
res = -errno;
pw_log_error( "can't create properties: %m");
goto out;
}
impl->props = props;
stream_props = pw_properties_new(NULL, NULL);
if (stream_props == NULL) {
res = -errno;
pw_log_error( "can't create properties: %m");
goto out;
}
impl->stream_props = stream_props;
impl->module = module;
impl->context = context;
impl->loop = pw_context_get_main_loop(context);
if ((sess_name = pw_properties_get(props, "sess.name")) == NULL)
sess_name = pw_get_host_name();
if (pw_properties_get(props, PW_KEY_NODE_NAME) == NULL)
pw_properties_setf(props, PW_KEY_NODE_NAME, "vban_session.%s", sess_name);
if (pw_properties_get(props, PW_KEY_NODE_DESCRIPTION) == NULL)
pw_properties_setf(props, PW_KEY_NODE_DESCRIPTION, "%s", sess_name);
if (pw_properties_get(props, PW_KEY_MEDIA_NAME) == NULL)
pw_properties_setf(props, PW_KEY_MEDIA_NAME, "VBAN Session with %s",
sess_name);
if ((str = pw_properties_get(props, "stream.props")) != NULL)
pw_properties_update_string(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_NODE_CHANNELNAMES);
copy_props(impl, props, PW_KEY_MEDIA_NAME);
copy_props(impl, props, PW_KEY_MEDIA_CLASS);
copy_props(impl, props, "net.mtu");
copy_props(impl, props, "sess.media");
copy_props(impl, props, "sess.name");
copy_props(impl, props, "sess.min-ptime");
copy_props(impl, props, "sess.max-ptime");
copy_props(impl, props, "sess.latency.msec");
copy_props(impl, props, "sess.ts-refclk");
str = pw_properties_get(props, "local.ifname");
impl->ifname = str ? strdup(str) : NULL;
if ((str = pw_properties_get(props, "source.ip")) == NULL)
str = DEFAULT_SOURCE_IP;
if ((res = parse_address(str, 0, &impl->src_addr, &impl->src_len)) < 0) {
pw_log_error("invalid source.ip %s: %s", str, spa_strerror(res));
goto out;
}
impl->dst_port = pw_properties_get_uint32(props, "destination.port", DEFAULT_PORT);
if ((str = pw_properties_get(props, "destination.ip")) == NULL)
str = DEFAULT_DESTINATION_IP;
if ((res = parse_address(str, impl->dst_port, &impl->dst_addr, &impl->dst_len)) < 0) {
pw_log_error("invalid destination.ip %s: %s", str, spa_strerror(res));
goto out;
}
impl->ttl = pw_properties_get_uint32(props, "net.ttl", DEFAULT_TTL);
impl->mcast_loop = pw_properties_get_bool(props, "net.loop", DEFAULT_LOOP);
impl->dscp = pw_properties_get_uint32(props, "net.dscp", DEFAULT_DSCP);
get_ip(&impl->src_addr, addr, sizeof(addr));
pw_properties_set(stream_props, "vban.source.ip", addr);
get_ip(&impl->dst_addr, addr, sizeof(addr));
pw_properties_set(stream_props, "vban.destination.ip", addr);
pw_properties_setf(stream_props, "vban.destination.port", "%u", impl->dst_port);
pw_properties_setf(stream_props, "vban.ttl", "%u", impl->ttl);
pw_properties_setf(stream_props, "vban.dscp", "%u", impl->dscp);
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 out;
}
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 = make_socket(&impl->src_addr, impl->src_len,
&impl->dst_addr, impl->dst_len,
impl->mcast_loop, impl->ttl, impl->dscp)) < 0) {
pw_log_error("can't make socket: %s", spa_strerror(res));
goto out;
}
impl->vban_fd = res;
impl->stream = vban_stream_new(impl->core,
PW_DIRECTION_INPUT, pw_properties_copy(stream_props),
&stream_events, impl);
if (impl->stream == NULL) {
res = -errno;
pw_log_error("can't create stream: %m");
goto out;
}
pw_impl_module_add_listener(module, &impl->module_listener, &module_events, impl);
pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_info));
pw_log_info("Successfully loaded module-vban-send");
return 0;
out:
impl_destroy(impl);
return res;
}

281
src/modules/module-vban/audio.c Normal file
View file

@ -0,0 +1,281 @@
/* PipeWire */
/* SPDX-FileCopyrightText: Copyright © 2022 Wim Taymans <wim.taymans@gmail.com> */
/* SPDX-License-Identifier: MIT */
static void vban_audio_process_playback(void *data)
{
struct impl *impl = data;
struct pw_buffer *buf;
struct spa_data *d;
uint32_t wanted, timestamp, target_buffer, stride, maxsize;
int32_t avail;
if ((buf = pw_stream_dequeue_buffer(impl->stream)) == NULL) {
pw_log_debug("Out of stream buffers: %m");
return;
}
d = buf->buffer->datas;
stride = impl->stride;
maxsize = d[0].maxsize / stride;
wanted = buf->requested ? SPA_MIN(buf->requested, maxsize) : maxsize;
avail = spa_ringbuffer_get_read_index(&impl->ring, &timestamp);
target_buffer = impl->target_buffer;
if (avail < (int32_t)wanted) {
enum spa_log_level level;
memset(d[0].data, 0, wanted * stride);
if (impl->have_sync) {
impl->have_sync = false;
level = SPA_LOG_LEVEL_WARN;
} else {
level = SPA_LOG_LEVEL_DEBUG;
}
pw_log(level, "underrun %d/%u < %u",
avail, target_buffer, wanted);
} else {
float error, corr;
if (impl->first) {
if ((uint32_t)avail > target_buffer) {
uint32_t skip = avail - target_buffer;
pw_log_debug("first: avail:%d skip:%u target:%u",
avail, skip, target_buffer);
timestamp += skip;
avail = target_buffer;
}
impl->first = false;
} else if (avail > (int32_t)SPA_MIN(target_buffer * 8, BUFFER_SIZE / stride)) {
pw_log_warn("overrun %u > %u", avail, target_buffer * 8);
timestamp += avail - target_buffer;
avail = target_buffer;
}
/* try to adjust our playback rate to keep the
* requested target_buffer bytes in the ringbuffer */
error = (float)target_buffer - (float)avail;
error = SPA_CLAMP(error, -impl->max_error, impl->max_error);
corr = spa_dll_update(&impl->dll, error);
pw_log_debug("avail:%u target:%u error:%f corr:%f", avail,
target_buffer, error, corr);
if (impl->io_rate_match) {
SPA_FLAG_SET(impl->io_rate_match->flags,
SPA_IO_RATE_MATCH_FLAG_ACTIVE);
impl->io_rate_match->rate = 1.0f / corr;
}
spa_ringbuffer_read_data(&impl->ring,
impl->buffer,
BUFFER_SIZE,
(timestamp * stride) & BUFFER_MASK,
d[0].data, wanted * stride);
timestamp += wanted;
spa_ringbuffer_read_update(&impl->ring, timestamp);
}
d[0].chunk->size = wanted * stride;
d[0].chunk->stride = stride;
d[0].chunk->offset = 0;
buf->size = wanted;
pw_stream_queue_buffer(impl->stream, buf);
}
static int vban_audio_receive(struct impl *impl, uint8_t *buffer, ssize_t len)
{
struct vban_header *hdr;
ssize_t hlen, plen;
uint32_t n_frames, timestamp, samples, write, expected_write;
uint32_t stride = impl->stride;
int32_t filled;
if (len < VBAN_HEADER_SIZE)
goto short_packet;
hdr = (struct vban_header*)buffer;
if (strncmp(hdr->vban, "VBAN", 3))
goto invalid_version;
impl->receiving = true;
hlen = VBAN_HEADER_SIZE;
plen = len - hlen;
samples = SPA_MIN(hdr->format_nbs, plen / stride);
n_frames = hdr->n_frames;
if (impl->have_sync && impl->n_frames != n_frames) {
pw_log_info("unexpected frame (%d != %d)",
n_frames, impl->n_frames);
impl->have_sync = false;
}
impl->n_frames = n_frames + 1;
timestamp = impl->timestamp;
impl->timestamp += samples;
filled = spa_ringbuffer_get_write_index(&impl->ring, &expected_write);
/* we always write to timestamp + delay */
write = timestamp + impl->target_buffer;
if (!impl->have_sync) {
pw_log_info("sync to timestamp:%u target:%u",
timestamp, impl->target_buffer);
/* we read from timestamp, keeping target_buffer of data
* in the ringbuffer. */
impl->ring.readindex = timestamp;
impl->ring.writeindex = write;
filled = impl->target_buffer;
spa_dll_init(&impl->dll);
spa_dll_set_bw(&impl->dll, SPA_DLL_BW_MIN, 128, impl->rate);
memset(impl->buffer, 0, BUFFER_SIZE);
impl->have_sync = true;
} else if (expected_write != write) {
pw_log_debug("unexpected write (%u != %u)",
write, expected_write);
}
if (filled + samples > BUFFER_SIZE / stride) {
pw_log_debug("capture overrun %u + %u > %u", filled, samples,
BUFFER_SIZE / stride);
impl->have_sync = false;
} else {
pw_log_debug("got samples:%u", samples);
spa_ringbuffer_write_data(&impl->ring,
impl->buffer,
BUFFER_SIZE,
(write * stride) & BUFFER_MASK,
&buffer[hlen], (samples * stride));
write += samples;
spa_ringbuffer_write_update(&impl->ring, write);
}
return 0;
short_packet:
pw_log_warn("short packet received");
return -EINVAL;
invalid_version:
pw_log_warn("invalid VBAN version");
spa_debug_mem(0, buffer, len);
return -EPROTO;
}
static inline void
set_iovec(struct spa_ringbuffer *rbuf, void *buffer, uint32_t size,
uint32_t offset, struct iovec *iov, uint32_t len)
{
iov[0].iov_len = SPA_MIN(len, size - offset);
iov[0].iov_base = SPA_PTROFF(buffer, offset, void);
iov[1].iov_len = len - iov[0].iov_len;
iov[1].iov_base = buffer;
}
static void vban_audio_flush_packets(struct impl *impl)
{
int32_t avail, tosend;
uint32_t stride, timestamp;
struct iovec iov[3];
struct vban_header header;
avail = spa_ringbuffer_get_read_index(&impl->ring, &timestamp);
tosend = impl->psamples;
if (avail < tosend)
return;
stride = impl->stride;
header = impl->header;
header.format_nbs = tosend - 1;
header.format_nbc = impl->stream_info.info.raw.channels - 1;
iov[0].iov_base = &header;
iov[0].iov_len = sizeof(header);
while (avail >= tosend) {
set_iovec(&impl->ring,
impl->buffer, BUFFER_SIZE,
(timestamp * stride) & BUFFER_MASK,
&iov[1], tosend * stride);
pw_log_trace("sending %d timestamp:%08x", tosend, timestamp);
vban_stream_emit_send_packet(impl, iov, 3);
timestamp += tosend;
avail -= tosend;
impl->header.n_frames++;
}
spa_ringbuffer_read_update(&impl->ring, timestamp);
}
static void vban_audio_process_capture(void *data)
{
struct impl *impl = data;
struct pw_buffer *buf;
struct spa_data *d;
uint32_t offs, size, timestamp, expected_timestamp, stride;
int32_t filled, wanted;
if ((buf = pw_stream_dequeue_buffer(impl->stream)) == NULL) {
pw_log_debug("Out of stream buffers: %m");
return;
}
d = buf->buffer->datas;
offs = SPA_MIN(d[0].chunk->offset, d[0].maxsize);
size = SPA_MIN(d[0].chunk->size, d[0].maxsize - offs);
stride = impl->stride;
wanted = size / stride;
filled = spa_ringbuffer_get_write_index(&impl->ring, &expected_timestamp);
if (SPA_LIKELY(impl->io_position)) {
uint32_t rate = impl->io_position->clock.rate.denom;
timestamp = impl->io_position->clock.position * impl->rate / rate;
} else
timestamp = expected_timestamp;
if (!impl->have_sync) {
pw_log_info("sync to timestamp:%u", timestamp);
impl->ring.readindex = impl->ring.writeindex = timestamp;
memset(impl->buffer, 0, BUFFER_SIZE);
impl->have_sync = true;
expected_timestamp = timestamp;
} else {
if (SPA_ABS((int32_t)expected_timestamp - (int32_t)timestamp) > 32) {
pw_log_warn("expected %u != timestamp %u", expected_timestamp, timestamp);
impl->have_sync = false;
} else if (filled + wanted > (int32_t)(BUFFER_SIZE / stride)) {
pw_log_warn("overrun %u + %u > %u", filled, wanted, BUFFER_SIZE / stride);
impl->have_sync = false;
}
}
spa_ringbuffer_write_data(&impl->ring,
impl->buffer,
BUFFER_SIZE,
(expected_timestamp * stride) & BUFFER_MASK,
SPA_PTROFF(d[0].data, offs, void), wanted * stride);
expected_timestamp += wanted;
spa_ringbuffer_write_update(&impl->ring, expected_timestamp);
pw_stream_queue_buffer(impl->stream, buf);
vban_audio_flush_packets(impl);
}
static int vban_audio_init(struct impl *impl, enum spa_direction direction)
{
if (direction == SPA_DIRECTION_INPUT)
impl->stream_events.process = vban_audio_process_capture;
else
impl->stream_events.process = vban_audio_process_playback;
impl->receive_vban = vban_audio_receive;
return 0;
}

479
src/modules/module-vban/stream.c Normal file
View file

@ -0,0 +1,479 @@
/* PipeWire */
/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
/* SPDX-License-Identifier: MIT */
#include <sys/socket.h>
#include <arpa/inet.h>
#include <spa/utils/result.h>
#include <spa/utils/json.h>
#include <spa/utils/ringbuffer.h>
#include <spa/utils/dll.h>
#include <spa/param/audio/format-utils.h>
#include <spa/control/control.h>
#include <spa/debug/types.h>
#include <spa/debug/mem.h>
#include "config.h"
#include <pipewire/pipewire.h>
#include <pipewire/impl.h>
#include <module-vban/vban.h>
#include <module-vban/stream.h>
#define BUFFER_SIZE (1u<<22)
#define BUFFER_MASK (BUFFER_SIZE-1)
#define BUFFER_SIZE2 (BUFFER_SIZE>>1)
#define BUFFER_MASK2 (BUFFER_SIZE2-1)
#define vban_stream_emit(s,m,v,...) spa_hook_list_call(&s->listener_list, \
struct vban_stream_events, m, v, ##__VA_ARGS__)
#define vban_stream_emit_destroy(s) vban_stream_emit(s, destroy, 0)
#define vban_stream_emit_state_changed(s,n,e) vban_stream_emit(s, state_changed,0,n,e)
#define vban_stream_emit_send_packet(s,i,l) vban_stream_emit(s, send_packet,0,i,l)
#define vban_stream_emit_send_feedback(s,seq) vban_stream_emit(s, send_feedback,0,seq)
struct impl {
struct spa_audio_info info;
struct spa_audio_info stream_info;
struct pw_stream *stream;
struct spa_hook stream_listener;
struct pw_stream_events stream_events;
struct spa_hook_list listener_list;
struct spa_hook listener;
const struct format_info *format_info;
void *stream_data;
uint32_t rate;
uint32_t stride;
uint32_t psamples;
uint32_t mtu;
struct vban_header header;
uint32_t timestamp;
uint32_t n_frames;
struct spa_ringbuffer ring;
uint8_t buffer[BUFFER_SIZE];
struct spa_io_rate_match *io_rate_match;
struct spa_io_position *io_position;
struct spa_dll dll;
double corr;
uint32_t target_buffer;
float max_error;
float last_timestamp;
float last_time;
unsigned always_process:1;
unsigned started:1;
unsigned have_sync:1;
unsigned receiving:1;
unsigned first:1;
int (*receive_vban)(struct impl *impl, uint8_t *buffer, ssize_t len);
};
#include "module-vban/audio.c"
//#include "module-vban/midi.c"
struct format_info {
uint32_t media_subtype;
uint32_t format;
uint32_t size;
uint8_t format_bit;
};
static const struct format_info audio_format_info[] = {
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_U8, 1, VBAN_DATATYPE_U8, },
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S16_LE, 2, VBAN_DATATYPE_INT16, },
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S24_LE, 3, VBAN_DATATYPE_INT24, },
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S32_LE, 4, VBAN_DATATYPE_INT32, },
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_F32_LE, 4, VBAN_DATATYPE_FLOAT32, },
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_F64_LE, 8, VBAN_DATATYPE_FLOAT64, },
{ SPA_MEDIA_SUBTYPE_control, 0, 1, },
};
static void stream_io_changed(void *data, uint32_t id, void *area, uint32_t size)
{
struct impl *impl = data;
switch (id) {
case SPA_IO_RateMatch:
impl->io_rate_match = area;
break;
case SPA_IO_Position:
impl->io_position = area;
break;
}
}
static void stream_destroy(void *d)
{
struct impl *impl = d;
spa_hook_remove(&impl->stream_listener);
impl->stream = NULL;
}
static int stream_start(struct impl *impl)
{
if (impl->started)
return 0;
vban_stream_emit_state_changed(impl, true, NULL);
impl->started = true;
return 0;
}
static int stream_stop(struct impl *impl)
{
if (!impl->started)
return 0;
vban_stream_emit_state_changed(impl, false, NULL);
impl->started = false;
return 0;
}
static void on_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_UNCONNECTED:
pw_log_info("stream disconnected");
break;
case PW_STREAM_STATE_ERROR:
pw_log_error("stream error: %s", error);
vban_stream_emit_state_changed(impl, false, error);
break;
case PW_STREAM_STATE_STREAMING:
if ((errno = -stream_start(impl)) < 0)
pw_log_error("failed to start RTP stream: %m");
break;
case PW_STREAM_STATE_PAUSED:
if (!impl->always_process)
stream_stop(impl);
impl->have_sync = false;
break;
default:
break;
}
}
static const struct pw_stream_events stream_events = {
PW_VERSION_STREAM_EVENTS,
.destroy = stream_destroy,
.state_changed = on_stream_state_changed,
.io_changed = stream_io_changed,
};
static const struct format_info *find_audio_format_info(const struct spa_audio_info *info)
{
SPA_FOR_EACH_ELEMENT_VAR(audio_format_info, f)
if (f->media_subtype == info->media_subtype &&
(f->format == 0 || f->format == info->info.raw.format))
return f;
return NULL;
}
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 void parse_audio_info(const struct pw_properties *props, struct spa_audio_info_raw *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));
info->rate = pw_properties_get_uint32(props, PW_KEY_AUDIO_RATE, info->rate);
if (info->rate == 0)
info->rate = DEFAULT_RATE;
info->channels = pw_properties_get_uint32(props, PW_KEY_AUDIO_CHANNELS, info->channels);
info->channels = SPA_MIN(info->channels, SPA_AUDIO_MAX_CHANNELS);
if ((str = pw_properties_get(props, SPA_KEY_AUDIO_POSITION)) != NULL)
parse_position(info, str, strlen(str));
if (info->channels == 0)
parse_position(info, DEFAULT_POSITION, strlen(DEFAULT_POSITION));
}
static uint32_t msec_to_samples(struct impl *impl, uint32_t msec)
{
return msec * impl->rate / 1000;
}
struct vban_stream *vban_stream_new(struct pw_core *core,
enum pw_direction direction, struct pw_properties *props,
const struct vban_stream_events *events, void *data)
{
struct impl *impl;
const char *str;
uint8_t buffer[1024];
struct spa_pod_builder b;
uint32_t n_params, min_samples, max_samples;
float min_ptime, max_ptime;
const struct spa_pod *params[1];
enum pw_stream_flags flags;
int latency_msec;
int res;
impl = calloc(1, sizeof(*impl));
if (impl == NULL) {
res = -errno;
goto out;
return NULL;
}
impl->first = true;
spa_hook_list_init(&impl->listener_list);
impl->stream_events = stream_events;
if ((str = pw_properties_get(props, "sess.media")) == NULL)
str = "audio";
if (spa_streq(str, "audio")) {
impl->info.media_type = SPA_MEDIA_TYPE_audio;
impl->info.media_subtype = SPA_MEDIA_SUBTYPE_raw;
}
else if (spa_streq(str, "midi")) {
impl->info.media_type = SPA_MEDIA_TYPE_application;
impl->info.media_subtype = SPA_MEDIA_SUBTYPE_control;
}
else {
pw_log_error("unsupported media type:%s", str);
res = -EINVAL;
goto out;
}
memcpy(impl->header.vban, "VBAN", 4);
if ((str = pw_properties_get(props, "sess.name")) == NULL)
str = "Stream1";
strcpy(impl->header.stream_name, str);
switch (impl->info.media_subtype) {
case SPA_MEDIA_SUBTYPE_raw:
parse_audio_info(props, &impl->info.info.raw);
impl->stream_info = impl->info;
impl->format_info = find_audio_format_info(&impl->info);
if (impl->format_info == NULL) {
pw_log_error("unsupported audio format:%d channels:%d",
impl->stream_info.info.raw.format,
impl->stream_info.info.raw.channels);
res = -EINVAL;
goto out;
}
impl->stride = impl->format_info->size * impl->stream_info.info.raw.channels;
impl->rate = impl->stream_info.info.raw.rate;
impl->header.format_SR = vban_sr_index(impl->rate);
if (impl->header.format_SR == VBAN_SR_MAXNUMBER) {
pw_log_error("unsupported audio rate:%u", impl->rate);
res = -EINVAL;
goto out;
}
impl->header.format_bit = impl->format_info->format_bit;
break;
case SPA_MEDIA_SUBTYPE_control:
impl->stream_info = impl->info;
impl->format_info = find_audio_format_info(&impl->info);
if (impl->format_info == NULL) {
res = -EINVAL;
goto out;
}
pw_properties_set(props, PW_KEY_FORMAT_DSP, "8 bit raw midi");
impl->stride = impl->format_info->size;
impl->rate = pw_properties_get_uint32(props, "midi.rate", 10000);
if (impl->rate == 0)
impl->rate = 10000;
break;
default:
spa_assert_not_reached();
break;
}
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_NODE_NETWORK) == NULL)
pw_properties_set(props, PW_KEY_NODE_NETWORK, "true");
impl->mtu = pw_properties_get_uint32(props, "net.mtu", DEFAULT_MTU);
str = pw_properties_get(props, "sess.min-ptime");
if (!spa_atof(str, &min_ptime))
min_ptime = DEFAULT_MIN_PTIME;
str = pw_properties_get(props, "sess.max-ptime");
if (!spa_atof(str, &max_ptime))
max_ptime = DEFAULT_MAX_PTIME;
min_samples = min_ptime * impl->rate / 1000;
max_samples = SPA_MIN(256, max_ptime * impl->rate / 1000);
float ptime = 0;
if ((str = pw_properties_get(props, "vban.ptime")) != NULL)
if (!spa_atof(str, &ptime))
ptime = 0.0;
if (ptime) {
impl->psamples = ptime * impl->rate / 1000;
} else {
impl->psamples = impl->mtu / impl->stride;
impl->psamples = SPA_CLAMP(impl->psamples, min_samples, max_samples);
if (direction == PW_DIRECTION_OUTPUT)
pw_properties_setf(props, "vban.ptime", "%f",
impl->psamples * 1000.0 / impl->rate);
}
latency_msec = pw_properties_get_uint32(props,
"sess.latency.msec", DEFAULT_SESS_LATENCY);
impl->target_buffer = msec_to_samples(impl, latency_msec);
impl->max_error = msec_to_samples(impl, ERROR_MSEC);
pw_properties_setf(props, PW_KEY_NODE_RATE, "1/%d", impl->rate);
if (direction == PW_DIRECTION_INPUT) {
pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%d/%d",
impl->psamples, impl->rate);
} else {
pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%d/%d",
impl->target_buffer / 2, impl->rate);
}
pw_properties_setf(props, "net.mtu", "%u", impl->mtu);
pw_properties_setf(props, "vban.rate", "%u", impl->rate);
if (impl->info.info.raw.channels > 0)
pw_properties_setf(props, "vban.channels", "%u", impl->info.info.raw.channels);
spa_dll_init(&impl->dll);
spa_dll_set_bw(&impl->dll, SPA_DLL_BW_MIN, 128, impl->rate);
impl->corr = 1.0;
impl->stream = pw_stream_new(core, "vban-session", props);
props = NULL;
if (impl->stream == NULL) {
res = -errno;
pw_log_error("can't create stream: %m");
goto out;
}
n_params = 0;
spa_pod_builder_init(&b, buffer, sizeof(buffer));
flags = PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS;
switch (impl->info.media_subtype) {
case SPA_MEDIA_SUBTYPE_raw:
params[n_params++] = spa_format_audio_build(&b,
SPA_PARAM_EnumFormat, &impl->stream_info);
flags |= PW_STREAM_FLAG_AUTOCONNECT;
vban_audio_init(impl, direction);
break;
case SPA_MEDIA_SUBTYPE_control:
params[n_params++] = spa_pod_builder_add_object(&b,
SPA_TYPE_OBJECT_Format, SPA_PARAM_EnumFormat,
SPA_FORMAT_mediaType, SPA_POD_Id(SPA_MEDIA_TYPE_application),
SPA_FORMAT_mediaSubtype, SPA_POD_Id(SPA_MEDIA_SUBTYPE_control));
// vban_midi_init(impl, direction);
break;
default:
res = -EINVAL;
goto out;
}
pw_stream_add_listener(impl->stream,
&impl->stream_listener,
&impl->stream_events, impl);
if ((res = pw_stream_connect(impl->stream,
direction,
PW_ID_ANY,
flags,
params, n_params)) < 0) {
pw_log_error("can't connect stream: %s", spa_strerror(res));
goto out;
}
if (impl->always_process &&
(res = stream_start(impl)) < 0)
goto out;
spa_hook_list_append(&impl->listener_list, &impl->listener, events, data);
return (struct vban_stream*)impl;
out:
pw_properties_free(props);
errno = -res;
return NULL;
}
void vban_stream_destroy(struct vban_stream *s)
{
struct impl *impl = (struct impl*)s;
vban_stream_emit_destroy(impl);
if (impl->stream)
pw_stream_destroy(impl->stream);
spa_hook_list_clean(&impl->listener_list);
free(impl);
}
int vban_stream_receive_packet(struct vban_stream *s, uint8_t *buffer, size_t len)
{
struct impl *impl = (struct impl*)s;
return impl->receive_vban(impl, buffer, len);
}
uint64_t vban_stream_get_time(struct vban_stream *s, uint64_t *rate)
{
struct impl *impl = (struct impl*)s;
struct spa_io_position *pos = impl->io_position;
if (pos == NULL)
return -EIO;
*rate = impl->rate;
return pos->clock.position * impl->rate *
pos->clock.rate.num / pos->clock.rate.denom;
}

54
src/modules/module-vban/stream.h Normal file
View file

@ -0,0 +1,54 @@
/* PipeWire */
/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
/* SPDX-License-Identifier: MIT */
#ifndef PIPEWIRE_VBAN_STREAM_H
#define PIPEWIRE_VBAN_STREAM_H
#ifdef __cplusplus
extern "C" {
#endif
struct vban_stream;
#define DEFAULT_FORMAT "S16LE"
#define DEFAULT_RATE 44100
#define DEFAULT_CHANNELS 2
#define DEFAULT_POSITION "[ FL FR ]"
#define ERROR_MSEC 2
#define DEFAULT_SESS_LATENCY 100
#define DEFAULT_MTU VBAN_PROTOCOL_MAX_SIZE
#define DEFAULT_MIN_PTIME 2
#define DEFAULT_MAX_PTIME 20
struct vban_stream_events {
#define VBAN_VERSION_STREAM_EVENTS 0
uint32_t version;
void (*destroy) (void *data);
void (*state_changed) (void *data, bool started, const char *error);
void (*send_packet) (void *data, struct iovec *iov, size_t iovlen);
void (*send_feedback) (void *data, uint32_t senum);
};
struct vban_stream *vban_stream_new(struct pw_core *core,
enum pw_direction direction, struct pw_properties *props,
const struct vban_stream_events *events, void *data);
void vban_stream_destroy(struct vban_stream *s);
int vban_stream_receive_packet(struct vban_stream *s, uint8_t *buffer, size_t len);
uint64_t vban_stream_get_time(struct vban_stream *s, uint64_t *rate);
#ifdef __cplusplus
}
#endif
#endif /* PIPEWIRE_VBAN_STREAM_H */

60
src/modules/module-vban/vban.h Normal file
View file

@ -0,0 +1,60 @@
/* PipeWire */
/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
/* SPDX-License-Identifier: MIT */
#ifndef PIPEWIRE_VBAN_H
#define PIPEWIRE_VBAN_H
#ifdef __cplusplus
extern "C" {
#endif
#define VBAN_HEADER_SIZE (4 + 4 + 16 + 4)
#define VBAN_STREAM_NAME_SIZE 16
#define VBAN_PROTOCOL_MAX_SIZE 1464
#define VBAN_DATA_MAX_SIZE (VBAN_PROTOCOL_MAX_SIZE - VBAN_HEADER_SIZE)
#define VBAN_CHANNELS_MAX_NB 256
#define VBAN_SAMPLES_MAX_NB 256
struct vban_header {
char vban[4]; /* contains 'V' 'B', 'A', 'N' */
uint8_t format_SR; /* SR index */
uint8_t format_nbs; /* nb sample per frame (1 to 256) */
uint8_t format_nbc; /* nb channel (1 to 256) */
uint8_t format_bit; /* bit format */
char stream_name[VBAN_STREAM_NAME_SIZE]; /* stream name */
uint32_t n_frames; /* growing frame number. */
} __attribute__ ((packed));
#define VBAN_SR_MAXNUMBER 21
static uint32_t const vban_SR[VBAN_SR_MAXNUMBER] = {
6000, 12000, 24000, 48000, 96000, 192000, 384000,
8000, 16000, 32000, 64000, 128000, 256000, 512000,
11025, 22050, 44100, 88200, 176400, 352800, 705600
};
static inline uint8_t vban_sr_index(uint32_t rate)
{
uint8_t i;
for (i = 0; i < SPA_N_ELEMENTS(vban_SR); i++) {
if (vban_SR[i] == rate)
return i;
}
return VBAN_SR_MAXNUMBER;
}
#define VBAN_DATATYPE_U8 0x00
#define VBAN_DATATYPE_INT16 0x01
#define VBAN_DATATYPE_INT24 0x02
#define VBAN_DATATYPE_INT32 0x03
#define VBAN_DATATYPE_FLOAT32 0x04
#define VBAN_DATATYPE_FLOAT64 0x05
#define VBAN_DATATYPE_12BITS 0x06
#define VBAN_DATATYPE_10BITS 0x07
#ifdef __cplusplus
}
#endif
#endif /* PIPEWIRE_VBAN_H */