5 files changed, 369 insertions, 257 deletions

diff --git a/src/modules/rtp/module-rtp-recv.c b/src/modules/rtp/module-rtp-recv.c
index 62ef561f..6c018931 100644
--- a/src/modules/rtp/module-rtp-recv.c
+++ b/src/modules/rtp/module-rtp-recv.c
@@ -24,7 +24,6 @@
 #include <config.h>
 #endif
 
-#include <assert.h>
 #include <stdio.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
@@ -32,6 +31,7 @@
 #include <errno.h>
 #include <string.h>
 #include <unistd.h>
+#include <poll.h>
 
 #include <pulse/timeval.h>
 #include <pulse/xmalloc.h>
@@ -47,6 +47,10 @@
 #include <pulsecore/modargs.h>
 #include <pulsecore/namereg.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"
 
@@ -66,7 +70,7 @@ PA_MODULE_USAGE(
 #define DEFAULT_SAP_ADDRESS "224.0.0.56"
 #define MEMBLOCKQ_MAXLENGTH (1024*170)
 #define MAX_SESSIONS 16
-#define DEATH_TIMEOUT 20000000
+#define DEATH_TIMEOUT 20
 
 static const char* const valid_modargs[] = {
     "sink",
@@ -76,102 +80,126 @@ static const char* const valid_modargs[] = {
 
 struct session {
     struct userdata *userdata;
+    PA_LLIST_FIELDS(struct session);
 
     pa_sink_input *sink_input;
     pa_memblockq *memblockq;
 
-    pa_time_event *death_event;
-
-    int first_packet;
+    pa_bool_t first_packet;
     uint32_t ssrc;
     uint32_t offset;
 
     struct pa_sdp_info sdp_info;
 
     pa_rtp_context rtp_context;
-    pa_io_event* rtp_event;
+
+    pa_rtpoll_item *rtpoll_item;
+
+    pa_atomic_t timestamp;
 };
 
 struct userdata {
     pa_module *module;
-    pa_core *core;
 
     pa_sap_context sap_context;
     pa_io_event* sap_event;
 
-    pa_hashmap *by_origin;
+    pa_time_event *check_death_event;
 
     char *sink_name;
 
+    PA_LLIST_HEAD(struct session, sessions);
+    pa_hashmap *by_origin;
     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) {
-    struct session *s;
-    assert(i);
-    s = i->userdata;
+/* 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;
 
-    return pa_memblockq_peek(s->memblockq, chunk);
+    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);
 }
 
-static void sink_input_drop(pa_sink_input *i, const pa_memchunk *chunk, size_t length) {
+/* Called from I/O thread context */
+static int sink_input_peek(pa_sink_input *i, size_t length, pa_memchunk *chunk) {
     struct session *s;
-    assert(i);
-    s = i->userdata;
+    pa_sink_input_assert_ref(i);
+    pa_assert_se(s = i->userdata);
 
-    pa_memblockq_drop(s->memblockq, chunk, length);
+    return pa_memblockq_peek(s->memblockq, chunk);
 }
 
-static void sink_input_kill(pa_sink_input* i) {
+/* Called from I/O thread context */
+static void sink_input_drop(pa_sink_input *i, size_t length) {
     struct session *s;
-    assert(i);
-    s = i->userdata;
+    pa_sink_input_assert_ref(i);
+    pa_assert_se(s = i->userdata);
 
-    session_free(s, 1);
+    pa_memblockq_drop(s->memblockq, length);
 }
 
-static pa_usec_t sink_input_get_latency(pa_sink_input *i) {
+/* Called from main context */
+static void sink_input_kill(pa_sink_input* i) {
     struct session *s;
-    assert(i);
-    s = i->userdata;
+    pa_sink_input_assert_ref(i);
+    pa_assert_se(s = i->userdata);
 
-    return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &i->sample_spec);
+    session_free(s);
 }
 
-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;
+/* Called from I/O thread context */
+static int rtpoll_work_cb(pa_rtpoll_item *i) {
     pa_memchunk chunk;
     int64_t k, j, delta;
-    struct timeval tv;
+    struct timeval now;
+    struct session *s;
+    struct pollfd *p;
 
-    assert(m);
-    assert(e);
-    assert(s);
-    assert(fd == s->rtp_context.fd);
-    assert(flags == PA_IO_EVENT_INPUT);
+    pa_assert_se(s = pa_rtpoll_item_get_userdata(i));
 
-    if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->core->mempool) < 0)
-        return;
+    p = pa_rtpoll_item_get_pollfd(i, NULL);
+
+    if (p->revents & (POLLERR|POLLNVAL|POLLHUP|POLLOUT)) {
+        pa_log("poll() signalled bad revents.");
+        return -1;
+    }
+
+    if ((p->revents & POLLIN) == 0)
+        return 0;
+
+    p->revents = 0;
+
+    if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->module->core->mempool) < 0)
+        return 0;
 
     if (s->sdp_info.payload != s->rtp_context.payload) {
         pa_memblock_unref(chunk.memblock);
-        return;
+        return 0;
     }
 
     if (!s->first_packet) {
-        s->first_packet = 1;
+        s->first_packet = TRUE;
 
         s->ssrc = s->rtp_context.ssrc;
         s->offset = s->rtp_context.timestamp;
 
-        if (s->ssrc == s->userdata->core->cookie)
-            pa_log_warn("WARNING! Detected RTP packet loop!");
+        if (s->ssrc == s->userdata->module->core->cookie)
+            pa_log_warn("Detected RTP packet loop!");
     } else {
         if (s->ssrc != s->rtp_context.ssrc) {
             pa_memblock_unref(chunk.memblock);
-            return;
+            return 0;
         }
     }
 
@@ -197,26 +225,49 @@ static void rtp_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event
 
     pa_memblock_unref(chunk.memblock);
 
-    /* Reset death timer */
-    pa_gettimeofday(&tv);
-    pa_timeval_add(&tv, DEATH_TIMEOUT);
-    m->time_restart(s->death_event, &tv);
+    pa_rtclock_get(&now);
+    pa_atomic_store(&s->timestamp, now.tv_sec);
+
+    return 1;
 }
 
-static void death_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *tv, void *userdata) {
-    struct session *s = userdata;
+/* Called from I/O thread context */
+static void sink_input_attach(pa_sink_input *i) {
+    struct session *s;
+    struct pollfd *p;
+
+    pa_sink_input_assert_ref(i);
+    pa_assert_se(s = i->userdata);
+
+    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);
+}
 
