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

module-rtp: port source and sink to new stream

Browse source
This commit is contained in:
Wim Taymans 2023-03-02 19:55:44 +01:00
parent e8119cb087
commit be09198249
7 changed files with 127 additions and 1434 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/modules/meson.build
View file

@ -519,7 +519,8 @@ roc_dep = dependency('roc', required: get_option('roc'))
summary({'ROC': roc_dep.found()}, bool_yn: true, section: 'Streaming between daemons')
pipewire_module_rtp_source = shared_library('pipewire-module-rtp-source',
[ 'module-rtp-source.c' ],
[ 'module-rtp-source.c',
'module-rtp/stream.c' ],
include_directories : [configinc],
install : true,
install_dir : modules_install_dir,
@ -528,7 +529,8 @@ pipewire_module_rtp_source = shared_library('pipewire-module-rtp-source',
)
pipewire_module_rtp_sink = shared_library('pipewire-module-rtp-sink',
[ 'module-rtp-sink.c' ],
[ 'module-rtp-sink.c',
'module-rtp/stream.c' ],
include_directories : [configinc],
install : true,
install_dir : modules_install_dir,

2
src/modules/module-rtp-sap.c
View file

@ -648,7 +648,7 @@ static struct session *session_new_announce(struct impl *impl, struct node *node
sdp->ntp = (uint32_t) time(NULL) + 2208988800U;
sess->props = props;
if ((str = pw_properties_get(props, "rtp.session")) != NULL)
if ((str = pw_properties_get(props, "sess.name")) != NULL)
sdp->session_name = strdup(str);
if ((str = pw_properties_get(props, "rtp.destination.port")) == NULL)

14
src/modules/module-rtp-session.c
View file

@ -113,18 +113,9 @@
PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
#define PW_LOG_TOPIC_DEFAULT mod_topic
#define BUFFER_SIZE (1u<<20)
#define BUFFER_MASK (BUFFER_SIZE-1)
#define DEFAULT_FORMAT "S16BE"
#define DEFAULT_RATE "48000"
#define DEFAULT_CHANNELS "2"
#define DEFAULT_POSITION "[ FL FR ]"
#define DEFAULT_CONTROL_IP "0.0.0.0"
#define DEFAULT_CONTROL_PORT 0
#define DEFAULT_TTL 1
#define DEFAULT_MTU 1280
#define DEFAULT_LOOP false
#define USAGE "control.ip=<destination IP address, default:"DEFAULT_CONTROL_IP"> " \
@ -1474,6 +1465,7 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
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");
impl->ttl = pw_properties_get_uint32(props, "net.ttl", DEFAULT_TTL);
impl->mcast_loop = pw_properties_get_bool(props, "net.loop", DEFAULT_LOOP);
@ -1485,8 +1477,8 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
if (spa_streq(str, "audio")) {
struct spa_dict_item items[] = {
{ "audio.format", DEFAULT_FORMAT },
{ "audio.rate", DEFAULT_RATE },
{ "audio.channels", DEFAULT_CHANNELS },
{ "audio.rate", SPA_STRINGIFY(DEFAULT_RATE) },
{ "audio.channels", SPA_STRINGIFY(DEFAULT_CHANNELS) },
{ "audio.position", DEFAULT_POSITION } };
pw_properties_add(stream_props, &SPA_DICT_INIT_ARRAY(items));
}

679
src/modules/module-rtp-sink.c
View file

@ -25,7 +25,7 @@
#include <pipewire/pipewire.h>
#include <pipewire/impl.h>
#include <module-rtp/rtp.h>
#include <module-rtp/stream.h>
/** \page page_module_rtp_sink PipeWire Module: RTP sink
*
@ -104,28 +104,13 @@
PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
#define PW_LOG_TOPIC_DEFAULT mod_topic
#define BUFFER_SIZE (1u<<20)
#define BUFFER_MASK (BUFFER_SIZE-1)
#define DEFAULT_SESS_MEDIA "audio"
#define DEFAULT_FORMAT "S16BE"
#define DEFAULT_RATE 48000
#define DEFAULT_CHANNELS 2
#define DEFAULT_POSITION "[ FL FR ]"
#define DEFAULT_PORT 46000
#define DEFAULT_SOURCE_IP "0.0.0.0"
#define DEFAULT_DESTINATION_IP "224.0.0.56"
#define DEFAULT_TTL 1
#define DEFAULT_MTU 1280
#define DEFAULT_LOOP false
#define DEFAULT_DSCP 34 /* Default to AES-67 AF41 (34) */
#define DEFAULT_MIN_PTIME 2
#define DEFAULT_MAX_PTIME 20
#define DEFAULT_TS_OFFSET -1
#define USAGE "source.ip=<source IP address, default:"DEFAULT_SOURCE_IP"> " \
"destination.ip=<destination IP address, default:"DEFAULT_DESTINATION_IP"> " \
"destination.port=<int, default random beteen 46000 and 47024> " \
@ -167,24 +152,15 @@ struct impl {
struct spa_source *timer;
struct pw_properties *stream_props;
struct pw_stream *stream;
struct spa_hook stream_listener;
struct spa_io_position *io_position;
struct rtp_stream *stream;
unsigned int do_disconnect:1;
char *ifname;
char *session_name;
uint32_t mtu;
bool ttl;
bool mcast_loop;
uint32_t dscp;
float min_ptime;
float max_ptime;
uint32_t psamples;
uint32_t min_samples;
uint32_t max_samples;
struct sockaddr_storage src_addr;
socklen_t src_len;
@ -193,72 +169,29 @@ struct impl {
struct sockaddr_storage dst_addr;
socklen_t dst_len;
struct spa_audio_info info;
const struct format_info *format_info;
uint32_t rate;
uint32_t stride;
int payload;
uint16_t seq;
uint32_t ssrc;
uint32_t ts_offset;
char *ts_refclk;
struct spa_ringbuffer ring;
uint8_t buffer[BUFFER_SIZE];
int rtp_fd;
unsigned sync:1;
unsigned apple_midi:1;
};
static void stream_destroy(void *d)
{
struct impl *impl = d;
spa_hook_remove(&impl->stream_listener);
impl->stream = NULL;
}
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;
}
struct format_info {
uint32_t media_subtype;
uint32_t format;
uint32_t size;
const char *mime;
const char *media_type;
};
static const struct format_info audio_format_info[] = {
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_U8, 1, "L8", "audio" },
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_ALAW, 1, "PCMA", "audio" },
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_ULAW, 1, "PCMU", "audio" },
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S16_BE, 2, "L16", "audio" },
{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S24_BE, 3, "L24", "audio" },
{ SPA_MEDIA_SUBTYPE_control, 0, 1, "rtp-midi", "audio" },
};
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 ssize_t send_packet(struct impl *impl, struct msghdr *msg)
static void stream_send_packet(void *data, struct iovec *iov, size_t iovlen)
{
struct impl *impl = data;
struct msghdr msg;
ssize_t n;
n = sendmsg(impl->rtp_fd, msg, MSG_NOSIGNAL);
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->rtp_fd, &msg, MSG_NOSIGNAL);
if (n < 0) {
switch (errno) {
case ECONNREFUSED:
@ -266,338 +199,27 @@ static ssize_t send_packet(struct impl *impl, struct msghdr *msg)
pw_log_debug("remote end not listening");
break;
default:
pw_log_warn("sendmsg() failed, seq:%u dropped: %m",
impl->seq);
pw_log_warn("sendmsg() failed: %m");
break;
}
}
impl->seq++;
return n;
}
static void flush_audio_packets(struct impl *impl)
{
int32_t avail;
uint32_t stride, timestamp;
struct iovec iov[3];
struct msghdr msg;
struct rtp_header header;
int32_t tosend;
avail = spa_ringbuffer_get_read_index(&impl->ring, &timestamp);
tosend = impl->psamples;
if (avail < tosend)
return;
stride = impl->stride;
spa_zero(header);
header.v = 2;
header.pt = impl->payload;
header.ssrc = htonl(impl->ssrc);
iov[0].iov_base = &header;
iov[0].iov_len = sizeof(header);
msg.msg_name = NULL;
msg.msg_namelen = 0;
msg.msg_iov = iov;
msg.msg_iovlen = 3;
msg.msg_control = NULL;
msg.msg_controllen = 0;
msg.msg_flags = 0;
while (avail >= tosend) {
header.sequence_number = htons(impl->seq);
header.timestamp = htonl(impl->ts_offset + timestamp);
set_iovec(&impl->ring,
impl->buffer, BUFFER_SIZE,
(timestamp * stride) & BUFFER_MASK,
&iov[1], tosend * stride);
pw_log_trace("sending %d timestamp:%d", tosend, timestamp);
send_packet(impl, &msg);
timestamp += tosend;
avail -= tosend;
}
spa_ringbuffer_read_update(&impl->ring, timestamp);
}
static void stream_audio_process(struct impl *impl)
{
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))
timestamp = impl->io_position->clock.position;
else
timestamp = expected_timestamp;
if (impl->sync) {
if (expected_timestamp != timestamp) {
pw_log_warn("expected %u != timestamp %u", expected_timestamp, timestamp);
impl->sync = false;
} else if (filled + wanted > (int32_t)(BUFFER_SIZE / stride)) {
pw_log_warn("overrun %u + %u > %u", filled, wanted, BUFFER_SIZE / stride);
impl->sync = false;
}
}
if (!impl->sync) {
pw_log_info("sync to timestamp:%u seq:%u ts_offset:%u SSRC:%u",
timestamp, impl->seq, impl->ts_offset, impl->ssrc);
impl->ring.readindex = impl->ring.writeindex = timestamp;
memset(impl->buffer, 0, BUFFER_SIZE);
impl->sync = true;
}
spa_ringbuffer_write_data(&impl->ring,
impl->buffer,
BUFFER_SIZE,
(timestamp * stride) & BUFFER_MASK,
SPA_PTROFF(d[0].data, offs, void), wanted * stride);
timestamp += wanted;
spa_ringbuffer_write_update(&impl->ring, timestamp);
pw_stream_queue_buffer(impl->stream, buf);
flush_audio_packets(impl);
}
static int write_event(uint8_t *p, uint32_t value, void *ev, uint32_t size)
{
uint64_t buffer;
uint8_t b;
int count = 0;
buffer = value & 0x7f;
while ((value >>= 7)) {
buffer <<= 8;
buffer |= ((value & 0x7f) | 0x80);
}
do {
b = buffer & 0xff;
p[count++] = b;
buffer >>= 8;
} while (b & 0x80);
memcpy(&p[count], ev, size);
return count + size;
}
static void flush_midi_packets(struct impl *impl, struct spa_pod_sequence *sequence, uint32_t timestamp)
{
struct spa_pod_control *c;
struct rtp_header header;
struct rtp_midi_header midi_header;
struct iovec iov[3];
struct msghdr msg;
uint32_t len, prev_offset, base;
spa_zero(header);
header.v = 2;
header.pt = impl->payload;
header.ssrc = htonl(impl->ssrc);
spa_zero(midi_header);
midi_header.b = 1;
midi_header.z = 1;
iov[0].iov_base = &header;
iov[0].iov_len = sizeof(header);
iov[1].iov_base = &midi_header;
iov[1].iov_len = sizeof(midi_header);
iov[2].iov_base = impl->buffer;
iov[2].iov_len = 0;
spa_zero(msg);
msg.msg_iov = iov;
msg.msg_iovlen = 3;
prev_offset = len = base = 0;
SPA_POD_SEQUENCE_FOREACH(sequence, c) {
void *ev;
uint32_t size, delta;
if (c->type != SPA_CONTROL_Midi)
continue;
ev = SPA_POD_BODY(&c->value),
size = SPA_POD_BODY_SIZE(&c->value);
if (len > 0 && (len + size > impl->mtu ||
c->offset - base > impl->max_samples)) {
/* flush packet when we have one and when it's either
* too large or has too much data. */
midi_header.len = (len >> 8) & 0xf;
midi_header.len_b = len & 0xff;
iov[2].iov_len = len;
pw_log_debug("sending %d timestamp:%d %u %u",
len, timestamp + base,
c->offset, impl->max_samples);
send_packet(impl, &msg);
len = 0;
}
if (len == 0) {
/* start new packet */
base = prev_offset = c->offset;
header.sequence_number = htons(impl->seq);
header.timestamp = htonl(impl->ts_offset + timestamp + base);
}
delta = c->offset - prev_offset;
prev_offset = c->offset;
len += write_event(&impl->buffer[len], delta, ev, size);
}
if (len > 0) {
/* flush last packet */
midi_header.len = (len >> 8) & 0xf;
midi_header.len_b = len & 0xff;
iov[2].iov_len = len;
pw_log_debug("sending %d timestamp:%d", len, base);
send_packet(impl, &msg);
}
}
static void send_cmd(struct impl *impl)
{
uint8_t buffer[16];
struct iovec iov[3];
struct msghdr msg;
spa_zero(buffer);
buffer[0] = 0xff;
buffer[1] = 0xff;
buffer[2] = 'I';
buffer[3] = 'N';
iov[0].iov_base = buffer;
iov[0].iov_len = sizeof(buffer);
spa_zero(msg);
msg.msg_iov = iov;
msg.msg_iovlen = 1;
send_packet(impl, &msg);
}
static void stream_midi_process(void *data)
static void stream_state_changed(void *data, bool started, const char *error)
{
struct impl *impl = data;
struct pw_buffer *buf;
struct spa_data *d;
uint32_t offs, size, timestamp;
struct spa_pod *pod;
void *ptr;
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);
if (SPA_LIKELY(impl->io_position))
timestamp = impl->io_position->clock.position;
else
timestamp = 0;
ptr = SPA_PTROFF(d[0].data, offs, void);
if ((pod = spa_pod_from_data(ptr, size, 0, size)) == NULL)
goto done;
if (!spa_pod_is_sequence(pod))
goto done;
if (!impl->sync) {
pw_log_info("sync to timestamp:%u seq:%u ts_offset:%u SSRC:%u",
timestamp, impl->seq, impl->ts_offset, impl->ssrc);
impl->sync = true;
if (impl->apple_midi)
send_cmd(impl);
}
flush_midi_packets(impl, (struct spa_pod_sequence*)pod, timestamp);
done:
pw_stream_queue_buffer(impl->stream, buf);
}
static void stream_process(void *data)
{
struct impl *impl = data;
switch (impl->info.media_type) {
case SPA_MEDIA_TYPE_audio:
stream_audio_process(impl);
break;
case SPA_MEDIA_TYPE_application:
stream_midi_process(impl);
break;
}
}
static void stream_io_changed(void *data, uint32_t id, void *area, uint32_t size)
{
struct impl *impl = data;
switch (id) {
case SPA_IO_Position:
impl->io_position = area;
break;
}
}
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, unloading");
pw_impl_module_schedule_destroy(impl->module);
break;
case PW_STREAM_STATE_ERROR:
if (error) {
pw_log_error("stream error: %s", error);
break;
case PW_STREAM_STATE_PAUSED:
impl->sync = false;
break;
default:
break;
pw_impl_module_schedule_destroy(impl->module);
}
}
static const struct pw_stream_events in_stream_events = {
PW_VERSION_STREAM_EVENTS,
static const struct rtp_stream_events stream_events = {
RTP_VERSION_STREAM_EVENTS,
.destroy = stream_destroy,
.io_changed = stream_io_changed,
.state_changed = on_stream_state_changed,
.process = stream_process
.state_changed = stream_state_changed,
.send_packet = stream_send_packet,
};
static int parse_address(const char *address, uint16_t port,
@ -681,75 +303,6 @@ error:
return res;
}
static int setup_stream(struct impl *impl)
{
const struct spa_pod *params[1];
struct spa_pod_builder b;
uint32_t n_params;
uint8_t buffer[1024];
struct pw_properties *props;
enum pw_stream_flags flags;
int res, fd;
props = pw_properties_copy(impl->stream_props);
if (props == NULL)
return -errno;
if (pw_properties_get(props, PW_KEY_NODE_LATENCY) == NULL) {
pw_properties_setf(props, PW_KEY_NODE_LATENCY,
"%d/%d", impl->psamples, impl->rate);
}
pw_properties_setf(props, PW_KEY_NODE_RATE, "1/%d", impl->rate);
impl->stream = pw_stream_new(impl->core,
"rtp-sink capture", props);
if (impl->stream == NULL)
return -errno;
pw_stream_add_listener(impl->stream,
&impl->stream_listener,
&in_stream_events, impl);
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_type) {
case SPA_MEDIA_TYPE_audio:
params[n_params++] = spa_format_audio_build(&b,
SPA_PARAM_EnumFormat, &impl->info);
flags |= PW_STREAM_FLAG_AUTOCONNECT;
break;
case SPA_MEDIA_TYPE_application:
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));
break;
default:
return -EINVAL;
}
if ((res = pw_stream_connect(impl->stream,
PW_DIRECTION_INPUT,
PW_ID_ANY,
flags,
params, n_params)) < 0)
return res;
if ((fd = make_socket(&impl->src_addr, impl->src_len,
&impl->dst_addr, impl->dst_len,
impl->mcast_loop, impl->ttl,
impl->dscp)) < 0)
return fd;
impl->rtp_fd = fd;
return 0;
}
static int get_ip(const struct sockaddr_storage *sa, char *ip, size_t len)
{
if (sa->ss_family == AF_INET) {
@ -778,7 +331,7 @@ static const struct pw_proxy_events core_proxy_events = {
static void impl_destroy(struct impl *impl)
{
if (impl->stream)
pw_stream_destroy(impl->stream);
rtp_stream_destroy(impl->stream);
if (impl->core && impl->do_disconnect)
pw_core_disconnect(impl->core);
@ -793,7 +346,6 @@ static void impl_destroy(struct impl *impl)
pw_properties_free(impl->props);
free(impl->ifname);
free(impl->ts_refclk);
free(impl->session_name);
free(impl);
}
@ -826,63 +378,6 @@ static const struct pw_core_events core_events = {
.error = on_core_error,
};
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 void copy_props(struct impl *impl, struct pw_properties *props, const char *key)
{
const char *str;
@ -898,11 +393,8 @@ 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;
uint32_t id = pw_global_get_id(pw_impl_module_get_global(module));
uint32_t pid = getpid();
int64_t ts_offset;
char addr[64];
const char *str;
const char *str, *sess_name;
int res = 0;
PW_LOG_TOPIC_INIT(mod_topic);
@ -936,18 +428,16 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
impl->context = context;
impl->loop = pw_context_get_main_loop(context);
if (pw_properties_get(props, PW_KEY_NODE_VIRTUAL) == NULL)
pw_properties_set(props, PW_KEY_NODE_VIRTUAL, "true");
if (pw_properties_get(stream_props, PW_KEY_NODE_NETWORK) == NULL)
pw_properties_set(stream_props, PW_KEY_NODE_NETWORK, "true");
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, "rtp-sink-%u-%u", pid, id);
pw_properties_setf(props, PW_KEY_NODE_NAME, "rtp_session.%s", sess_name);
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));
pw_properties_setf(props, PW_KEY_NODE_DESCRIPTION, "%s", sess_name);
if (pw_properties_get(props, PW_KEY_MEDIA_NAME) == NULL)
pw_properties_set(props, PW_KEY_MEDIA_NAME, "RTP Sender Stream");
pw_properties_setf(props, PW_KEY_MEDIA_NAME, "RTP Session with %s",
sess_name);
if ((str = pw_properties_get(props, "stream.props")) != NULL)
pw_properties_update_string(stream_props, str, strlen(str));
@ -964,54 +454,12 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
copy_props(impl, props, PW_KEY_NODE_CHANNELNAMES);
copy_props(impl, props, PW_KEY_MEDIA_NAME);
copy_props(impl, props, PW_KEY_MEDIA_CLASS);
if ((str = pw_properties_get(props, "sess.media")) == NULL)
str = DEFAULT_SESS_MEDIA;
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;
}
switch (impl->info.media_type) {
case SPA_MEDIA_TYPE_audio:
parse_audio_info(impl->stream_props, &impl->info.info.raw);
impl->format_info = find_audio_format_info(&impl->info);
if (impl->format_info == NULL) {
pw_log_error("unsupported audio format:%d channels:%d",
impl->info.info.raw.format, impl->info.info.raw.channels);
res = -EINVAL;
goto out;
}
impl->stride = impl->format_info->size * impl->info.info.raw.channels;
impl->rate = impl->info.info.raw.rate;
break;
case SPA_MEDIA_TYPE_application:
impl->format_info = find_audio_format_info(&impl->info);
if (impl->format_info == NULL) {
res = -EINVAL;
goto out;
}
pw_properties_set(impl->stream_props, PW_KEY_FORMAT_DSP, "8 bit raw midi");
impl->stride = impl->format_info->size;
impl->rate = 48000;
break;
default:
spa_assert_not_reached();
break;
}
impl->payload = 127;
impl->seq = pw_rand32();
impl->ssrc = pw_rand32();
copy_props(impl, props, "net.mtu");
copy_props(impl, props, "rtp.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;
@ -1032,56 +480,17 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
goto out;
}
impl->mtu = pw_properties_get_uint32(props, "net.mtu", DEFAULT_MTU);
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);
ts_offset = pw_properties_get_int64(props, "sess.ts-offset", DEFAULT_TS_OFFSET);
impl->ts_offset = ts_offset < 0 ? pw_rand32() : ts_offset;
str = pw_properties_get(props, "sess.ts-refclk");
impl->ts_refclk = str ? strdup(str) : NULL;
str = pw_properties_get(props, "sess.min-ptime");
if (!spa_atof(str, &impl->min_ptime))
impl->min_ptime = DEFAULT_MIN_PTIME;
str = pw_properties_get(props, "sess.max-ptime");
if (!spa_atof(str, &impl->max_ptime))
impl->max_ptime = DEFAULT_MAX_PTIME;
impl->min_samples = impl->min_ptime * impl->rate / 1000;
impl->max_samples = impl->max_ptime * impl->rate / 1000;
impl->psamples = impl->mtu / impl->stride;
impl->psamples = SPA_CLAMP(impl->psamples, impl->min_samples, impl->max_samples);
if ((str = pw_properties_get(props, "sess.name")) == NULL)
pw_properties_setf(props, "sess.name", "PipeWire RTP Stream on %s",
pw_get_host_name());
str = pw_properties_get(props, "sess.name");
impl->session_name = str ? strdup(str) : NULL;
pw_properties_set(stream_props, "rtp.session", impl->session_name);
get_ip(&impl->src_addr, addr, sizeof(addr));
pw_properties_set(stream_props, "rtp.source.ip", addr);
get_ip(&impl->dst_addr, addr, sizeof(addr));
pw_properties_set(stream_props, "rtp.destination.ip", addr);
pw_properties_setf(stream_props, "rtp.destination.port", "%u", impl->dst_port);
pw_properties_setf(stream_props, "rtp.mtu", "%u", impl->mtu);
pw_properties_setf(stream_props, "rtp.ttl", "%u", impl->ttl);
pw_properties_setf(stream_props, "rtp.ptime", "%u",
impl->psamples * 1000 / impl->rate);
pw_properties_setf(stream_props, "rtp.dscp", "%u", impl->dscp);
pw_properties_setf(stream_props, "rtp.media", "%s", impl->format_info->media_type);
pw_properties_setf(stream_props, "rtp.mime", "%s", impl->format_info->mime);
pw_properties_setf(stream_props, "rtp.payload", "%u", impl->payload);
pw_properties_setf(stream_props, "rtp.rate", "%u", impl->rate);
if (impl->info.info.raw.channels > 0)
pw_properties_setf(stream_props, "rtp.channels", "%u", impl->info.info.raw.channels);
pw_properties_setf(stream_props, "rtp.ts-offset", "%u", impl->ts_offset);
if (impl->ts_refclk != NULL)
pw_properties_set(stream_props, "rtp.ts-refclk", impl->ts_refclk);
impl->core = pw_context_get_object(impl->context, PW_TYPE_INTERFACE_Core);
if (impl->core == NULL) {
@ -1106,8 +515,22 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
&impl->core_listener,
&core_events, impl);
if ((res = setup_stream(impl)) < 0)
if ((res = make_socket(&impl->src_addr, impl->src_len,
&impl->dst_addr, impl->dst_len,
impl->mcast_loop, impl->ttl)) < 0) {
pw_log_error("can't make socket: %s", spa_strerror(res));
goto out;
}
impl->rtp_fd = res;
impl->stream = rtp_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);

