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

module-rtp: add new rtp-session module

Browse source 
The module uses the apple session setup for managing peer connections.

Make a generic rtp stream object, make midi and audio implementations.
This commit is contained in:
Wim Taymans 2023-02-28 16:14:19 +01:00
parent c36d9bef38
commit 7da031c969
6 changed files with 2401 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

10
src/modules/meson.build
View file

@ -536,6 +536,16 @@ pipewire_module_rtp_sink = shared_library('pipewire-module-rtp-sink',
dependencies : [mathlib, dl_lib, rt_lib, pipewire_dep],
)
pipewire_module_rtp_sink = shared_library('pipewire-module-rtp-session',
[ 'module-rtp/stream.c',
'module-rtp-session.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',

1123
src/modules/module-rtp-session.c Normal file
View file

File diff suppressed because it is too large Load diff

303
src/modules/module-rtp/audio.c Normal file
View file

@ -0,0 +1,303 @@
/* PipeWire */
/* SPDX-FileCopyrightText: Copyright © 2022 Wim Taymans <wim.taymans@gmail.com> */
/* SPDX-License-Identifier: MIT */
static void process_audio_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;
if (impl->io_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->io_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->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 void receive_rtp_audio(struct impl *impl, uint8_t *buffer, ssize_t len)
{
struct rtp_header *hdr;
ssize_t hlen, plen;
uint16_t seq;
uint32_t timestamp, samples, write, expected_write;
uint32_t stride = impl->stride;
int32_t filled;
if (len < 12)
goto short_packet;
hdr = (struct rtp_header*)buffer;
if (hdr->v != 2)
goto invalid_version;
hlen = 12 + hdr->cc * 4;
if (hlen > len)
goto invalid_len;
if (impl->have_ssrc && impl->ssrc != hdr->ssrc)
goto unexpected_ssrc;
impl->ssrc = hdr->ssrc;
impl->have_ssrc = true;
seq = ntohs(hdr->sequence_number);
if (impl->have_seq && impl->seq != seq) {
pw_log_info("unexpected seq (%d != %d) SSRC:%u",
seq, impl->seq, hdr->ssrc);
impl->have_sync = false;
}
impl->seq = seq + 1;
impl->have_seq = true;
timestamp = ntohl(hdr->timestamp) - impl->ts_offset;
impl->receiving = true;
plen = len - hlen;
samples = plen / stride;
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->seq-1, impl->ts_offset, impl->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,
&buffer[hlen], (samples * stride));
write += samples;
spa_ringbuffer_write_update(&impl->ring, write);
}
return;
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->ssrc, hdr->ssrc);
return;
}
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 flush_audio_packets(struct impl *impl)
{
int32_t avail;
uint32_t stride, timestamp;
struct iovec iov[3];
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);
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);
rtp_stream_emit_send_packet(impl, iov, 3);
impl->seq++;
timestamp += tosend;
avail -= tosend;
}
spa_ringbuffer_read_update(&impl->ring, timestamp);
}
static void process_audio_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))
timestamp = impl->io_position->clock.position;
else
timestamp = expected_timestamp;
if (impl->have_sync) {
if (expected_timestamp != timestamp) {
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;
}
}
if (!impl->have_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->have_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);
}

472
src/modules/module-rtp/midi.c Normal file
View file