-    assert(m);
-    assert(t);
-    assert(tv);
-    assert(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);
 
-    session_free(s, 1);
+    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) {
     int af, fd = -1, r, one;
 
+    pa_assert(sa);
+    pa_assert(salen > 0);
+
     af = sa->sa_family;
     if ((fd = socket(af, SOCK_DGRAM, 0)) < 0) {
         pa_log("Failed to create socket: %s", pa_cstrerror(errno));
@@ -262,27 +313,34 @@ fail:
 
 static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_info) {
     struct session *s = NULL;
-    struct timeval tv;
     char *c;
     pa_sink *sink;
     int fd = -1;
     pa_memblock *silence;
     pa_sink_input_new_data data;
+    struct timeval now;
+
+    pa_assert(u);
+    pa_assert(sdp_info);
 
     if (u->n_sessions >= MAX_SESSIONS) {
-        pa_log("session limit reached.");
+        pa_log("Session limit reached.");
         goto fail;
     }
 
-    if (!(sink = pa_namereg_get(u->core, u->sink_name, PA_NAMEREG_SINK, 1))) {
-        pa_log("sink does not exist.");
+    if (!(sink = pa_namereg_get(u->module->core, u->sink_name, PA_NAMEREG_SINK, 1))) {
+        pa_log("Sink does not exist.");
         goto fail;
     }
 
     s = pa_xnew0(struct session, 1);
     s->userdata = u;
-    s->first_packet = 0;
+    s->first_packet = FALSE;
     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)
         goto fail;
@@ -299,25 +357,27 @@ static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_in
     data.module = u->module;
     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);
 
     if (!s->sink_input) {
-        pa_log("failed to create sink input.");
+        pa_log("Failed to create sink input.");
         goto fail;
     }
 
     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->drop = sink_input_drop;
     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,
-                                      &s->sink_input->sample_spec,
-                                      (pa_bytes_per_second(&s->sink_input->sample_spec)/128/pa_frame_size(&s->sink_input->sample_spec))*
-                                      pa_frame_size(&s->sink_input->sample_spec));
+    silence = pa_silence_memblock_new(
+            s->userdata->module->core->mempool,
+            &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(
             0,
@@ -330,54 +390,44 @@ static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_in
 
     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_rtp_context_init_recv(&s->rtp_context, fd, pa_frame_size(&s->sdp_info.sample_spec));
 
     pa_hashmap_put(s->userdata->by_origin, s->sdp_info.origin, s);
+    u->n_sessions++;
+    PA_LLIST_PREPEND(struct session, s->userdata->sessions, s);
 
-    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_sink_input_put(s->sink_input);
 
-    u->n_sessions++;
+    pa_log_info("New session '%s'", s->sdp_info.session_name);
 
     return s;
 
 fail:
-    if (s) {
-        if (fd >= 0)
-            close(fd);
+    pa_xfree(s);
 
-        pa_xfree(s);
-    }
+    if (fd >= 0)
+        pa_close(fd);
 
     return NULL;
 }
 
-static void session_free(struct session *s, int from_hash) {
-    assert(s);
+static void session_free(struct session *s) {
+    pa_assert(s);
 
     pa_log_info("Freeing session '%s'", s->sdp_info.session_name);
 
-    s->userdata->core->mainloop->time_free(s->death_event);
-    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_unlink(s->sink_input);
     pa_sink_input_unref(s->sink_input);
 
+    PA_LLIST_REMOVE(struct session, s->userdata->sessions, s);
+    pa_assert(s->userdata->n_sessions >= 1);
+    s->userdata->n_sessions--;
+    pa_hashmap_remove(s->userdata->by_origin, s->sdp_info.origin);
+
     pa_memblockq_free(s->memblockq);
     pa_sdp_info_destroy(&s->sdp_info);
     pa_rtp_context_destroy(&s->rtp_context);
 
-    assert(s->userdata->n_sessions >= 1);
-    s->userdata->n_sessions--;
-
     pa_xfree(s);
 }
 
@@ -387,11 +437,11 @@ static void sap_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event
     pa_sdp_info info;
     struct session *s;
 
-    assert(m);
-    assert(e);
-    assert(u);
-    assert(fd == u->sap_context.fd);
-    assert(flags == PA_IO_EVENT_INPUT);
+    pa_assert(m);
+    pa_assert(e);
+    pa_assert(u);
+    pa_assert(fd == u->sap_context.fd);
+    pa_assert(flags == PA_IO_EVENT_INPUT);
 
     if (pa_sap_recv(&u->sap_context, &goodbye) < 0)
         return;
@@ -402,7 +452,7 @@ static void sap_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event
     if (goodbye) {
 
         if ((s = pa_hashmap_get(u->by_origin, info.origin)))
-            session_free(s, 1);
+            session_free(s);
 
         pa_sdp_info_destroy(&info);
     } else {
@@ -412,18 +462,47 @@ static void sap_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event
                 pa_sdp_info_destroy(&info);
 
         } else {
-            struct timeval tv;
-
-            pa_gettimeofday(&tv);
-            pa_timeval_add(&tv, DEATH_TIMEOUT);
-            m->time_restart(s->death_event, &tv);
+            struct timeval now;
+            pa_rtclock_get(&now);
+            pa_atomic_store(&s->timestamp, now.tv_sec);
 
             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 timeval now;
+    struct timeval tv;
+
+    pa_assert(m);
+    pa_assert(t);
+    pa_assert(ptv);
+    pa_assert(u);
+
+    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;
     pa_modargs *ma = NULL;
     struct sockaddr_in sa4;
@@ -432,9 +511,9 @@ int pa__init(pa_core *c, pa_module*m) {
     socklen_t salen;
     const char *sap_address;
     int fd = -1;
+    struct timeval tv;
 
-    assert(c);
-    assert(m);
+    pa_assert(m);
 
     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
         pa_log("failed to parse module arguments");
@@ -454,7 +533,7 @@ int pa__init(pa_core *c, pa_module*m) {
         sa = (struct sockaddr*) &sa4;
         salen = sizeof(sa4);
     } else {
-        pa_log("invalid SAP address '%s'", sap_address);
+        pa_log("Invalid SAP address '%s'", sap_address);
         goto fail;
     }
 
@@ -464,15 +543,18 @@ int pa__init(pa_core *c, pa_module*m) {
     u = pa_xnew(struct userdata, 1);
     m->userdata = u;
     u->module = m;
-    u->core = c;
     u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
-    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);
+    pa_sap_context_init_recv(&u->sap_context, fd);
 
+    PA_LLIST_HEAD_INIT(struct session, u->sessions);
+    u->n_sessions = 0;
     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_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);
 
@@ -483,27 +565,34 @@ fail:
         pa_modargs_free(ma);
 
     if (fd >= 0)
-        close(fd);
+        pa_close(fd);
 
     return -1;
 }
 
-static void free_func(void *p, PA_GCC_UNUSED void *userdata) {
-    session_free(p, 0);
-}
-
-void pa__done(pa_core *c, pa_module*m) {
+void pa__done(pa_module*m) {
     struct userdata *u;
-    assert(c);
-    assert(m);
+    struct session *s;
+
+    pa_assert(m);
 
     if (!(u = m->userdata))
         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_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);
diff --git a/src/modules/rtp/module-rtp-send.c b/src/modules/rtp/module-rtp-send.c
index 8c9e5f23..f36989bd 100644
--- a/src/modules/rtp/module-rtp-send.c
+++ b/src/modules/rtp/module-rtp-send.c
@@ -25,7 +25,6 @@
 #include <config.h>
 #endif
 
-#include <assert.h>
 #include <stdio.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
@@ -48,6 +47,9 @@
 #include <pulsecore/core-util.h>
 #include <pulsecore/modargs.h>
 #include <pulsecore/namereg.h>
+#include <pulsecore/sample-util.h>
+#include <pulsecore/macro.h>
+#include <pulsecore/socket-util.h>
 
 #include "module-rtp-send-symdef.h"
 
@@ -74,7 +76,7 @@ PA_MODULE_USAGE(
 #define DEFAULT_DESTINATION "224.0.0.56"
 #define MEMBLOCKQ_MAXLENGTH (1024*170)
 #define DEFAULT_MTU 1280
-#define SAP_INTERVAL 5000000
+#define SAP_INTERVAL 5
 
 static const char* const valid_modargs[] = {
     "source",
@@ -90,7 +92,6 @@ static const char* const valid_modargs[] = {
 
 struct userdata {
     pa_module *module;
-    pa_core *core;
 
     pa_source_output *source_output;
     pa_memblockq *memblockq;
@@ -102,56 +103,67 @@ struct userdata {
     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) {
     struct userdata *u;
-    assert(o);
-    u = o->userdata;
+    pa_source_output_assert_ref(o);
+    pa_assert_se(u = o->userdata);
 
     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;
     }
 
     pa_rtp_send(&u->rtp_context, u->mtu, u->memblockq);
 }
 
+/* Called from main context */
 static void source_output_kill(pa_source_output* o) {
     struct userdata *u;
-    assert(o);
-    u = o->userdata;
+    pa_source_output_assert_ref(o);
+    pa_assert_se(u = o->userdata);
 
     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);
     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) {
     struct userdata *u = userdata;
     struct timeval next;
 
-    assert(m);
-    assert(t);
-    assert(tv);
-    assert(u);
+    pa_assert(m);
+    pa_assert(t);
+    pa_assert(tv);
+    pa_assert(u);
 
     pa_sap_send(&u->sap_context, 0);
 
     pa_gettimeofday(&next);
-    pa_timeval_add(&next, SAP_INTERVAL);
+    pa_timeval_add(&next, SAP_INTERVAL * PA_USEC_PER_SEC);
     m->time_restart(t, &next);
 }
 
-int pa__init(pa_core *c, pa_module*m) {
+int pa__init(pa_module*m) {
     struct userdata *u;
     pa_modargs *ma = NULL;
     const char *dest;
@@ -173,21 +185,20 @@ int pa__init(pa_core *c, pa_module*m) {
     int loop = 0;
     pa_source_output_new_data data;
 
-    assert(c);
-    assert(m);
+    pa_assert(m);
 
     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;
     }
 
     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;
     }
 
     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;
     }
 
@@ -195,12 +206,12 @@ int pa__init(pa_core *c, pa_module*m) {
     pa_rtp_sample_spec_fixup(&ss);
     cm = s->channel_map;
     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;
     }
 
     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;
     }
 
@@ -209,10 +220,10 @@ int pa__init(pa_core *c, pa_module*m) {
 
     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) {
-        pa_log("invalid mtu.");
+        pa_log("Invalid MTU.");
         goto fail;
     }
 
@@ -223,7 +234,7 @@ int pa__init(pa_core *c, pa_module*m) {
     }
 
     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);
 
@@ -238,7 +249,7 @@ int pa__init(pa_core *c, pa_module*m) {
         sap_sa4 = sa4;
         sap_sa4.sin_port = htons(SAP_PORT);
     } else {
-        pa_log("invalid destination '%s'", dest);
+        pa_log("Invalid destination '%s'", dest);
         goto fail;
     }
 
@@ -268,6 +279,12 @@ int pa__init(pa_core *c, pa_module*m) {
         goto fail;
     }
 
+    /* If the socket queue is full, let's drop packets */
+    pa_make_fd_nonblock(fd);
+    pa_make_udp_socket_low_delay(fd);
+    pa_make_fd_cloexec(fd);
+    pa_make_fd_cloexec(sap_fd);
+
     pa_source_output_new_data_init(&data);
     data.name = "RTP Monitor Stream";
     data.driver = __FILE__;
@@ -276,21 +293,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_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.");
         goto fail;
     }
 
+    o->parent.process_msg = source_output_process_msg;
     o->push = source_output_push;
     o->kill = source_output_kill;
-    o->get_latency = source_output_get_latency;
 
     u = pa_xnew(struct userdata, 1);
     m->userdata = u;
     o->userdata = u;
 
     u->module = m;
-    u->core = c;
     u->source_output = o;
 
     u->memblockq = pa_memblockq_new(
@@ -305,8 +321,7 @@ int pa__init(pa_core *c, pa_module*m) {
     u->mtu = mtu;
 
     k = sizeof(sa_dst);
-    r = getsockname(fd, (struct sockaddr*) &sa_dst, &k);
-    assert(r >= 0);
+    pa_assert_se((r = getsockname(fd, (struct sockaddr*) &sa_dst, &k)) >= 0);
 
     n = pa_sprintf_malloc("PulseAudio RTP Stream on %s", pa_get_fqdn(hn, sizeof(hn)));
 
@@ -317,17 +332,19 @@ int pa__init(pa_core *c, pa_module*m) {
 
     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_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_gettimeofday(&tv);
-    pa_timeval_add(&tv, SAP_INTERVAL);
-    u->sap_event = c->mainloop->time_new(c->mainloop, &tv, sap_event_cb, u);
+    pa_timeval_add(&tv, SAP_INTERVAL * PA_USEC_PER_SEC);
+    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);
 
@@ -338,31 +355,31 @@ fail:
         pa_modargs_free(ma);
 
     if (fd >= 0)
-        close(fd);
+        pa_close(fd);
 
     if (sap_fd >= 0)
-        close(sap_fd);
+        pa_close(sap_fd);
 
     if (o) {
-        pa_source_output_disconnect(o);
+        pa_source_output_unlink(o);
         pa_source_output_unref(o);
     }
 
     return -1;
 }
 
-void pa__done(pa_core *c, pa_module*m) {
+void pa__done(pa_module*m) {
     struct userdata *u;
-    assert(c);
-    assert(m);
+    pa_assert(m);
 
     if (!(u = m->userdata))
         return;
 
-    c->mainloop->time_free(u->sap_event);
+    if (u->sap_event)
+        m->core->mainloop->time_free(u->sap_event);
 
     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);
     }
 
@@ -371,7 +388,8 @@ void pa__done(pa_core *c, pa_module*m) {
     pa_sap_send(&u->sap_context, 1);
     pa_sap_context_destroy(&u->sap_context);
 
-    pa_memblockq_free(u->memblockq);
+    if (u->memblockq)
+        pa_memblockq_free(u->memblockq);
 
     pa_xfree(u);
 }
diff --git a/src/modules/rtp/rtp.c b/src/modules/rtp/rtp.c
index 03a01412..997fcc34 100644
--- a/src/modules/rtp/rtp.c
+++ b/src/modules/rtp/rtp.c
@@ -25,7 +25,6 @@
 #include <config.h>
 #endif
 
-#include <assert.h>
 #include <fcntl.h>
 #include <stdlib.h>
 #include <string.h>
@@ -40,12 +39,14 @@
 
 #include <pulsecore/core-error.h>
 #include <pulsecore/log.h>
+#include <pulsecore/macro.h>
+#include <pulsecore/core-util.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) {
-    assert(c);
-    assert(fd >= 0);
+    pa_assert(c);
+    pa_assert(fd >= 0);
 
     c->fd = fd;
     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];
     pa_memblock* mb[MAX_IOVECS];
     int iov_idx = 1;
-    size_t n = 0, skip = 0;
+    size_t n = 0;
 
-    assert(c);
-    assert(size > 0);
-    assert(q);
+    pa_assert(c);
+    pa_assert(size > 0);
+    pa_assert(q);
 
     if (pa_memblockq_get_length(q) < size)
         return 0;
@@ -76,24 +77,26 @@ int pa_rtp_send(pa_rtp_context *c, size_t size, pa_memblockq *q) {
         int r;
         pa_memchunk chunk;
 
+        pa_memchunk_reset(&chunk);
+
         if ((r = pa_memblockq_peek(q, &chunk)) >= 0) {
 
             size_t k = n + chunk.length > size ? size - n : chunk.length;
 
-            if (chunk.memblock) {
-                iov[iov_idx].iov_base = (void*)((uint8_t*) chunk.memblock->data + chunk.index);
-                iov[iov_idx].iov_len = k;
-                mb[iov_idx] = chunk.memblock;
-                iov_idx ++;
+            pa_assert(chunk.memblock);
 
-                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;
-            pa_memblockq_drop(q, &chunk, k);
+            n += 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];
             struct msghdr m;
             int k, i;
@@ -116,17 +119,19 @@ int pa_rtp_send(pa_rtp_context *c, size_t size, pa_memblockq *q) {
 
                 k = sendmsg(c->fd, &m, MSG_DONTWAIT);
 
-                for (i = 1; i < iov_idx; i++)
+                for (i = 1; i < iov_idx; i++) {
+                    pa_memblock_release(mb[i]);
                     pa_memblock_unref(mb[i]);
+                }
 
                 c->sequence++;
             } else
                 k = 0;
 
-            c->timestamp += skip/c->frame_size;
+            c->timestamp += n/c->frame_size;
 
             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));
                 return -1;
             }
@@ -135,7 +140,6 @@ int pa_rtp_send(pa_rtp_context *c, size_t size, pa_memblockq *q) {
                 break;
 
             n = 0;
-            skip = 0;
             iov_idx = 1;
         }
     }