824
src/modules/module-rtp-source.c
View file

@ -28,7 +28,7 @@
#include <pipewire/pipewire.h>
#include <pipewire/impl.h>
#include <module-rtp/rtp.h>
#include <module-rtp/stream.h>
#ifdef __FreeBSD__
#define ifr_ifindex ifr_index
@ -97,22 +97,8 @@ PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
#define PW_LOG_TOPIC_DEFAULT mod_topic
#define DEFAULT_CLEANUP_SEC 60
#define ERROR_MSEC 2
#define DEFAULT_SESS_LATENCY 100
#define DEFAULT_SOURCE_IP "224.0.0.56"
#define DEFAULT_FORMAT "S16BE"
#define DEFAULT_RATE 48000
#define DEFAULT_CHANNELS 2
#define DEFAULT_POSITION "[ FL FR ]"
#define BUFFER_SIZE (1u<<22)
#define BUFFER_MASK (BUFFER_SIZE-1)
#define BUFFER_SIZE2 (BUFFER_SIZE>>1)
#define BUFFER_MASK2 (BUFFER_SIZE2-1)
#define USAGE "local.ifname=<local interface name to use> " \
"source.ip=<source IP address, default:"DEFAULT_SOURCE_IP"> " \
"source.port=<int, source port> " \
@ -147,549 +133,37 @@ struct impl {
char *ifname;
bool always_process;
int sess_latency_msec;
uint32_t cleanup_interval;
struct spa_source *timer;
struct spa_audio_info info;
struct pw_properties *stream_props;
struct pw_stream *stream;
struct spa_hook stream_listener;
struct rtp_stream *stream;
uint16_t src_port;
struct sockaddr_storage src_addr;
socklen_t src_len;
struct spa_source *source;
uint32_t rate;
uint32_t stride;
uint32_t expected_ssrc;
uint16_t expected_seq;
unsigned have_ssrc:1;
unsigned have_seq:1;
unsigned have_sync:1;
uint32_t ts_offset;
struct spa_ringbuffer ring;
uint8_t buffer[BUFFER_SIZE];
struct spa_io_rate_match *rate_match;
struct spa_io_position *position;
struct spa_dll dll;
double corr;
uint32_t target_buffer;
float max_error;
unsigned first:1;
unsigned receiving:1;
unsigned direct_timestamp:1;
float last_timestamp;
float last_time;
};
struct format_info {
uint32_t media_subtype;
uint32_t format;
uint32_t size;
const char *mime;
const char *media_type;
};
static uint32_t audio_get_stride(const struct spa_audio_info *info)
{
uint32_t stride = 0;
if (info->media_type != SPA_MEDIA_TYPE_audio ||
info->media_subtype != SPA_MEDIA_SUBTYPE_raw)
return 0;
switch (info->info.raw.format) {
case SPA_AUDIO_FORMAT_U8:
case SPA_AUDIO_FORMAT_ALAW:
case SPA_AUDIO_FORMAT_ULAW:
stride = 1;
break;
case SPA_AUDIO_FORMAT_S16_BE:
stride = 2;
break;
case SPA_AUDIO_FORMAT_S24_BE:
stride = 3;
break;
default:
break;
}
return stride * info->info.raw.channels;
}
static void process_audio(struct impl *impl)
{
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;
if (impl->position && impl->direct_timestamp) {
/* in direct mode, read directly from the timestamp index,
* because sender and receiver are in sync, this would keep
* target_buffer of samples available. */
spa_ringbuffer_read_update(&impl->ring,
impl->position->clock.position);
}
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;
}
if (!impl->direct_timestamp) {
/* when not using direct timestamp and clocks are not
* in sync, 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->rate_match) {
SPA_FLAG_SET(impl->rate_match->flags,
SPA_IO_RATE_MATCH_FLAG_ACTIVE);
impl->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 void receive_audio(struct impl *impl, uint8_t *packet,
uint32_t timestamp, uint32_t payload_offset, uint32_t len)
{
uint32_t plen = len - payload_offset;
uint8_t *payload = &packet[payload_offset];
uint32_t stride = impl->stride;
uint32_t samples = plen / stride;
uint32_t write, expected_write;
int32_t filled;
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 seq:%u ts_offset:%u SSRC:%u direct:%d",
write, impl->expected_seq-1, impl->ts_offset, impl->expected_ssrc,
impl->direct_timestamp);
/* 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,
payload, (samples * stride));
write += samples;
spa_ringbuffer_write_update(&impl->ring, write);
}
}
static void process_midi(struct impl *impl)
{
struct pw_buffer *buf;
struct spa_data *d;
uint32_t timestamp, duration, maxsize, read;
struct spa_pod_builder b;
struct spa_pod_frame f[1];
void *ptr;
struct spa_pod *pod;
struct spa_pod_control *c;
if ((buf = pw_stream_dequeue_buffer(impl->stream)) == NULL) {
pw_log_debug("Out of stream buffers: %m");
return;
}
d = buf->buffer->datas;
maxsize = d[0].maxsize;
/* we always use the graph position to select events, the receiver side is
* responsible for smoothing out the RTP timestamps to graph time */
duration = impl->position->clock.duration;
if (impl->position)
timestamp = impl->position->clock.position;
else
timestamp = 0;
/* we copy events into the buffer based on the rtp timestamp + delay. */
spa_pod_builder_init(&b, d[0].data, maxsize);
spa_pod_builder_push_sequence(&b, &f[0], 0);
while (true) {
int32_t avail = spa_ringbuffer_get_read_index(&impl->ring, &read);
if (avail <= 0)
break;
ptr = SPA_PTROFF(impl->buffer, read & BUFFER_MASK2, void);
if ((pod = spa_pod_from_data(ptr, avail, 0, avail)) == NULL)
goto done;
if (!spa_pod_is_sequence(pod))
goto done;
/* the ringbuffer contains series of sequences, one for each
* received packet */
SPA_POD_SEQUENCE_FOREACH((struct spa_pod_sequence*)pod, c) {
/* try to render with given delay */
uint32_t target = c->offset + impl->target_buffer;
if (timestamp != 0) {
/* skip old packets */
if (target < timestamp)
continue;
/* event for next cycle */
if (target >= timestamp + duration)
goto complete;
} else {
timestamp = target;
}
spa_pod_builder_control(&b, target - timestamp, SPA_CONTROL_Midi);
spa_pod_builder_bytes(&b,
SPA_POD_BODY(&c->value),
SPA_POD_BODY_SIZE(&c->value));
}
/* we completed a sequence (one RTP packet), advance ringbuffer
* and go to the next packet */
read += SPA_PTRDIFF(c, ptr);
spa_ringbuffer_read_update(&impl->ring, read);
}
complete:
spa_pod_builder_pop(&b, &f[0]);
if (b.state.offset > maxsize) {
pw_log_warn("overflow buffer %u %u", b.state.offset, maxsize);
b.state.offset = 0;
}
d[0].chunk->size = b.state.offset;
d[0].chunk->stride = 1;
d[0].chunk->offset = 0;
done:
pw_stream_queue_buffer(impl->stream, buf);
}
static int parse_varlen(uint8_t *p, uint32_t avail, uint32_t *result)
{
uint32_t value = 0, offs = 0;
while (offs < avail) {
uint8_t b = p[offs++];
value = (value << 7) | (b & 0x7f);
if ((b & 0x80) == 0)
break;
}
*result = value;
return offs;
}
static int get_midi_size(uint8_t *p, uint32_t avail)
{
int size;
uint32_t offs = 0, value;
switch (p[offs++]) {
case 0xc0 ... 0xdf:
size = 2;
break;
case 0x80 ... 0xbf:
case 0xe0 ... 0xef:
size = 3;
break;
case 0xff:
case 0xf0:
case 0xf7:
size = parse_varlen(&p[offs], avail - offs, &value);
size += value + 1;
break;
default:
return -EINVAL;
}
return size;
}
static double get_time(struct impl *impl)
{
struct timespec ts;
double t;
clock_gettime(CLOCK_MONOTONIC, &ts);
t = impl->position->clock.position / (double) impl->position->clock.rate.denom;
t += (SPA_TIMESPEC_TO_NSEC(&ts) - impl->position->clock.nsec) / (double)SPA_NSEC_PER_SEC;
return t;
}
static void receive_midi(struct impl *impl, uint8_t *packet,
uint32_t timestamp, uint32_t payload_offset, uint32_t plen)
{
uint32_t write;
struct rtp_midi_header *hdr;
int32_t filled;
struct spa_pod_builder b;
struct spa_pod_frame f[1];
void *ptr;
uint32_t offs = payload_offset, len, end;
bool first = true;
if (impl->direct_timestamp) {
/* in direct timestamp we attach the RTP timestamp directly on the
* midi events and render them in the corresponding cycle */
if (!impl->have_sync) {
pw_log_info("sync to timestamp:%u seq:%u ts_offset:%u SSRC:%u direct:%d",
timestamp, impl->expected_seq-1, impl->ts_offset, impl->expected_ssrc,
impl->direct_timestamp);
impl->have_sync = true;
}
} else {
/* in non-direct timestamp mode, we relate the graph clock against
* the RTP timestamps */
double ts = timestamp / (float) impl->rate;
double t = get_time(impl);
double elapsed, estimated, diff;
/* the elapsed time between RTP timestamps */
elapsed = ts - impl->last_timestamp;
/* for that elapsed time, our clock should have advanced
* by this amount since the last estimation */
estimated = impl->last_time + elapsed * impl->corr;
/* calculate the diff between estimated and current clock time in
* samples */
diff = (estimated - t) * impl->rate;
/* no sync or we drifted too far, resync */
if (!impl->have_sync || fabs(diff) > impl->target_buffer) {
impl->corr = 1.0;
spa_dll_set_bw(&impl->dll, SPA_DLL_BW_MIN, 256, impl->rate);
pw_log_info("sync to timestamp:%u seq:%u ts_offset:%u SSRC:%u direct:%d",
timestamp, impl->expected_seq-1, impl->ts_offset, impl->expected_ssrc,
impl->direct_timestamp);
impl->have_sync = true;
impl->ring.readindex = impl->ring.writeindex;
} else {
/* update our new rate correction */
impl->corr = spa_dll_update(&impl->dll, diff);
/* our current time is now the estimated time */
t = estimated;
}
pw_log_debug("%f %f %f %f", t, estimated, diff, impl->corr);
timestamp = t * impl->rate;
impl->last_timestamp = ts;
impl->last_time = t;
}
filled = spa_ringbuffer_get_write_index(&impl->ring, &write);
if (filled > (int32_t)BUFFER_SIZE2)
return;
hdr = (struct rtp_midi_header *)&packet[offs++];
len = hdr->len;
if (hdr->b) {
len = (len << 8) | hdr->len_b;
offs++;
}
end = len + offs;
if (end > plen)
return;
ptr = SPA_PTROFF(impl->buffer, write & BUFFER_MASK2, void);
/* each packet is written as a sequence of events. The offset is
* the RTP timestamp */
spa_pod_builder_init(&b, ptr, BUFFER_SIZE2 - filled);
spa_pod_builder_push_sequence(&b, &f[0], 0);
while (offs < end) {
uint32_t delta;
int size;
if (first && !hdr->z)
delta = 0;
else
offs += parse_varlen(&packet[offs], end - offs, &delta);
timestamp += delta * impl->corr;
spa_pod_builder_control(&b, timestamp, SPA_CONTROL_Midi);
size = get_midi_size(&packet[offs], end - offs);
if (size <= 0 || offs + size > end) {
pw_log_warn("invalid size (%08x) %d (%u %u)",
packet[offs], size, offs, end);
break;
}
spa_pod_builder_bytes(&b, &packet[offs], size);
offs += size;
first = false;
}
spa_pod_builder_pop(&b, &f[0]);
write += b.state.offset;
spa_ringbuffer_write_update(&impl->ring, write);
}
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->rate_match = area;
break;
case SPA_IO_Position:
impl->position = area;
break;
}
}
static void stream_destroy(void *d)
{
struct impl *impl = d;
spa_hook_remove(&impl->stream_listener);
impl->stream = NULL;
}
static void stream_process(void *data)
{
struct impl *impl = data;
switch (impl->info.media_type) {
case SPA_MEDIA_TYPE_audio:
process_audio(impl);
break;
case SPA_MEDIA_TYPE_application:
process_midi(impl);
break;
}
}
static void
on_rtp_io(void *data, int fd, uint32_t mask)
{
struct impl *impl = data;
struct rtp_header *hdr;
ssize_t len, hlen;
ssize_t len;
uint8_t buffer[2048];
if (mask & SPA_IO_IN) {
uint16_t seq;
uint32_t timestamp;
if ((len = recv(fd, buffer, sizeof(buffer), 0)) < 0)
goto receive_error;
if (len < 12)
goto short_packet;
hdr = (struct rtp_header*)buffer;
if (hdr->v != 2)
goto invalid_version;
rtp_stream_receive_packet(impl->stream, buffer, len);
hlen = 12 + hdr->cc * 4;
if (hlen > len)
goto invalid_len;
if (impl->have_ssrc && impl->expected_ssrc != hdr->ssrc)
goto unexpected_ssrc;
impl->expected_ssrc = hdr->ssrc;
impl->have_ssrc = true;
seq = ntohs(hdr->sequence_number);
if (impl->have_seq && impl->expected_seq != seq) {
pw_log_info("unexpected seq (%d != %d) SSRC:%u",
seq, impl->expected_seq, hdr->ssrc);
impl->have_sync = false;
}
impl->expected_seq = seq + 1;
impl->have_seq = true;
timestamp = ntohl(hdr->timestamp) - impl->ts_offset;
switch (impl->info.media_type) {
case SPA_MEDIA_TYPE_audio:
receive_audio(impl, buffer, timestamp, hlen, len);
break;
case SPA_MEDIA_TYPE_application:
receive_midi(impl, buffer, timestamp, hlen, len);
}
impl->receiving = true;
}
return;
@ -700,17 +174,6 @@ receive_error:
short_packet:
pw_log_warn("short packet received");
return;
invalid_version:
pw_log_warn("invalid RTP version");
spa_debug_mem(0, buffer, len);
return;
invalid_len:
pw_log_warn("invalid RTP length");
return;
unexpected_ssrc:
pw_log_warn("unexpected SSRC (expected %u != %u)",
impl->expected_ssrc, hdr->ssrc);
return;
}
static int parse_address(const char *address, uint16_t port,
@ -810,11 +273,6 @@ error:
return res;
}
static uint32_t msec_to_samples(struct impl *impl, uint32_t msec)
{
return msec * impl->rate / 1000;
}
static int stream_start(struct impl *impl)
{
int fd;
@ -851,113 +309,34 @@ static void stream_stop(struct impl *impl)
impl->source = NULL;
}
static void on_stream_state_changed(void *d, enum pw_stream_state old,
enum pw_stream_state state, const char *error)
static void stream_destroy(void *d)
{
struct impl *impl = d;
switch (state) {
case PW_STREAM_STATE_UNCONNECTED:
pw_log_info("stream disconnected, unloading");
pw_impl_module_schedule_destroy(impl->module);
break;
case PW_STREAM_STATE_ERROR:
pw_log_error("stream error: %s", 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);
break;
default:
break;
}
impl->stream = NULL;
}
static const struct pw_stream_events out_stream_events = {
PW_VERSION_STREAM_EVENTS,
.destroy = stream_destroy,
.state_changed = on_stream_state_changed,
.io_changed = stream_io_changed,
.process = stream_process
};
static int setup_stream(struct impl *impl)
static void stream_state_changed(void *data, bool started, const char *error)
{
const struct spa_pod *params[1];
struct spa_pod_builder b;
uint32_t n_params;
enum pw_stream_flags flags;
uint8_t buffer[1024];
struct pw_properties *props;
int res;
struct impl *impl = data;
impl->first = true;
props = pw_properties_copy(impl->stream_props);
if (props == NULL) {
res = -errno;
goto error;
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 RTP stream: %m");
} else {
if (!impl->always_process)
stream_stop(impl);
}
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(impl->core,
"rtp-source playback", props);
if (impl->stream == NULL) {
res = -errno;
pw_log_error("can't create stream: %m");
goto error;
}
pw_stream_add_listener(impl->stream,
&impl->stream_listener,
&out_stream_events, impl);
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_type) {
case SPA_MEDIA_TYPE_audio:
params[n_params++] = spa_format_audio_build(&b,
SPA_PARAM_EnumFormat, &impl->info);
flags |= PW_STREAM_FLAG_AUTOCONNECT;
break;
case SPA_MEDIA_TYPE_application:
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));
break;
default:
return -EINVAL;
}
if ((res = pw_stream_connect(impl->stream,
PW_DIRECTION_OUTPUT,
PW_ID_ANY,
flags,
params, n_params)) < 0) {
pw_log_error("can't connect stream: %s", spa_strerror(res));
goto error;
}
if (impl->always_process &&
(res = stream_start(impl)) < 0)
goto error;
return 0;
error:
return res;
}
static const struct rtp_stream_events stream_events = {
RTP_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;
@ -986,7 +365,7 @@ static const struct pw_proxy_events core_proxy_events = {
static void impl_destroy(struct impl *impl)
{
if (impl->stream)
pw_stream_destroy(impl->stream);
rtp_stream_destroy(impl->stream);
if (impl->source)
pw_loop_destroy_source(impl->data_loop, impl->source);
@ -1031,63 +410,6 @@ static const struct pw_core_events core_events = {
.error = on_core_error,
};
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 void copy_props(struct impl *impl, struct pw_properties *props, const char *key)
{
const char *str;
@ -1101,10 +423,8 @@ SPA_EXPORT
int pipewire__module_init(struct pw_impl_module *module, const char *args)
{
struct pw_context *context = pw_impl_module_get_context(module);
uint32_t id = pw_global_get_id(pw_impl_module_get_global(module));
uint32_t pid = getpid();
struct impl *impl;
const char *str;
const char *str, *sess_name;
struct timespec value, interval;
struct pw_properties *props, *stream_props;
int res = 0;
@ -1131,18 +451,16 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
impl->loop = pw_context_get_main_loop(context);
impl->data_loop = pw_data_loop_get_loop(pw_context_get_data_loop(context));
if (pw_properties_get(stream_props, PW_KEY_NODE_VIRTUAL) == NULL)
pw_properties_set(stream_props, PW_KEY_NODE_VIRTUAL, "true");
if (pw_properties_get(stream_props, PW_KEY_NODE_NETWORK) == NULL)
pw_properties_set(stream_props, PW_KEY_NODE_NETWORK, "true");
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, "rtp-source-%u-%u", pid, id);
pw_properties_setf(props, PW_KEY_NODE_NAME, "rtp_session.%s", sess_name);
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));
pw_properties_setf(props, PW_KEY_NODE_DESCRIPTION, "%s", sess_name);
if (pw_properties_get(props, PW_KEY_MEDIA_NAME) == NULL)
pw_properties_set(props, PW_KEY_MEDIA_NAME, "RTP Receiver Stream");
pw_properties_setf(props, PW_KEY_MEDIA_NAME, "RTP Session with %s",
sess_name);
if ((str = pw_properties_get(props, "stream.props")) != NULL)
pw_properties_update_string(stream_props, str, strlen(str));
@ -1159,46 +477,12 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
copy_props(impl, props, PW_KEY_NODE_CHANNELNAMES);
copy_props(impl, props, PW_KEY_MEDIA_NAME);
copy_props(impl, props, PW_KEY_MEDIA_CLASS);
impl->info.media_type = SPA_MEDIA_TYPE_audio;
impl->info.media_subtype = SPA_MEDIA_SUBTYPE_raw;
if ((str = pw_properties_get(stream_props, "rtp.media")) != NULL) {
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;
}
}
switch (impl->info.media_type) {
case SPA_MEDIA_TYPE_audio:
parse_audio_info(stream_props, &impl->info.info.raw);
impl->stride = audio_get_stride(&impl->info);
if (impl->stride == 0) {
pw_log_error("unsupported audio format:%d channels:%d",
impl->info.info.raw.format, impl->info.info.raw.channels);
res = -EINVAL;
goto out;
}
impl->rate = impl->info.info.raw.rate;
break;
case SPA_MEDIA_TYPE_application:
pw_properties_set(stream_props, PW_KEY_FORMAT_DSP, "8 bit raw midi");
impl->stride = 1;
impl->rate = 48000;
break;
default:
spa_assert_not_reached();
break;
}
copy_props(impl, props, "net.mtu");
copy_props(impl, props, "rtp.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;
@ -1220,32 +504,6 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
impl->cleanup_interval = pw_properties_get_uint32(props,
"cleanup.sec", DEFAULT_CLEANUP_SEC);
if ((str = pw_properties_get(props, "sess.name")) != NULL) {
pw_properties_set(stream_props, "rtp.session", str);
if (pw_properties_get(stream_props, PW_KEY_MEDIA_NAME) == NULL)
pw_properties_setf(stream_props, PW_KEY_MEDIA_NAME, "RTP Stream (%s)", str);
if (pw_properties_get(stream_props, PW_KEY_NODE_NAME) == NULL)
pw_properties_setf(stream_props, PW_KEY_NODE_NAME, "%s", str);
} else {
if (pw_properties_get(stream_props, PW_KEY_MEDIA_NAME) == NULL)
pw_properties_set(stream_props, PW_KEY_MEDIA_NAME, "RTP Stream");
}
impl->ts_offset = pw_properties_get_int64(stream_props, "sess.ts-offset", 0);
pw_properties_setf(stream_props, "rtp.ts-offset", "%u", impl->ts_offset);
impl->direct_timestamp = pw_properties_get_bool(stream_props,
"sess.ts-direct", false);
impl->sess_latency_msec = pw_properties_get_uint32(stream_props,
"sess.latency.msec", DEFAULT_SESS_LATENCY);
impl->target_buffer = msec_to_samples(impl, impl->sess_latency_msec);
impl->max_error = msec_to_samples(impl, ERROR_MSEC);
pw_properties_setf(stream_props, PW_KEY_NODE_RATE, "1/%d", impl->rate);
pw_properties_setf(stream_props, PW_KEY_NODE_LATENCY, "%d/%d",
impl->target_buffer / 2, impl->rate);
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);
@ -1281,8 +539,14 @@ int pipewire__module_init(struct pw_impl_module *module, const char *args)
interval.tv_nsec = 0;
pw_loop_update_timer(impl->loop, impl->timer, &value, &interval, false);
if ((res = setup_stream(impl)) < 0)
impl->stream = rtp_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);