@ -0,0 +1,472 @@
/* PipeWire */
/* SPDX-FileCopyrightText: Copyright © 2022 Wim Taymans <wim.taymans@gmail.com> */
/* SPDX-License-Identifier: MIT */
static void process_midi_playback(void *data)
{
struct impl *impl = data;
struct pw_buffer *buf;
struct spa_data *d;
uint32_t timestamp, duration, maxsize, read, rate;
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->io_position->clock.duration;
if (impl->io_position) {
timestamp = impl->io_position->clock.position;
rate = impl->io_position->clock.rate.denom;
} else {
timestamp = 0;
rate = impl->rate;
}
/* 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;
target = (uint64_t)target * rate / impl->rate;
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;
struct spa_io_position *pos;
double t;
clock_gettime(CLOCK_MONOTONIC, &ts);
if ((pos = impl->io_position) != NULL) {
t = pos->clock.position / (double) pos->clock.rate.denom;
t += (SPA_TIMESPEC_TO_NSEC(&ts) - pos->clock.nsec) / (double)SPA_NSEC_PER_SEC;
} else {
t = SPA_TIMESPEC_TO_NSEC(&ts);
}
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->seq-1, impl->ts_offset, impl->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->seq-1, impl->ts_offset, impl->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) {
pw_log_warn("overflow");
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) {
pw_log_warn("invalid packet %d > %d", 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 receive_rtp_midi(struct impl *impl, uint8_t *buffer, ssize_t len)
{
struct rtp_header *hdr;
ssize_t hlen;
uint16_t seq;
uint32_t timestamp;
if (len < 12)
goto short_packet;
hdr = (struct rtp_header*)buffer;
if (hdr->v != 2)
goto invalid_version;
hlen = 12 + hdr->cc * 4;
if (hlen > len)
goto invalid_len;
if (impl->have_ssrc && impl->ssrc != hdr->ssrc)
goto unexpected_ssrc;
impl->ssrc = hdr->ssrc;
impl->have_ssrc = true;
seq = ntohs(hdr->sequence_number);
if (impl->have_seq && impl->seq != seq) {
pw_log_info("unexpected seq (%d != %d) SSRC:%u",
seq, impl->seq, hdr->ssrc);
impl->have_sync = false;
}
impl->seq = seq + 1;
impl->have_seq = true;
timestamp = ntohl(hdr->timestamp) - impl->ts_offset;
impl->receiving = true;
receive_midi(impl, buffer, timestamp, hlen, len);
return;
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->ssrc, hdr->ssrc);
return;
}
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];
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);
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;
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->psamples)) {
/* flush packet when we have one and when it's either
* too large or has too much data. */
if (len < 16) {
midi_header.b = 0;
midi_header.len = len;
iov[1].iov_len = sizeof(midi_header) - 1;
} else {
midi_header.b = 1;
midi_header.len = (len >> 8) & 0xf;
midi_header.len_b = len & 0xff;
iov[1].iov_len = sizeof(midi_header);
}
iov[2].iov_len = len;
pw_log_debug("sending %d timestamp:%d %u %u",
len, timestamp + base,
c->offset, impl->psamples);
rtp_stream_emit_send_packet(impl, iov, 3);
impl->seq++;
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);
memcpy(&impl->buffer[len], ev, size);
len += size;
} else {
delta = c->offset - prev_offset;
prev_offset = c->offset;
len += write_event(&impl->buffer[len], delta, ev, size);
}
}
if (len > 0) {
/* flush last packet */
if (len < 16) {
midi_header.b = 0;
midi_header.len = len;
iov[1].iov_len = sizeof(midi_header) - 1;
} else {
midi_header.b = 1;
midi_header.len = (len >> 8) & 0xf;
midi_header.len_b = len & 0xff;
iov[1].iov_len = sizeof(midi_header);
}
iov[2].iov_len = len;
pw_log_debug("sending %d timestamp:%d", len, base);
rtp_stream_emit_send_packet(impl, iov, 3);
impl->seq++;
}
}
static void process_midi_capture(void *data)
{
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->have_sync) {
pw_log_info("sync to timestamp:%u seq:%u ts_offset:%u SSRC:%u",
timestamp, impl->seq, impl->ts_offset, impl->ssrc);
impl->have_sync = true;
}
flush_midi_packets(impl, (struct spa_pod_sequence*)pod, timestamp);
done:
pw_stream_queue_buffer(impl->stream, buf);
}

455
src/modules/module-rtp/stream.c Normal file
View file