@@ -144,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) {
-    assert(c);
+    pa_assert(c);
 
     c->fd = fd;
     c->frame_size = frame_size;
@@ -159,13 +163,13 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
     int cc;
     ssize_t r;
 
-    assert(c);
-    assert(chunk);
+    pa_assert(c);
+    pa_assert(chunk);
 
-    chunk->memblock = NULL;
+    pa_memchunk_reset(chunk);
 
     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;
     }
 
@@ -174,7 +178,7 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
 
     chunk->memblock = pa_memblock_new(pool, size);
 
-    iov.iov_base = chunk->memblock->data;
+    iov.iov_base = pa_memblock_acquire(chunk->memblock);
     iov.iov_len = size;
 
     m.msg_name = NULL;
@@ -185,36 +189,41 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
     m.msg_controllen = 0;
     m.msg_flags = 0;
 
-    if ((r = recvmsg(c->fd, &m, 0)) != size) {
-        pa_log("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch");
+    r = recvmsg(c->fd, &m, 0);
+    pa_memblock_release(chunk->memblock);
+
+    if (r != size) {
+        if (r < 0 && errno != EAGAIN && errno != EINTR)
+            pa_log_warn("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch");
+
         goto fail;
     }
 
     if (size < 12) {
-        pa_log("RTP packet too short.");
+        pa_log_warn("RTP packet too short.");
         goto fail;
     }
 
-    memcpy(&header, chunk->memblock->data, sizeof(uint32_t));
-    memcpy(&c->timestamp, (uint8_t*) chunk->memblock->data + 4, sizeof(uint32_t));
-    memcpy(&c->ssrc, (uint8_t*) chunk->memblock->data + 8, sizeof(uint32_t));
+    memcpy(&header, iov.iov_base, sizeof(uint32_t));
+    memcpy(&c->timestamp, (uint8_t*) iov.iov_base + 4, sizeof(uint32_t));
+    memcpy(&c->ssrc, (uint8_t*) iov.iov_base + 8, sizeof(uint32_t));
 
     header = ntohl(header);
     c->timestamp = ntohl(c->timestamp);
     c->ssrc = ntohl(c->ssrc);
 
     if ((header >> 30) != 2) {
-        pa_log("Unsupported RTP version.");
+        pa_log_warn("Unsupported RTP version.");
         goto fail;
     }
 
     if ((header >> 29) & 1) {
-        pa_log("RTP padding not supported.");
+        pa_log_warn("RTP padding not supported.");
         goto fail;
     }
 
     if ((header >> 28) & 1) {
-        pa_log("RTP header extensions not supported.");
+        pa_log_warn("RTP header extensions not supported.");
         goto fail;
     }
 
@@ -223,7 +232,7 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
     c->sequence = header & 0xFFFF;
 
     if (12 + cc*4 > size) {
-        pa_log("RTP packet too short. (CSRC)");
+        pa_log_warn("RTP packet too short. (CSRC)");
         goto fail;
     }
 
@@ -231,7 +240,7 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
     chunk->length = size - chunk->index;
 
     if (chunk->length % c->frame_size != 0) {
-        pa_log("Vad RTP packet size.");
+        pa_log_warn("Bad RTP packet size.");
         goto fail;
     }
 
@@ -245,7 +254,7 @@ fail:
 }
 
 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)
         return 0;
