mirrors/pulseaudio
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pulseaudio/pulseaudio.git synced 2025-11-05 13:29:57 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

port module-rtp-send.c to lock-free core

Browse source 
git-svn-id: file:///home/lennart/svn/public/pulseaudio/branches/lennart@1855 fefdeb5f-60dc-0310-8127-8f9354f1896f
This commit is contained in:
Lennart Poettering 2007-09-18 17:41:51 +00:00
parent 08d4b237a3
commit a558e9312e
7 changed files with 353 additions and 266 deletions
Download patch file Download diff file Expand all files Collapse all files

26
src/Makefile.am
View file

@ -947,11 +947,11 @@ modlibexec_LTLIBRARIES += \
# module-esound-sink.la # module-esound-sink.la
# See comment at librtp.la above # See comment at librtp.la above
#if !OS_IS_WIN32 if !OS_IS_WIN32
#modlibexec_LTLIBRARIES += \ modlibexec_LTLIBRARIES += \
# module-rtp-send.la \ module-rtp-send.la \
# module-rtp-recv.la module-rtp-recv.la
#endif endif
if HAVE_AF_UNIX if HAVE_AF_UNIX
modlibexec_LTLIBRARIES += \ modlibexec_LTLIBRARIES += \
@ -1342,15 +1342,15 @@ module_suspend_on_idle_la_LIBADD = $(AM_LIBADD) libpulsecore.la
module_suspend_on_idle_la_CFLAGS = $(AM_CFLAGS) module_suspend_on_idle_la_CFLAGS = $(AM_CFLAGS)
# RTP modules # RTP modules
#module_rtp_send_la_SOURCES = modules/rtp/module-rtp-send.c module_rtp_send_la_SOURCES = modules/rtp/module-rtp-send.c
#module_rtp_send_la_LDFLAGS = -module -avoid-version module_rtp_send_la_LDFLAGS = -module -avoid-version
#module_rtp_send_la_LIBADD = $(AM_LIBADD) libpulsecore.la librtp.la module_rtp_send_la_LIBADD = $(AM_LIBADD) libpulsecore.la librtp.la
#module_rtp_send_la_CFLAGS = $(AM_CFLAGS) module_rtp_send_la_CFLAGS = $(AM_CFLAGS)
#module_rtp_recv_la_SOURCES = modules/rtp/module-rtp-recv.c module_rtp_recv_la_SOURCES = modules/rtp/module-rtp-recv.c
#module_rtp_recv_la_LDFLAGS = -module -avoid-version module_rtp_recv_la_LDFLAGS = -module -avoid-version
#module_rtp_recv_la_LIBADD = $(AM_LIBADD) libpulsecore.la librtp.la module_rtp_recv_la_LIBADD = $(AM_LIBADD) libpulsecore.la librtp.la
#module_rtp_recv_la_CFLAGS = $(AM_CFLAGS) module_rtp_recv_la_CFLAGS = $(AM_CFLAGS)
# JACK # JACK

309
src/modules/rtp/module-rtp-recv.c
View file