24
src/modules/module-rtp/stream.c
View file

@ -21,23 +21,11 @@
#include <module-rtp/stream.h>
#include <module-rtp/apple-midi.h>
#define DEFAULT_FORMAT "S16BE"
#define DEFAULT_RATE 48000
#define DEFAULT_CHANNELS 2
#define DEFAULT_POSITION "[ FL FR ]"
#define BUFFER_SIZE (1u<<22)
#define BUFFER_MASK (BUFFER_SIZE-1)
#define BUFFER_SIZE2 (BUFFER_SIZE>>1)
#define BUFFER_MASK2 (BUFFER_SIZE2-1)
#define DEFAULT_MTU 1280
#define DEFAULT_MIN_PTIME 2
#define DEFAULT_MAX_PTIME 20
#define ERROR_MSEC 2
#define DEFAULT_SESS_LATENCY 100
#define rtp_stream_emit(s,m,v,...) spa_hook_list_call(&s->listener_list, \
struct rtp_stream_events, m, v, ##__VA_ARGS__)
#define rtp_stream_emit_destroy(s) rtp_stream_emit(s, destroy, 0)
@ -357,6 +345,8 @@ struct rtp_stream *rtp_stream_new(struct pw_core *core,
impl->payload = pw_properties_get_uint32(props, "rtp.payload", impl->payload);
impl->mtu = pw_properties_get_uint32(props, "net.mtu", DEFAULT_MTU);
impl->seq = pw_rand32();
str = pw_properties_get(props, "sess.min-ptime");
if (!spa_atof(str, &min_ptime))
min_ptime = DEFAULT_MIN_PTIME;
@ -382,6 +372,16 @@ struct rtp_stream *rtp_stream_new(struct pw_core *core,
pw_properties_setf(props, "net.mtu", "%u", impl->mtu);
pw_properties_setf(props, "rtp.ptime", "%u",
impl->psamples * 1000 / impl->rate);
pw_properties_setf(props, "rtp.media", "%s", impl->format_info->media_type);
pw_properties_setf(props, "rtp.mime", "%s", impl->format_info->mime);
pw_properties_setf(props, "rtp.payload", "%u", impl->payload);
pw_properties_setf(props, "rtp.rate", "%u", impl->rate);
if (impl->info.info.raw.channels > 0)
pw_properties_setf(props, "rtp.channels", "%u", impl->info.info.raw.channels);
if ((str = pw_properties_get(props, "sess.ts-refclk")) != NULL) {
pw_properties_setf(props, "rtp.ts-offset", "%u", impl->ts_offset);
pw_properties_set(props, "rtp.ts-refclk", str);
}
spa_dll_init(&impl->dll);
spa_dll_set_bw(&impl->dll, SPA_DLL_BW_MIN, 128, impl->rate);

12
src/modules/module-rtp/stream.h
View file

@ -11,6 +11,18 @@ extern "C" {
struct rtp_stream;
#define DEFAULT_FORMAT "S16BE"
#define DEFAULT_RATE 48000
#define DEFAULT_CHANNELS 2
#define DEFAULT_POSITION "[ FL FR ]"
#define ERROR_MSEC 2
#define DEFAULT_SESS_LATENCY 100
#define DEFAULT_MTU 1280
#define DEFAULT_MIN_PTIME 2
#define DEFAULT_MAX_PTIME 20
struct rtp_stream_events {
#define RTP_VERSION_STREAM_EVENTS 0
uint32_t version;