@@ -260,7 +269,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) {
-    assert(ss);
+    pa_assert(ss);
 
     switch (payload) {
         case 0:
@@ -295,17 +304,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) {
-    assert(ss);
+    pa_assert(ss);
 
     if (!pa_rtp_sample_spec_valid(ss))
         ss->format = PA_SAMPLE_S16BE;
 
-    assert(pa_rtp_sample_spec_valid(ss));
+    pa_assert(pa_rtp_sample_spec_valid(ss));
     return ss;
 }
 
 int pa_rtp_sample_spec_valid(const pa_sample_spec *ss) {
-    assert(ss);
+    pa_assert(ss);
 
     if (!pa_sample_spec_valid(ss))
         return 0;
@@ -318,9 +327,9 @@ int pa_rtp_sample_spec_valid(const pa_sample_spec *ss) {
 }
 
 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) {
@@ -339,7 +348,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) {
-    assert(s);
+    pa_assert(s);
 
     if (!(strcmp(s, "L16")))
         return PA_SAMPLE_S16BE;
diff --git a/src/modules/rtp/sap.c b/src/modules/rtp/sap.c
index 82421807..ed7eb0be 100644
--- a/src/modules/rtp/sap.c
+++ b/src/modules/rtp/sap.c
@@ -25,7 +25,6 @@
 #include <config.h>
 #endif
 
-#include <assert.h>
 #include <time.h>
 #include <stdlib.h>
 #include <sys/types.h>
@@ -46,6 +45,7 @@
 #include <pulsecore/core-error.h>
 #include <pulsecore/core-util.h>
 #include <pulsecore/log.h>
+#include <pulsecore/macro.h>
 
 #include "sap.h"
 #include "sdp.h"
@@ -53,9 +53,9 @@
 #define MIME_TYPE "application/sdp"
 
 pa_sap_context* pa_sap_context_init_send(pa_sap_context *c, int fd, char *sdp_data) {
-    assert(c);
-    assert(fd >= 0);
-    assert(sdp_data);
+    pa_assert(c);
+    pa_assert(fd >= 0);
+    pa_assert(sdp_data);
 
     c->fd = fd;
     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) {
-    assert(c);
+    pa_assert(c);
 
-    close(c->fd);
+    pa_close(c->fd);
     pa_xfree(c->sdp_data);
 }
 
@@ -85,7 +85,7 @@ int pa_sap_send(pa_sap_context *c, int goodbye) {
         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) |
                    (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;
 
     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;
 }
 
 pa_sap_context* pa_sap_context_init_recv(pa_sap_context *c, int fd) {
-    assert(c);
-    assert(fd >= 0);
+    pa_assert(c);
+    pa_assert(fd >= 0);
 
     c->fd = fd;
     c->sdp_data = NULL;
@@ -136,11 +136,11 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) {
     int six, ac;
     ssize_t r;
 
-    assert(c);
-    assert(goodbye);
+    pa_assert(c);
+    pa_assert(goodbye);
 
     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;
     }
 
@@ -159,12 +159,12 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) {
     m.msg_flags = 0;
 
     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;
     }
 
     if (size < 4) {
-        pa_log("SAP packet too short.");
+        pa_log_warn("SAP packet too short.");
         goto fail;
     }
 
@@ -172,17 +172,17 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) {
     header = ntohl(header);
 
     if (header >> 29 != 1) {
-        pa_log("Unsupported SAP version.");
+        pa_log_warn("Unsupported SAP version.");
         goto fail;
     }
 
     if ((header >> 25) & 1) {
-        pa_log("Encrypted SAP not supported.");
+        pa_log_warn("Encrypted SAP not supported.");
         goto fail;
     }
 
     if ((header >> 24) & 1) {
-        pa_log("Compressed SAP not supported.");
+        pa_log_warn("Compressed SAP not supported.");
         goto fail;
     }
 
@@ -191,7 +191,7 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) {
 
     k = 4 + (six ? 16 : 4) + ac*4;
     if (size < k) {
-        pa_log("SAP packet too short (AD).");
+        pa_log_warn("SAP packet too short (AD).");
         goto fail;
     }
 
@@ -202,7 +202,7 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) {
         e += 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)) {
-        pa_log("Invalid SDP header.");
+        pa_log_warn("Invalid SDP header.");
         goto fail;
     }
 