@ -0,0 +1,455 @@
/* 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 <pipewire/pipewire.h>
#include <pipewire/impl.h>
#include <module-rtp/rtp.h>
#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)
#define rtp_stream_emit_state_changed(s,n,e) rtp_stream_emit(s, state_changed,0,n,e)
#define rtp_stream_emit_send_packet(s,i,l) rtp_stream_emit(s, send_packet,0,i,l)
struct impl {
struct spa_audio_info 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;
uint32_t rate;
uint32_t stride;
uint8_t payload;
uint32_t ssrc;
uint16_t seq;
unsigned have_ssrc:1;
unsigned have_seq:1;
uint32_t ts_offset;
uint32_t psamples;
uint32_t mtu;
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 direct_timestamp:1;
unsigned always_process:1;
unsigned started:1;
unsigned have_sync:1;
unsigned receiving:1;
unsigned first:1;
void (*receive_rtp)(struct impl *impl, uint8_t *buffer, ssize_t len);
};
#include "module-rtp/audio.c"
#include "module-rtp/midi.c"
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 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;
rtp_stream_emit_state_changed(impl, true, NULL);
impl->started = true;
return 0;
}
static int stream_stop(struct impl *impl)
{
if (!impl->started)
return 0;
rtp_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);
rtp_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 rtp_stream *rtp_stream_new(struct pw_core *core,
enum pw_direction direction, struct pw_properties *props,
const struct rtp_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;
impl->info.media_type = SPA_MEDIA_TYPE_audio;
impl->info.media_subtype = SPA_MEDIA_SUBTYPE_raw;
if ((str = pw_properties_get(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;
impl->payload = 127;
}
else if (spa_streq(str, "midi")) {
impl->info.media_type = SPA_MEDIA_TYPE_application;
impl->info.media_subtype = SPA_MEDIA_SUBTYPE_control;
impl->payload = 0x61;
}
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(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(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;
}
pw_properties_setf(props, "rtp.media", "%s", impl->format_info->media_type);
pw_properties_setf(props, "rtp.mime", "%s", impl->format_info->mime);
if (direction == PW_DIRECTION_INPUT)
impl->ssrc = pw_properties_get_uint32(props, "rtp.sender-ssrc", pw_rand32());
else
impl->ssrc = pw_properties_get_uint32(props, "rtp.receiver-ssrc", pw_rand32());
impl->payload = pw_properties_get_uint32(props, "rtp.payload", impl->payload);
impl->mtu = pw_properties_get_uint32(props, "rtp.mtu", DEFAULT_MTU);
str = pw_properties_get(props, "rtp.min-ptime");
if (!spa_atof(str, &min_ptime))
min_ptime = DEFAULT_MIN_PTIME;
str = pw_properties_get(props, "rtp.max-ptime");
if (!spa_atof(str, &max_ptime))
max_ptime = DEFAULT_MAX_PTIME;
min_samples = min_ptime * impl->rate / 1000;
max_samples = max_ptime * impl->rate / 1000;
impl->psamples = impl->mtu / impl->stride;
impl->psamples = SPA_CLAMP(impl->psamples, min_samples, max_samples);
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);
pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%d/%d",
impl->target_buffer / 2, impl->rate);
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, "rtp-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_type) {
case SPA_MEDIA_TYPE_audio:
params[n_params++] = spa_format_audio_build(&b,
SPA_PARAM_EnumFormat, &impl->info);
flags |= PW_STREAM_FLAG_AUTOCONNECT;
if (direction == SPA_DIRECTION_INPUT)
impl->stream_events.process = process_audio_capture;
else
impl->stream_events.process = process_audio_playback;
impl->receive_rtp = receive_rtp_audio;
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));
if (direction == SPA_DIRECTION_INPUT)
impl->stream_events.process = process_midi_capture;
else
impl->stream_events.process = process_midi_playback;
impl->receive_rtp = receive_rtp_midi;
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 rtp_stream*)impl;
out:
pw_properties_free(props);
errno = -res;
return NULL;
}
void rtp_stream_destroy(struct rtp_stream *s)
{
struct impl *impl = (struct impl*)s;
rtp_stream_emit_destroy(impl);
if (impl->stream)
pw_stream_destroy(impl->stream);
spa_hook_list_clean(&impl->listener_list);
free(impl);
}
int rtp_stream_receive_packet(struct rtp_stream *s, uint8_t *buffer, size_t len)
{
struct impl *impl = (struct impl*)s;
impl->receive_rtp(impl, buffer, len);
return 0;
}

38
src/modules/module-rtp/stream.h Normal file
View file

@ -0,0 +1,38 @@
/* PipeWire */
/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
/* SPDX-License-Identifier: MIT */
#ifndef PIPEWIRE_RTP_STREAM_H
#define PIPEWIRE_RTP_STREAM_H
#ifdef __cplusplus
extern "C" {
#endif
struct rtp_stream;
struct rtp_stream_events {
#define RTP_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);
};
struct rtp_stream *rtp_stream_new(struct pw_core *core,
enum pw_direction direction, struct pw_properties *props,
const struct rtp_stream_events *events, void *data);
void rtp_stream_destroy(struct rtp_stream *s);
int rtp_stream_receive_packet(struct rtp_stream *s, uint8_t *buffer, size_t len);
#ifdef __cplusplus
}
#endif
#endif /* PIPEWIRE_RTP_STREAM_H */