@ -24,7 +24,6 @@
#include <config.h> #include <config.h>
#endif #endif
#include <assert.h>
#include <stdio.h> #include <stdio.h>
#include <sys/socket.h> #include <sys/socket.h>
#include <netinet/in.h> #include <netinet/in.h>
@ -47,6 +46,10 @@
#include <pulsecore/modargs.h> #include <pulsecore/modargs.h>
#include <pulsecore/namereg.h> #include <pulsecore/namereg.h>
#include <pulsecore/sample-util.h> #include <pulsecore/sample-util.h>
#include <pulsecore/macro.h>
#include <pulsecore/atomic.h>
#include <pulsecore/rtclock.h>
#include <pulsecore/atomic.h>
#include "module-rtp-recv-symdef.h" #include "module-rtp-recv-symdef.h"
@ -66,7 +69,7 @@ PA_MODULE_USAGE(
#define DEFAULT_SAP_ADDRESS "224.0.0.56" #define DEFAULT_SAP_ADDRESS "224.0.0.56"
#define MEMBLOCKQ_MAXLENGTH (1024*170) #define MEMBLOCKQ_MAXLENGTH (1024*170)
#define MAX_SESSIONS 16 #define MAX_SESSIONS 16
#define DEATH_TIMEOUT 20000000 #define DEATH_TIMEOUT 20
static const char* const valid_modargs[] = { static const char* const valid_modargs[] = {
"sink", "sink",
@ -76,102 +79,113 @@ static const char* const valid_modargs[] = {
struct session { struct session {
struct userdata *userdata; struct userdata *userdata;
PA_LLIST_FIELDS(struct session);
pa_sink_input *sink_input; pa_sink_input *sink_input;
pa_memblockq *memblockq; pa_memblockq *memblockq;
pa_time_event *death_event; pa_bool_t first_packet;
int first_packet;
uint32_t ssrc; uint32_t ssrc;
uint32_t offset; uint32_t offset;
struct pa_sdp_info sdp_info; struct pa_sdp_info sdp_info;
pa_rtp_context rtp_context; pa_rtp_context rtp_context;
pa_io_event* rtp_event;
pa_rtpoll_item *rtpoll_item;
pa_atomic_t timestamp;
}; };
struct userdata { struct userdata {
pa_module *module; pa_module *module;
pa_core *core;
pa_sap_context sap_context; pa_sap_context sap_context;
pa_io_event* sap_event; pa_io_event* sap_event;
pa_hashmap *by_origin; pa_time_event *check_death_event;
char *sink_name; char *sink_name;
PA_LLIST_HEAD(struct session, sessions);
pa_hashmap *by_origin;
int n_sessions; int n_sessions;
}; };
static void session_free(struct session *s, int from_hash); static void session_free(struct session *s);
static int sink_input_peek(pa_sink_input *i, pa_memchunk *chunk) { /* Called from I/O thread context */
static int sink_input_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct session *s = PA_SINK_INPUT(o)->userdata;
switch (code) {
case PA_SINK_INPUT_MESSAGE_GET_LATENCY:
*((pa_usec_t*) data) = pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &s->sink_input->sample_spec);
/* Fall through, the default handler will add in the extra
* latency added by the resampler */
break;
}
return pa_sink_input_process_msg(o, code, data, offset, chunk);
}
/* Called from I/O thread context */
static int sink_input_peek(pa_sink_input *i, size_t length, pa_memchunk *chunk) {
struct session *s; struct session *s;
assert(i); pa_sink_input_assert_ref(i);
s = i->userdata; pa_assert_se(s = i->userdata);
return pa_memblockq_peek(s->memblockq, chunk); return pa_memblockq_peek(s->memblockq, chunk);
} }
static void sink_input_drop(pa_sink_input *i, const pa_memchunk *chunk, size_t length) { /* Called from I/O thread context */
static void sink_input_drop(pa_sink_input *i, size_t length) {
struct session *s; struct session *s;
assert(i); pa_sink_input_assert_ref(i);
s = i->userdata; pa_assert_se(s = i->userdata);
pa_memblockq_drop(s->memblockq, chunk, length); pa_memblockq_drop(s->memblockq, length);
} }
/* Called from main context */
static void sink_input_kill(pa_sink_input* i) { static void sink_input_kill(pa_sink_input* i) {
struct session *s; struct session *s;
assert(i); pa_sink_input_assert_ref(i);
s = i->userdata; pa_assert_se(s = i->userdata);
session_free(s, 1); session_free(s);
} }
static pa_usec_t sink_input_get_latency(pa_sink_input *i) { /* Called from I/O thread context */
struct session *s; static int rtpoll_work_cb(pa_rtpoll_item *i) {
assert(i);
s = i->userdata;
return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &i->sample_spec);
}
static void rtp_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event_flags_t flags, void *userdata) {
struct session *s = userdata;
pa_memchunk chunk; pa_memchunk chunk;
int64_t k, j, delta; int64_t k, j, delta;
struct timeval tv; struct timespec now;
struct session *s;
assert(m); pa_assert_se(s = pa_rtpoll_item_get_userdata(i));
assert(e);
assert(s);
assert(fd == s->rtp_context.fd);
assert(flags == PA_IO_EVENT_INPUT);
if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->core->mempool) < 0) if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->module->core->mempool) < 0)
return; return 0;
if (s->sdp_info.payload != s->rtp_context.payload) { if (s->sdp_info.payload != s->rtp_context.payload) {
pa_memblock_unref(chunk.memblock); pa_memblock_unref(chunk.memblock);
return; return 0;
} }
if (!s->first_packet) { if (!s->first_packet) {
s->first_packet = 1; s->first_packet = TRUE;
s->ssrc = s->rtp_context.ssrc; s->ssrc = s->rtp_context.ssrc;
s->offset = s->rtp_context.timestamp; s->offset = s->rtp_context.timestamp;
if (s->ssrc == s->userdata->core->cookie) if (s->ssrc == s->userdata->module->core->cookie)
pa_log_warn("WARNING! Detected RTP packet loop!"); pa_log_warn("Detected RTP packet loop!");
} else { } else {
if (s->ssrc != s->rtp_context.ssrc) { if (s->ssrc != s->rtp_context.ssrc) {
pa_memblock_unref(chunk.memblock); pa_memblock_unref(chunk.memblock);
return; return 0;
} }
} }
@ -197,26 +211,49 @@ static void rtp_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event
pa_memblock_unref(chunk.memblock); pa_memblock_unref(chunk.memblock);
/* Reset death timer */ pa_rtclock_get(&now);
pa_gettimeofday(&tv); pa_atomic_store(&s->timestamp, now.tv_sec);
pa_timeval_add(&tv, DEATH_TIMEOUT);
m->time_restart(s->death_event, &tv); return 1;
} }
static void death_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *tv, void *userdata) { /* Called from I/O thread context */
struct session *s = userdata; static void sink_input_attach(pa_sink_input *i) {
struct session *s;
struct pollfd *p;
assert(m); pa_sink_input_assert_ref(i);
assert(t); pa_assert_se(s = i->userdata);
assert(tv);
assert(s);
session_free(s, 1); pa_assert(!s->rtpoll_item);
s->rtpoll_item = pa_rtpoll_item_new(i->sink->rtpoll, PA_RTPOLL_LATE, 1);
p = pa_rtpoll_item_get_pollfd(s->rtpoll_item, NULL);
p->fd = s->rtp_context.fd;
p->events = POLLIN;
p->revents = 0;
pa_rtpoll_item_set_work_callback(s->rtpoll_item, rtpoll_work_cb);
pa_rtpoll_item_set_userdata(s->rtpoll_item, s);
}
/* Called from I/O thread context */
static void sink_input_detach(pa_sink_input *i) {
struct session *s;
pa_sink_input_assert_ref(i);
pa_assert_se(s = i->userdata);
pa_assert(s->rtpoll_item);
pa_rtpoll_item_free(s->rtpoll_item);
s->rtpoll_item = NULL;
} }
static int mcast_socket(const struct sockaddr* sa, socklen_t salen) { static int mcast_socket(const struct sockaddr* sa, socklen_t salen) {
int af, fd = -1, r, one; int af, fd = -1, r, one;
pa_assert(sa);
pa_assert(salen > 0);
af = sa->sa_family; af = sa->sa_family;
if ((fd = socket(af, SOCK_DGRAM, 0)) < 0) { if ((fd = socket(af, SOCK_DGRAM, 0)) < 0) {
pa_log("Failed to create socket: %s", pa_cstrerror(errno)); pa_log("Failed to create socket: %s", pa_cstrerror(errno));
@ -262,27 +299,34 @@ fail:
static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_info) { static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_info) {
struct session *s = NULL; struct session *s = NULL;
struct timeval tv;
char *c; char *c;
pa_sink *sink; pa_sink *sink;
int fd = -1; int fd = -1;
pa_memblock *silence; pa_memblock *silence;
pa_sink_input_new_data data; pa_sink_input_new_data data;
struct timespec now;
pa_assert(u);
pa_assert(sdp_info);
if (u->n_sessions >= MAX_SESSIONS) { if (u->n_sessions >= MAX_SESSIONS) {
pa_log("session limit reached."); pa_log("Session limit reached.");
goto fail; goto fail;
} }
if (!(sink = pa_namereg_get(u->core, u->sink_name, PA_NAMEREG_SINK, 1))) { if (!(sink = pa_namereg_get(u->module->core, u->sink_name, PA_NAMEREG_SINK, 1))) {
pa_log("sink does not exist."); pa_log("Sink does not exist.");
goto fail; goto fail;
} }
s = pa_xnew0(struct session, 1); s = pa_xnew0(struct session, 1);
s->userdata = u; s->userdata = u;
s->first_packet = 0; s->first_packet = FALSE;
s->sdp_info = *sdp_info; s->sdp_info = *sdp_info;
s->rtpoll_item = NULL;
pa_rtclock_get(&now);
pa_atomic_store(&s->timestamp, now.tv_sec);
if ((fd = mcast_socket((const struct sockaddr*) &sdp_info->sa, sdp_info->salen)) < 0) if ((fd = mcast_socket((const struct sockaddr*) &sdp_info->sa, sdp_info->salen)) < 0)
goto fail; goto fail;
@ -299,25 +343,27 @@ static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_in
data.module = u->module; data.module = u->module;
pa_sink_input_new_data_set_sample_spec(&data, &sdp_info->sample_spec); pa_sink_input_new_data_set_sample_spec(&data, &sdp_info->sample_spec);
s->sink_input = pa_sink_input_new(u->core, &data, 0); s->sink_input = pa_sink_input_new(u->module->core, &data, 0);
pa_xfree(c); pa_xfree(c);
if (!s->sink_input) { if (!s->sink_input) {
pa_log("failed to create sink input."); pa_log("Failed to create sink input.");
goto fail; goto fail;
} }
s->sink_input->userdata = s; s->sink_input->userdata = s;
s->sink_input->parent.process_msg = sink_input_process_msg;
s->sink_input->peek = sink_input_peek; s->sink_input->peek = sink_input_peek;
s->sink_input->drop = sink_input_drop; s->sink_input->drop = sink_input_drop;
s->sink_input->kill = sink_input_kill; s->sink_input->kill = sink_input_kill;
s->sink_input->get_latency = sink_input_get_latency; s->sink_input->attach = sink_input_attach;
s->sink_input->detach = sink_input_detach;
silence = pa_silence_memblock_new(s->userdata->core->mempool, silence = pa_silence_memblock_new(
&s->sink_input->sample_spec, s->userdata->module->core->mempool,
(pa_bytes_per_second(&s->sink_input->sample_spec)/128/pa_frame_size(&s->sink_input->sample_spec))* &s->sink_input->sample_spec,
pa_frame_size(&s->sink_input->sample_spec)); pa_frame_align(pa_bytes_per_second(&s->sink_input->sample_spec)/128, &s->sink_input->sample_spec));
s->memblockq = pa_memblockq_new( s->memblockq = pa_memblockq_new(
0, 0,
@ -330,53 +376,43 @@ static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_in
pa_memblock_unref(silence); pa_memblock_unref(silence);
s->rtp_event = u->core->mainloop->io_new(u->core->mainloop, fd, PA_IO_EVENT_INPUT, rtp_event_cb, s);
pa_gettimeofday(&tv);
pa_timeval_add(&tv, DEATH_TIMEOUT);
s->death_event = u->core->mainloop->time_new(u->core->mainloop, &tv, death_event_cb, s);
pa_hashmap_put(s->userdata->by_origin, s->sdp_info.origin, s);
pa_rtp_context_init_recv(&s->rtp_context, fd, pa_frame_size(&s->sdp_info.sample_spec)); pa_rtp_context_init_recv(&s->rtp_context, fd, pa_frame_size(&s->sdp_info.sample_spec));
pa_log_info("Found new session '%s'", s->sdp_info.session_name); pa_hashmap_put(s->userdata->by_origin, s->sdp_info.origin, s);
u->n_sessions++; u->n_sessions++;
PA_LLIST_PREPEND(struct session, s->userdata->sessions, s);
pa_sink_input_put(s->sink_input);
pa_log_info("New session '%s'", s->sdp_info.session_name);
return s; return s;
fail: fail:
if (s) { pa_xfree(s);
if (fd >= 0)
close(fd);
pa_xfree(s); if (fd >= 0)
} pa_close(fd);
return NULL; return NULL;
} }
static void session_free(struct session *s, int from_hash) { static void session_free(struct session *s) {
assert(s); pa_assert(s);
pa_log_info("Freeing session '%s'", s->sdp_info.session_name); pa_log_info("Freeing session '%s'", s->sdp_info.session_name);
s->userdata->core->mainloop->time_free(s->death_event); pa_sink_input_unlink(s->sink_input);
s->userdata->core->mainloop->io_free(s->rtp_event);
if (from_hash)
pa_hashmap_remove(s->userdata->by_origin, s->sdp_info.origin);
pa_sink_input_disconnect(s->sink_input);
pa_sink_input_unref(s->sink_input); pa_sink_input_unref(s->sink_input);
pa_memblockq_free(s->memblockq); pa_memblockq_free(s->memblockq);
pa_sdp_info_destroy(&s->sdp_info); pa_sdp_info_destroy(&s->sdp_info);
pa_rtp_context_destroy(&s->rtp_context); pa_rtp_context_destroy(&s->rtp_context);
assert(s->userdata->n_sessions >= 1); PA_LLIST_REMOVE(struct session, s->userdata->sessions, s);
pa_assert(s->userdata->n_sessions >= 1);
s->userdata->n_sessions--; s->userdata->n_sessions--;
pa_hashmap_remove(s->userdata->by_origin, s->sdp_info.origin);
pa_xfree(s); pa_xfree(s);
} }
@ -387,11 +423,11 @@ static void sap_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event
pa_sdp_info info; pa_sdp_info info;
struct session *s; struct session *s;
assert(m); pa_assert(m);
assert(e); pa_assert(e);
assert(u); pa_assert(u);
assert(fd == u->sap_context.fd); pa_assert(fd == u->sap_context.fd);
assert(flags == PA_IO_EVENT_INPUT); pa_assert(flags == PA_IO_EVENT_INPUT);
if (pa_sap_recv(&u->sap_context, &goodbye) < 0) if (pa_sap_recv(&u->sap_context, &goodbye) < 0)
return; return;
@ -402,7 +438,7 @@ static void sap_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event
if (goodbye) { if (goodbye) {
if ((s = pa_hashmap_get(u->by_origin, info.origin))) if ((s = pa_hashmap_get(u->by_origin, info.origin)))
session_free(s, 1); session_free(s);
pa_sdp_info_destroy(&info); pa_sdp_info_destroy(&info);
} else { } else {
@ -412,18 +448,47 @@ static void sap_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event
pa_sdp_info_destroy(&info); pa_sdp_info_destroy(&info);
} else { } else {
struct timeval tv; struct timespec now;
pa_rtclock_get(&now);
pa_gettimeofday(&tv); pa_atomic_store(&s->timestamp, now.tv_sec);
pa_timeval_add(&tv, DEATH_TIMEOUT);
m->time_restart(s->death_event, &tv);
pa_sdp_info_destroy(&info); pa_sdp_info_destroy(&info);
} }
} }
} }
int pa__init(pa_core *c, pa_module*m) { static void check_death_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *ptv, void *userdata) {
struct session *s, *n;
struct userdata *u = userdata;
struct timespec now;
struct timeval tv;
pa_assert(m);
pa_assert(t);
pa_assert(ptv);
pa_assert(s);
pa_rtclock_get(&now);
pa_log_debug("Checking for dead streams ...");
for (s = u->sessions; s; s = n) {
int k;
n = s->next;
k = pa_atomic_load(&s->timestamp);
if (k + DEATH_TIMEOUT < now.tv_sec)
session_free(s);
}
/* Restart timer */
pa_gettimeofday(&tv);
pa_timeval_add(&tv, DEATH_TIMEOUT*PA_USEC_PER_SEC);
m->time_restart(t, &tv);
}
int pa__init(pa_module*m) {
struct userdata *u; struct userdata *u;
pa_modargs *ma = NULL; pa_modargs *ma = NULL;
struct sockaddr_in sa4; struct sockaddr_in sa4;
@ -432,9 +497,9 @@ int pa__init(pa_core *c, pa_module*m) {
socklen_t salen; socklen_t salen;
const char *sap_address; const char *sap_address;
int fd = -1; int fd = -1;
struct timeval tv;
assert(c); pa_assert(m);
assert(m);
if (!(ma = pa_modargs_new(m->argument, valid_modargs))) { if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
pa_log("failed to parse module arguments"); pa_log("failed to parse module arguments");
@ -454,7 +519,7 @@ int pa__init(pa_core *c, pa_module*m) {
sa = (struct sockaddr*) &sa4; sa = (struct sockaddr*) &sa4;
salen = sizeof(sa4); salen = sizeof(sa4);
} else { } else {
pa_log("invalid SAP address '%s'", sap_address); pa_log("Invalid SAP address '%s'", sap_address);
goto fail; goto fail;
} }
@ -464,16 +529,19 @@ int pa__init(pa_core *c, pa_module*m) {
u = pa_xnew(struct userdata, 1); u = pa_xnew(struct userdata, 1);
m->userdata = u; m->userdata = u;
u->module = m; u->module = m;
u->core = c;
u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL)); u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
u->n_sessions = 0; u->n_sessions = 0;
u->sap_event = c->mainloop->io_new(c->mainloop, fd, PA_IO_EVENT_INPUT, sap_event_cb, u); u->sap_event = m->core->mainloop->io_new(m->core->mainloop, fd, PA_IO_EVENT_INPUT, sap_event_cb, u);
u->by_origin = pa_hashmap_new(pa_idxset_string_hash_func, pa_idxset_string_compare_func); u->by_origin = pa_hashmap_new(pa_idxset_string_hash_func, pa_idxset_string_compare_func);
pa_sap_context_init_recv(&u->sap_context, fd); pa_sap_context_init_recv(&u->sap_context, fd);
pa_gettimeofday(&tv);
pa_timeval_add(&tv, DEATH_TIMEOUT * PA_USEC_PER_SEC);
u->check_death_event = m->core->mainloop->time_new(m->core->mainloop, &tv, check_death_event_cb, u);
pa_modargs_free(ma); pa_modargs_free(ma);
return 0; return 0;
@ -483,27 +551,34 @@ fail:
pa_modargs_free(ma); pa_modargs_free(ma);
if (fd >= 0) if (fd >= 0)
close(fd); pa_close(fd);
return -1; return -1;
} }
static void free_func(void *p, PA_GCC_UNUSED void *userdata) { void pa__done(pa_module*m) {
session_free(p, 0);
}
void pa__done(pa_core *c, pa_module*m) {
struct userdata *u; struct userdata *u;
assert(c); struct session *s;
assert(m);
pa_assert(m);
if (!(u = m->userdata)) if (!(u = m->userdata))
return; return;
c->mainloop->io_free(u->sap_event); if (u->sap_event)
m->core->mainloop->io_free(u->sap_event);
if (u->check_death_event)
m->core->mainloop->time_free(u->check_death_event);
pa_sap_context_destroy(&u->sap_context); pa_sap_context_destroy(&u->sap_context);
pa_hashmap_free(u->by_origin, free_func, NULL); if (u->by_origin) {
while ((s = pa_hashmap_get_first(u->by_origin)))
session_free(s);
pa_hashmap_free(u->by_origin, NULL, NULL);
}
pa_xfree(u->sink_name); pa_xfree(u->sink_name);
pa_xfree(u); pa_xfree(u);

118
src/modules/rtp/module-rtp-send.c
View file

@ -25,7 +25,6 @@
#include <config.h> #include <config.h>
#endif #endif
#include <assert.h>
#include <stdio.h> #include <stdio.h>
#include <sys/socket.h> #include <sys/socket.h>
#include <netinet/in.h> #include <netinet/in.h>
@ -48,6 +47,8 @@
#include <pulsecore/core-util.h> #include <pulsecore/core-util.h>
#include <pulsecore/modargs.h> #include <pulsecore/modargs.h>
#include <pulsecore/namereg.h> #include <pulsecore/namereg.h>
#include <pulsecore/sample-util.h>
#include <pulsecore/macro.h>
#include "module-rtp-send-symdef.h" #include "module-rtp-send-symdef.h"
@ -74,7 +75,7 @@ PA_MODULE_USAGE(
#define DEFAULT_DESTINATION "224.0.0.56" #define DEFAULT_DESTINATION "224.0.0.56"
#define MEMBLOCKQ_MAXLENGTH (1024*170) #define MEMBLOCKQ_MAXLENGTH (1024*170)
#define DEFAULT_MTU 1280 #define DEFAULT_MTU 1280
#define SAP_INTERVAL 5000000 #define SAP_INTERVAL 5
static const char* const valid_modargs[] = { static const char* const valid_modargs[] = {
"source", "source",
@ -90,7 +91,6 @@ static const char* const valid_modargs[] = {
struct userdata { struct userdata {
pa_module *module; pa_module *module;
pa_core *core;
pa_source_output *source_output; pa_source_output *source_output;
pa_memblockq *memblockq; pa_memblockq *memblockq;
@ -102,56 +102,67 @@ struct userdata {
pa_time_event *sap_event; pa_time_event *sap_event;
}; };
/* Called from I/O thread context */
static int source_output_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct userdata *u;
pa_assert_se(u = PA_SOURCE_OUTPUT(o)->userdata);
switch (code) {
case PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY:
*((pa_usec_t*) data) = pa_bytes_to_usec(pa_memblockq_get_length(u->memblockq), &u->source_output->sample_spec);
/* Fall through, the default handler will add in the extra
* latency added by the resampler */
break;
}
return pa_source_output_process_msg(o, code, data, offset, chunk);
}
/* Called from I/O thread context */
static void source_output_push(pa_source_output *o, const pa_memchunk *chunk) { static void source_output_push(pa_source_output *o, const pa_memchunk *chunk) {
struct userdata *u; struct userdata *u;
assert(o); pa_source_output_assert_ref(o);
u = o->userdata; pa_assert_se(u = o->userdata);
if (pa_memblockq_push(u->memblockq, chunk) < 0) { if (pa_memblockq_push(u->memblockq, chunk) < 0) {
pa_log("Failed to push chunk into memblockq."); pa_log_warn("Failed to push chunk into memblockq.");
return; return;
} }
pa_rtp_send(&u->rtp_context, u->mtu, u->memblockq); pa_rtp_send(&u->rtp_context, u->mtu, u->memblockq);
} }
/* Called from main context */
static void source_output_kill(pa_source_output* o) { static void source_output_kill(pa_source_output* o) {
struct userdata *u; struct userdata *u;
assert(o); pa_source_output_assert_ref(o);
u = o->userdata; pa_assert_se(u = o->userdata);
pa_module_unload_request(u->module); pa_module_unload_request(u->module);
pa_source_output_disconnect(u->source_output); pa_source_output_unlink(u->source_output);
pa_source_output_unref(u->source_output); pa_source_output_unref(u->source_output);
u->source_output = NULL; u->source_output = NULL;
} }
static pa_usec_t source_output_get_latency (pa_source_output *o) {
struct userdata *u;
assert(o);
u = o->userdata;
return pa_bytes_to_usec(pa_memblockq_get_length(u->memblockq), &o->sample_spec);
}
static void sap_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *tv, void *userdata) { static void sap_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *tv, void *userdata) {
struct userdata *u = userdata; struct userdata *u = userdata;
struct timeval next; struct timeval next;
assert(m); pa_assert(m);
assert(t); pa_assert(t);
assert(tv); pa_assert(tv);
assert(u); pa_assert(u);
pa_sap_send(&u->sap_context, 0); pa_sap_send(&u->sap_context, 0);
pa_gettimeofday(&next); pa_gettimeofday(&next);
pa_timeval_add(&next, SAP_INTERVAL); pa_timeval_add(&next, SAP_INTERVAL * PA_USEC_PER_SEC);
m->time_restart(t, &next); m->time_restart(t, &next);
} }
int pa__init(pa_core *c, pa_module*m) { int pa__init(pa_module*m) {
struct userdata *u; struct userdata *u;
pa_modargs *ma = NULL; pa_modargs *ma = NULL;
const char *dest; const char *dest;
@ -173,21 +184,20 @@ int pa__init(pa_core *c, pa_module*m) {
int loop = 0; int loop = 0;
pa_source_output_new_data data; pa_source_output_new_data data;
assert(c); pa_assert(m);
assert(m);
if (!(ma = pa_modargs_new(m->argument, valid_modargs))) { if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
pa_log("failed to parse module arguments"); pa_log("Failed to parse module arguments");
goto fail; goto fail;
} }
if (!(s = pa_namereg_get(m->core, pa_modargs_get_value(ma, "source", NULL), PA_NAMEREG_SOURCE, 1))) { if (!(s = pa_namereg_get(m->core, pa_modargs_get_value(ma, "source", NULL), PA_NAMEREG_SOURCE, 1))) {
pa_log("source does not exist."); pa_log("Source does not exist.");
goto fail; goto fail;
} }
if (pa_modargs_get_value_boolean(ma, "loop", &loop) < 0) { if (pa_modargs_get_value_boolean(ma, "loop", &loop) < 0) {
pa_log("failed to parse \"loop\" parameter."); pa_log("Failed to parse \"loop\" parameter.");
goto fail; goto fail;
} }
@ -195,12 +205,12 @@ int pa__init(pa_core *c, pa_module*m) {
pa_rtp_sample_spec_fixup(&ss); pa_rtp_sample_spec_fixup(&ss);
cm = s->channel_map; cm = s->channel_map;
if (pa_modargs_get_sample_spec(ma, &ss) < 0) { if (pa_modargs_get_sample_spec(ma, &ss) < 0) {
pa_log("failed to parse sample specification"); pa_log("Failed to parse sample specification");
goto fail; goto fail;
} }
if (!pa_rtp_sample_spec_valid(&ss)) { if (!pa_rtp_sample_spec_valid(&ss)) {
pa_log("specified sample type not compatible with RTP"); pa_log("Specified sample type not compatible with RTP");
goto fail; goto fail;
} }
@ -209,10 +219,10 @@ int pa__init(pa_core *c, pa_module*m) {
payload = pa_rtp_payload_from_sample_spec(&ss); payload = pa_rtp_payload_from_sample_spec(&ss);
mtu = (DEFAULT_MTU/pa_frame_size(&ss))*pa_frame_size(&ss); mtu = pa_frame_align(DEFAULT_MTU, &ss);
if (pa_modargs_get_value_u32(ma, "mtu", &mtu) < 0 || mtu < 1 || mtu % pa_frame_size(&ss) != 0) { if (pa_modargs_get_value_u32(ma, "mtu", &mtu) < 0 || mtu < 1 || mtu % pa_frame_size(&ss) != 0) {
pa_log("invalid mtu."); pa_log("Invalid MTU.");
goto fail; goto fail;
} }
@ -223,7 +233,7 @@ int pa__init(pa_core *c, pa_module*m) {
} }
if (port & 1) if (port & 1)
pa_log_warn("WARNING: port number not even as suggested in RFC3550!"); pa_log_warn("Port number not even as suggested in RFC3550!");
dest = pa_modargs_get_value(ma, "destination", DEFAULT_DESTINATION); dest = pa_modargs_get_value(ma, "destination", DEFAULT_DESTINATION);
@ -238,7 +248,7 @@ int pa__init(pa_core *c, pa_module*m) {
sap_sa4 = sa4; sap_sa4 = sa4;
sap_sa4.sin_port = htons(SAP_PORT); sap_sa4.sin_port = htons(SAP_PORT);
} else { } else {
pa_log("invalid destination '%s'", dest); pa_log("Invalid destination '%s'", dest);
goto fail; goto fail;
} }
@ -268,6 +278,9 @@ int pa__init(pa_core *c, pa_module*m) {
goto fail; goto fail;
} }
/* If the socket queue is full, let's drop packets */
pa_make_nonblock_fd(fd);
pa_source_output_new_data_init(&data); pa_source_output_new_data_init(&data);
data.name = "RTP Monitor Stream"; data.name = "RTP Monitor Stream";
data.driver = __FILE__; data.driver = __FILE__;
@ -276,21 +289,20 @@ int pa__init(pa_core *c, pa_module*m) {
pa_source_output_new_data_set_sample_spec(&data, &ss); pa_source_output_new_data_set_sample_spec(&data, &ss);
pa_source_output_new_data_set_channel_map(&data, &cm); pa_source_output_new_data_set_channel_map(&data, &cm);
if (!(o = pa_source_output_new(c, &data, 0))) { if (!(o = pa_source_output_new(m->core, &data, 0))) {
pa_log("failed to create source output."); pa_log("failed to create source output.");
goto fail; goto fail;
} }
o->parent.process_msg = source_output_process_msg;
o->push = source_output_push; o->push = source_output_push;
o->kill = source_output_kill; o->kill = source_output_kill;
o->get_latency = source_output_get_latency;
u = pa_xnew(struct userdata, 1); u = pa_xnew(struct userdata, 1);
m->userdata = u; m->userdata = u;
o->userdata = u; o->userdata = u;
u->module = m; u->module = m;
u->core = c;
u->source_output = o; u->source_output = o;
u->memblockq = pa_memblockq_new( u->memblockq = pa_memblockq_new(
@ -305,8 +317,7 @@ int pa__init(pa_core *c, pa_module*m) {
u->mtu = mtu; u->mtu = mtu;
k = sizeof(sa_dst); k = sizeof(sa_dst);
r = getsockname(fd, (struct sockaddr*) &sa_dst, &k); pa_assert_se((r = getsockname(fd, (struct sockaddr*) &sa_dst, &k)) >= 0);
assert(r >= 0);
n = pa_sprintf_malloc("PulseAudio RTP Stream on %s", pa_get_fqdn(hn, sizeof(hn))); n = pa_sprintf_malloc("PulseAudio RTP Stream on %s", pa_get_fqdn(hn, sizeof(hn)));
@ -317,17 +328,19 @@ int pa__init(pa_core *c, pa_module*m) {
pa_xfree(n); pa_xfree(n);
pa_rtp_context_init_send(&u->rtp_context, fd, c->cookie, payload, pa_frame_size(&ss)); pa_rtp_context_init_send(&u->rtp_context, fd, m->core->cookie, payload, pa_frame_size(&ss));
pa_sap_context_init_send(&u->sap_context, sap_fd, p); pa_sap_context_init_send(&u->sap_context, sap_fd, p);
pa_log_info("RTP stream initialized with mtu %u on %s:%u, SSRC=0x%08x, payload=%u, initial sequence #%u", mtu, dest, port, u->rtp_context.ssrc, payload, u->rtp_context.sequence); pa_log_info("RTP stream initialized with mtu %u on %s:%u, SSRC=0x%08x, payload=%u, initial sequence #%u", mtu, dest, port, u->rtp_context.ssrc, payload, u->rtp_context.sequence);
pa_log_info("SDP-Data:\n%s\n"__FILE__": EOF", p); pa_log_info("SDP-Data:\n%s\nEOF", p);
pa_sap_send(&u->sap_context, 0); pa_sap_send(&u->sap_context, 0);
pa_gettimeofday(&tv); pa_gettimeofday(&tv);
pa_timeval_add(&tv, SAP_INTERVAL); pa_timeval_add(&tv, SAP_INTERVAL * PA_USEC_PER_SEC);
u->sap_event = c->mainloop->time_new(c->mainloop, &tv, sap_event_cb, u); u->sap_event = m->core->mainloop->time_new(m->core->mainloop, &tv, sap_event_cb, u);
pa_source_output_put(u->source_output);
pa_modargs_free(ma); pa_modargs_free(ma);
@ -338,31 +351,31 @@ fail:
pa_modargs_free(ma); pa_modargs_free(ma);
if (fd >= 0) if (fd >= 0)
close(fd); pa_close(fd);
if (sap_fd >= 0) if (sap_fd >= 0)
close(sap_fd); pa_close(sap_fd);
if (o) { if (o) {
pa_source_output_disconnect(o); pa_source_output_unlink(o);
pa_source_output_unref(o); pa_source_output_unref(o);
} }
return -1; return -1;
} }
void pa__done(pa_core *c, pa_module*m) { void pa__done(pa_module*m) {
struct userdata *u; struct userdata *u;
assert(c); pa_assert(m);
assert(m);
if (!(u = m->userdata)) if (!(u = m->userdata))
return; return;
c->mainloop->time_free(u->sap_event); if (u->sap_event)
m->core->mainloop->time_free(u->sap_event);
if (u->source_output) { if (u->source_output) {
pa_source_output_disconnect(u->source_output); pa_source_output_unlink(u->source_output);
pa_source_output_unref(u->source_output); pa_source_output_unref(u->source_output);
} }
@ -371,7 +384,8 @@ void pa__done(pa_core *c, pa_module*m) {
pa_sap_send(&u->sap_context, 1); pa_sap_send(&u->sap_context, 1);
pa_sap_context_destroy(&u->sap_context); pa_sap_context_destroy(&u->sap_context);
pa_memblockq_free(u->memblockq); if (u->memblockq)
pa_memblockq_free(u->memblockq);
pa_xfree(u); pa_xfree(u);
} }

80
src/modules/rtp/rtp.c
View file

@ -25,7 +25,6 @@
#include <config.h> #include <config.h>
#endif #endif
#include <assert.h>
#include <fcntl.h> #include <fcntl.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
@ -40,12 +39,14 @@
#include <pulsecore/core-error.h> #include <pulsecore/core-error.h>
#include <pulsecore/log.h> #include <pulsecore/log.h>
#include <pulsecore/macro.h>
#include <pulsecore/core-util.h>
#include "rtp.h" #include "rtp.h"
pa_rtp_context* pa_rtp_context_init_send(pa_rtp_context *c, int fd, uint32_t ssrc, uint8_t payload, size_t frame_size) { pa_rtp_context* pa_rtp_context_init_send(pa_rtp_context *c, int fd, uint32_t ssrc, uint8_t payload, size_t frame_size) {
assert(c); pa_assert(c);
assert(fd >= 0); pa_assert(fd >= 0);
c->fd = fd; c->fd = fd;
c->sequence = (uint16_t) (rand()*rand()); c->sequence = (uint16_t) (rand()*rand());
@ -63,11 +64,11 @@ int pa_rtp_send(pa_rtp_context *c, size_t size, pa_memblockq *q) {
struct iovec iov[MAX_IOVECS]; struct iovec iov[MAX_IOVECS];
pa_memblock* mb[MAX_IOVECS]; pa_memblock* mb[MAX_IOVECS];
int iov_idx = 1; int iov_idx = 1;
size_t n = 0, skip = 0; size_t n = 0;
assert(c); pa_assert(c);
assert(size > 0); pa_assert(size > 0);
assert(q); pa_assert(q);
if (pa_memblockq_get_length(q) < size) if (pa_memblockq_get_length(q) < size)
return 0; return 0;
@ -76,24 +77,26 @@ int pa_rtp_send(pa_rtp_context *c, size_t size, pa_memblockq *q) {
int r; int r;
pa_memchunk chunk; pa_memchunk chunk;
pa_memchunk_reset(&chunk);
if ((r = pa_memblockq_peek(q, &chunk)) >= 0) { if ((r = pa_memblockq_peek(q, &chunk)) >= 0) {
size_t k = n + chunk.length > size ? size - n : chunk.length; size_t k = n + chunk.length > size ? size - n : chunk.length;
if (chunk.memblock) { pa_assert(chunk.memblock);
iov[iov_idx].iov_base = (void*)((uint8_t*) pa_memblock_acquire(chunk.memblock) + chunk.index);
iov[iov_idx].iov_len = k;
mb[iov_idx] = chunk.memblock;
iov_idx ++;
n += k; iov[iov_idx].iov_base = ((uint8_t*) pa_memblock_acquire(chunk.memblock) + chunk.index);
} iov[iov_idx].iov_len = k;
mb[iov_idx] = chunk.memblock;
iov_idx ++;
skip += k; n += k;
pa_memblockq_drop(q, k); pa_memblockq_drop(q, k);
} }
if (r < 0 || !chunk.memblock || n >= size || iov_idx >= MAX_IOVECS) { pa_assert(n % c->frame_size == 0);
if (r < 0 || n >= size || iov_idx >= MAX_IOVECS) {
uint32_t header[3]; uint32_t header[3];
struct msghdr m; struct msghdr m;
int k, i; int k, i;
@ -125,10 +128,10 @@ int pa_rtp_send(pa_rtp_context *c, size_t size, pa_memblockq *q) {
} else } else
k = 0; k = 0;
c->timestamp += skip/c->frame_size; c->timestamp += n/c->frame_size;
if (k < 0) { if (k < 0) {
if (errno != EAGAIN) /* If the queue is full, just ignore it */ if (errno != EAGAIN && errno != EINTR) /* If the queue is full, just ignore it */
pa_log("sendmsg() failed: %s", pa_cstrerror(errno)); pa_log("sendmsg() failed: %s", pa_cstrerror(errno));
return -1; return -1;
} }
@ -137,7 +140,6 @@ int pa_rtp_send(pa_rtp_context *c, size_t size, pa_memblockq *q) {
break; break;
n = 0; n = 0;
skip = 0;
iov_idx = 1; iov_idx = 1;
} }
} }
@ -146,7 +148,7 @@ int pa_rtp_send(pa_rtp_context *c, size_t size, pa_memblockq *q) {
} }
pa_rtp_context* pa_rtp_context_init_recv(pa_rtp_context *c, int fd, size_t frame_size) { pa_rtp_context* pa_rtp_context_init_recv(pa_rtp_context *c, int fd, size_t frame_size) {
assert(c); pa_assert(c);
c->fd = fd; c->fd = fd;
c->frame_size = frame_size; c->frame_size = frame_size;
@ -161,13 +163,13 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
int cc; int cc;
ssize_t r; ssize_t r;
assert(c); pa_assert(c);
assert(chunk); pa_assert(chunk);
chunk->memblock = NULL; chunk->memblock = NULL;
if (ioctl(c->fd, FIONREAD, &size) < 0) { if (ioctl(c->fd, FIONREAD, &size) < 0) {
pa_log("FIONREAD failed: %s", pa_cstrerror(errno)); pa_log_warn("FIONREAD failed: %s", pa_cstrerror(errno));
goto fail; goto fail;
} }
@ -188,12 +190,14 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
m.msg_flags = 0; m.msg_flags = 0;
if ((r = recvmsg(c->fd, &m, 0)) != size) { if ((r = recvmsg(c->fd, &m, 0)) != size) {
pa_log("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch"); if (r < 0 && errno != EAGAIN && errno != EINTR)
pa_log_warn("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch");
goto fail; goto fail;
} }
if (size < 12) { if (size < 12) {
pa_log("RTP packet too short."); pa_log_warn("RTP packet too short.");
goto fail; goto fail;
} }
@ -206,17 +210,17 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
c->ssrc = ntohl(c->ssrc); c->ssrc = ntohl(c->ssrc);
if ((header >> 30) != 2) { if ((header >> 30) != 2) {
pa_log("Unsupported RTP version."); pa_log_warn("Unsupported RTP version.");
goto fail; goto fail;
} }
if ((header >> 29) & 1) { if ((header >> 29) & 1) {
pa_log("RTP padding not supported."); pa_log_warn("RTP padding not supported.");
goto fail; goto fail;
} }
if ((header >> 28) & 1) { if ((header >> 28) & 1) {
pa_log("RTP header extensions not supported."); pa_log_warn("RTP header extensions not supported.");
goto fail; goto fail;
} }
@ -225,7 +229,7 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
c->sequence = header & 0xFFFF; c->sequence = header & 0xFFFF;
if (12 + cc*4 > size) { if (12 + cc*4 > size) {
pa_log("RTP packet too short. (CSRC)"); pa_log_warn("RTP packet too short. (CSRC)");
goto fail; goto fail;
} }
@ -233,7 +237,7 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
chunk->length = size - chunk->index; chunk->length = size - chunk->index;
if (chunk->length % c->frame_size != 0) { if (chunk->length % c->frame_size != 0) {
pa_log("Vad RTP packet size."); pa_log_warn("Bad RTP packet size.");
goto fail; goto fail;
} }
@ -249,7 +253,7 @@ fail:
} }
uint8_t pa_rtp_payload_from_sample_spec(const pa_sample_spec *ss) { uint8_t pa_rtp_payload_from_sample_spec(const pa_sample_spec *ss) {
assert(ss); pa_assert(ss);
if (ss->format == PA_SAMPLE_ULAW && ss->rate == 8000 && ss->channels == 1) if (ss->format == PA_SAMPLE_ULAW && ss->rate == 8000 && ss->channels == 1)
return 0; return 0;
@ -264,7 +268,7 @@ uint8_t pa_rtp_payload_from_sample_spec(const pa_sample_spec *ss) {
} }
pa_sample_spec *pa_rtp_sample_spec_from_payload(uint8_t payload, pa_sample_spec *ss) { pa_sample_spec *pa_rtp_sample_spec_from_payload(uint8_t payload, pa_sample_spec *ss) {
assert(ss); pa_assert(ss);
switch (payload) { switch (payload) {
case 0: case 0:
@ -299,17 +303,17 @@ pa_sample_spec *pa_rtp_sample_spec_from_payload(uint8_t payload, pa_sample_spec
} }
pa_sample_spec *pa_rtp_sample_spec_fixup(pa_sample_spec * ss) { pa_sample_spec *pa_rtp_sample_spec_fixup(pa_sample_spec * ss) {
assert(ss); pa_assert(ss);
if (!pa_rtp_sample_spec_valid(ss)) if (!pa_rtp_sample_spec_valid(ss))
ss->format = PA_SAMPLE_S16BE; ss->format = PA_SAMPLE_S16BE;
assert(pa_rtp_sample_spec_valid(ss)); pa_assert(pa_rtp_sample_spec_valid(ss));
return ss; return ss;
} }
int pa_rtp_sample_spec_valid(const pa_sample_spec *ss) { int pa_rtp_sample_spec_valid(const pa_sample_spec *ss) {
assert(ss); pa_assert(ss);
if (!pa_sample_spec_valid(ss)) if (!pa_sample_spec_valid(ss))
return 0; return 0;
@ -322,9 +326,9 @@ int pa_rtp_sample_spec_valid(const pa_sample_spec *ss) {
} }
void pa_rtp_context_destroy(pa_rtp_context *c) { void pa_rtp_context_destroy(pa_rtp_context *c) {
assert(c); pa_assert(c);
close(c->fd); pa_close(c->fd);
} }
const char* pa_rtp_format_to_string(pa_sample_format_t f) { const char* pa_rtp_format_to_string(pa_sample_format_t f) {
@ -343,7 +347,7 @@ const char* pa_rtp_format_to_string(pa_sample_format_t f) {
} }
pa_sample_format_t pa_rtp_string_to_format(const char *s) { pa_sample_format_t pa_rtp_string_to_format(const char *s) {
assert(s); pa_assert(s);
if (!(strcmp(s, "L16"))) if (!(strcmp(s, "L16")))
return PA_SAMPLE_S16BE; return PA_SAMPLE_S16BE;

40
src/modules/rtp/sap.c
View file

@ -25,7 +25,6 @@
#include <config.h> #include <config.h>
#endif #endif
#include <assert.h>
#include <time.h> #include <time.h>
#include <stdlib.h> #include <stdlib.h>
#include <sys/types.h> #include <sys/types.h>
@ -46,6 +45,7 @@
#include <pulsecore/core-error.h> #include <pulsecore/core-error.h>
#include <pulsecore/core-util.h> #include <pulsecore/core-util.h>
#include <pulsecore/log.h> #include <pulsecore/log.h>
#include <pulsecore/macro.h>
#include "sap.h" #include "sap.h"
#include "sdp.h" #include "sdp.h"
@ -53,9 +53,9 @@
#define MIME_TYPE "application/sdp" #define MIME_TYPE "application/sdp"
pa_sap_context* pa_sap_context_init_send(pa_sap_context *c, int fd, char *sdp_data) { pa_sap_context* pa_sap_context_init_send(pa_sap_context *c, int fd, char *sdp_data) {
assert(c); pa_assert(c);
assert(fd >= 0); pa_assert(fd >= 0);
assert(sdp_data); pa_assert(sdp_data);
c->fd = fd; c->fd = fd;
c->sdp_data = sdp_data; c->sdp_data = sdp_data;
@ -65,9 +65,9 @@ pa_sap_context* pa_sap_context_init_send(pa_sap_context *c, int fd, char *sdp_da
} }
void pa_sap_context_destroy(pa_sap_context *c) { void pa_sap_context_destroy(pa_sap_context *c) {
assert(c); pa_assert(c);
close(c->fd); pa_close(c->fd);
pa_xfree(c->sdp_data); pa_xfree(c->sdp_data);
} }
@ -85,7 +85,7 @@ int pa_sap_send(pa_sap_context *c, int goodbye) {
return -1; return -1;
} }
assert(sa->sa_family == AF_INET || sa->sa_family == AF_INET6); pa_assert(sa->sa_family == AF_INET || sa->sa_family == AF_INET6);
header = htonl(((uint32_t) 1 << 29) | header = htonl(((uint32_t) 1 << 29) |
(sa->sa_family == AF_INET6 ? (uint32_t) 1 << 28 : 0) | (sa->sa_family == AF_INET6 ? (uint32_t) 1 << 28 : 0) |
@ -113,14 +113,14 @@ int pa_sap_send(pa_sap_context *c, int goodbye) {
m.msg_flags = 0; m.msg_flags = 0;
if ((k = sendmsg(c->fd, &m, MSG_DONTWAIT)) < 0) if ((k = sendmsg(c->fd, &m, MSG_DONTWAIT)) < 0)
pa_log("sendmsg() failed: %s\n", pa_cstrerror(errno)); pa_log_warn("sendmsg() failed: %s\n", pa_cstrerror(errno));
return k; return k;
} }
pa_sap_context* pa_sap_context_init_recv(pa_sap_context *c, int fd) { pa_sap_context* pa_sap_context_init_recv(pa_sap_context *c, int fd) {
assert(c); pa_assert(c);
assert(fd >= 0); pa_assert(fd >= 0);
c->fd = fd; c->fd = fd;
c->sdp_data = NULL; c->sdp_data = NULL;
@ -136,11 +136,11 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) {
int six, ac; int six, ac;
ssize_t r; ssize_t r;
assert(c); pa_assert(c);
assert(goodbye); pa_assert(goodbye);
if (ioctl(c->fd, FIONREAD, &size) < 0) { if (ioctl(c->fd, FIONREAD, &size) < 0) {
pa_log("FIONREAD failed: %s", pa_cstrerror(errno)); pa_log_warn("FIONREAD failed: %s", pa_cstrerror(errno));
goto fail; goto fail;
} }
@ -159,12 +159,12 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) {
m.msg_flags = 0; m.msg_flags = 0;
if ((r = recvmsg(c->fd, &m, 0)) != size) { if ((r = recvmsg(c->fd, &m, 0)) != size) {
pa_log("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch"); pa_log_warn("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch");
goto fail; goto fail;
} }
if (size < 4) { if (size < 4) {
pa_log("SAP packet too short."); pa_log_warn("SAP packet too short.");
goto fail; goto fail;
} }
@ -172,17 +172,17 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) {
header = ntohl(header); header = ntohl(header);
if (header >> 29 != 1) { if (header >> 29 != 1) {
pa_log("Unsupported SAP version."); pa_log_warn("Unsupported SAP version.");
goto fail; goto fail;
} }
if ((header >> 25) & 1) { if ((header >> 25) & 1) {
pa_log("Encrypted SAP not supported."); pa_log_warn("Encrypted SAP not supported.");
goto fail; goto fail;
} }
if ((header >> 24) & 1) { if ((header >> 24) & 1) {
pa_log("Compressed SAP not supported."); pa_log_warn("Compressed SAP not supported.");
goto fail; goto fail;
} }
@ -191,7 +191,7 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) {
k = 4 + (six ? 16 : 4) + ac*4; k = 4 + (six ? 16 : 4) + ac*4;
if (size < k) { if (size < k) {
pa_log("SAP packet too short (AD)."); pa_log_warn("SAP packet too short (AD).");
goto fail; goto fail;
} }
@ -202,7 +202,7 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) {
e += sizeof(MIME_TYPE); e += sizeof(MIME_TYPE);
size -= sizeof(MIME_TYPE); size -= sizeof(MIME_TYPE);
} else if ((unsigned) size < sizeof(PA_SDP_HEADER)-1 || strncmp(e, PA_SDP_HEADER, sizeof(PA_SDP_HEADER)-1)) { } else if ((unsigned) size < sizeof(PA_SDP_HEADER)-1 || strncmp(e, PA_SDP_HEADER, sizeof(PA_SDP_HEADER)-1)) {
pa_log("Invalid SDP header."); pa_log_warn("Invalid SDP header.");
goto fail; goto fail;
} }

35
src/modules/rtp/sdp.c
View file

@ -25,7 +25,6 @@
#include <config.h> #include <config.h>
#endif #endif
#include <assert.h>
#include <time.h> #include <time.h>
#include <stdlib.h> #include <stdlib.h>
#include <sys/types.h> #include <sys/types.h>
@ -38,33 +37,29 @@
#include <pulsecore/core-util.h> #include <pulsecore/core-util.h>
#include <pulsecore/log.h> #include <pulsecore/log.h>
#include <pulsecore/macro.h>
#include "sdp.h" #include "sdp.h"
#include "rtp.h" #include "rtp.h"
char *pa_sdp_build(int af, const void *src, const void *dst, const char *name, uint16_t port, uint8_t payload, const pa_sample_spec *ss) { char *pa_sdp_build(int af, const void *src, const void *dst, const char *name, uint16_t port, uint8_t payload, const pa_sample_spec *ss) {
uint32_t ntp; uint32_t ntp;
char buf_src[64], buf_dst[64]; char buf_src[64], buf_dst[64], un[64];
const char *u, *f, *a; const char *u, *f, *a;
assert(src); pa_assert(src);
assert(dst); pa_assert(dst);
assert(af == AF_INET || af == AF_INET6); pa_assert(af == AF_INET || af == AF_INET6);
f = pa_rtp_format_to_string(ss->format); pa_assert_se(f = pa_rtp_format_to_string(ss->format));
assert(f);
if (!(u = getenv("USER"))) if (!(u = pa_get_user_name(un, sizeof(un))))
if (!(u = getenv("USERNAME"))) u = "-";
u = "-";
ntp = time(NULL) + 2208988800U; ntp = time(NULL) + 2208988800U;
a = inet_ntop(af, src, buf_src, sizeof(buf_src)); pa_assert_se(a = inet_ntop(af, src, buf_src, sizeof(buf_src)));
assert(a); pa_assert_se(a = inet_ntop(af, dst, buf_dst, sizeof(buf_dst)));
a = inet_ntop(af, dst, buf_dst, sizeof(buf_dst));
assert(a);
return pa_sprintf_malloc( return pa_sprintf_malloc(
PA_SDP_HEADER PA_SDP_HEADER
@ -86,8 +81,8 @@ char *pa_sdp_build(int af, const void *src, const void *dst, const char *name, u
static pa_sample_spec *parse_sdp_sample_spec(pa_sample_spec *ss, char *c) { static pa_sample_spec *parse_sdp_sample_spec(pa_sample_spec *ss, char *c) {
unsigned rate, channels; unsigned rate, channels;
assert(ss); pa_assert(ss);
assert(c); pa_assert(c);
if (pa_startswith(c, "L16/")) { if (pa_startswith(c, "L16/")) {
ss->format = PA_SAMPLE_S16BE; ss->format = PA_SAMPLE_S16BE;
@ -123,8 +118,8 @@ pa_sdp_info *pa_sdp_parse(const char *t, pa_sdp_info *i, int is_goodbye) {
uint16_t port = 0; uint16_t port = 0;
int ss_valid = 0; int ss_valid = 0;
assert(t); pa_assert(t);
assert(i); pa_assert(i);
i->origin = i->session_name = NULL; i->origin = i->session_name = NULL;
i->salen = 0; i->salen = 0;
@ -258,7 +253,7 @@ fail:
} }
void pa_sdp_info_destroy(pa_sdp_info *i) { void pa_sdp_info_destroy(pa_sdp_info *i) {
assert(i); pa_assert(i);
pa_xfree(i->origin); pa_xfree(i->origin);
pa_xfree(i->session_name); pa_xfree(i->session_name);

1
src/pulsecore/rtpoll.h
View file

@ -103,7 +103,6 @@ void pa_rtpoll_item_set_before_callback(pa_rtpoll_item *i, int (*before_cb)(pa_r
* entered the sleeping poll */ * entered the sleeping poll */
void pa_rtpoll_item_set_after_callback(pa_rtpoll_item *i, void (*after_cb)(pa_rtpoll_item *i)); void pa_rtpoll_item_set_after_callback(pa_rtpoll_item *i, void (*after_cb)(pa_rtpoll_item *i));
void pa_rtpoll_item_set_userdata(pa_rtpoll_item *i, void *userdata); void pa_rtpoll_item_set_userdata(pa_rtpoll_item *i, void *userdata);
void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i); void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i);