diff --git a/src/modules/rtp/sdp.c b/src/modules/rtp/sdp.c
index 8b0bd535..50ac157a 100644
--- a/src/modules/rtp/sdp.c
+++ b/src/modules/rtp/sdp.c
@@ -25,7 +25,6 @@
 #include <config.h>
 #endif
 
-#include <assert.h>
 #include <time.h>
 #include <stdlib.h>
 #include <sys/types.h>
@@ -35,36 +34,33 @@
 #include <string.h>
 
 #include <pulse/xmalloc.h>
+#include <pulse/util.h>
 
 #include <pulsecore/core-util.h>
 #include <pulsecore/log.h>
+#include <pulsecore/macro.h>
 
 #include "sdp.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) {
     uint32_t ntp;
-    char buf_src[64], buf_dst[64];
+    char buf_src[64], buf_dst[64], un[64];
     const char *u, *f, *a;
 
-    assert(src);
-    assert(dst);
-    assert(af == AF_INET || af == AF_INET6);
+    pa_assert(src);
+    pa_assert(dst);
+    pa_assert(af == AF_INET || af == AF_INET6);
 
-    f = pa_rtp_format_to_string(ss->format);
-    assert(f);
+    pa_assert_se(f = pa_rtp_format_to_string(ss->format));
 
-    if (!(u = getenv("USER")))
-        if (!(u = getenv("USERNAME")))
-            u = "-";
+    if (!(u = pa_get_user_name(un, sizeof(un))))
+        u = "-";
 
     ntp = time(NULL) + 2208988800U;
 
-    a = inet_ntop(af, src, buf_src, sizeof(buf_src));
-    assert(a);
-    a = inet_ntop(af, dst, buf_dst, sizeof(buf_dst));
-    assert(a);
+    pa_assert_se(a = inet_ntop(af, src, buf_src, sizeof(buf_src)));
+    pa_assert_se(a = inet_ntop(af, dst, buf_dst, sizeof(buf_dst)));
 
     return pa_sprintf_malloc(
             PA_SDP_HEADER
@@ -86,8 +82,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) {
     unsigned rate, channels;
-    assert(ss);
-    assert(c);
+    pa_assert(ss);
+    pa_assert(c);
 
     if (pa_startswith(c, "L16/")) {
         ss->format = PA_SAMPLE_S16BE;
@@ -123,8 +119,8 @@ pa_sdp_info *pa_sdp_parse(const char *t, pa_sdp_info *i, int is_goodbye) {
     uint16_t port = 0;
     int ss_valid = 0;
 
-    assert(t);
-    assert(i);
+    pa_assert(t);
+    pa_assert(i);
 
     i->origin = i->session_name = NULL;
     i->salen = 0;
@@ -258,7 +254,7 @@ fail:
 }
 
 void pa_sdp_info_destroy(pa_sdp_info *i) {
-    assert(i);
+    pa_assert(i);
 
     pa_xfree(i->origin);
     pa_xfree(i->session_name);