#include <errno.h>
#include <string.h>
#include <unistd.h>
+#include <math.h>
#include <pulse/rtclock.h>
#include <pulse/timeval.h>
#include <pulsecore/namereg.h>
#include <pulsecore/sample-util.h>
#include <pulsecore/macro.h>
-#include <pulsecore/atomic.h>
-#include <pulsecore/atomic.h>
#include <pulsecore/socket-util.h>
+#include <pulsecore/atomic.h>
#include <pulsecore/once.h>
#include <pulsecore/poll.h>
#include <pulsecore/arpa-inet.h>
PA_MODULE_AUTHOR("Lennart Poettering");
PA_MODULE_DESCRIPTION("Receive data from a network via RTP/SAP/SDP");
PA_MODULE_VERSION(PACKAGE_VERSION);
-PA_MODULE_LOAD_ONCE(FALSE);
+PA_MODULE_LOAD_ONCE(false);
PA_MODULE_USAGE(
"sink=<name of the sink> "
"sap_address=<multicast address to listen on> "
+ "latency_msec=<latency in ms> "
);
#define SAP_PORT 9875
#define DEFAULT_SAP_ADDRESS "224.0.0.56"
+#define DEFAULT_LATENCY_MSEC 500
#define MEMBLOCKQ_MAXLENGTH (1024*1024*40)
#define MAX_SESSIONS 16
#define DEATH_TIMEOUT 20
#define RATE_UPDATE_INTERVAL (5*PA_USEC_PER_SEC)
-#define LATENCY_USEC (500*PA_USEC_PER_MSEC)
static const char* const valid_modargs[] = {
"sink",
"sap_address",
+ "latency_msec",
NULL
};
pa_sink_input *sink_input;
pa_memblockq *memblockq;
- pa_bool_t first_packet;
+ bool first_packet;
uint32_t ssrc;
uint32_t offset;
PA_LLIST_HEAD(struct session, sessions);
pa_hashmap *by_origin;
int n_sessions;
+
+ pa_usec_t latency;
};
static void session_free(struct session *s);
pa_sink_input_assert_ref(i);
pa_assert_se(s = i->userdata);
- session_free(s);
+ pa_hashmap_remove_and_free(s->userdata->by_origin, s->sdp_info.origin);
}
/* Called from IO context */
-static void sink_input_suspend_within_thread(pa_sink_input* i, pa_bool_t b) {
+static void sink_input_suspend_within_thread(pa_sink_input* i, bool b) {
struct session *s;
pa_sink_input_assert_ref(i);
pa_assert_se(s = i->userdata);
if (b)
pa_memblockq_flush_read(s->memblockq);
else
- s->first_packet = FALSE;
+ s->first_packet = false;
}
/* Called from I/O thread context */
}
if (!s->first_packet) {
- s->first_packet = TRUE;
+ s->first_packet = true;
s->ssrc = s->rtp_context.ssrc;
s->offset = s->rtp_context.timestamp;
else
delta = j;
- pa_memblockq_seek(s->memblockq, delta * (int64_t) s->rtp_context.frame_size, PA_SEEK_RELATIVE, TRUE);
+ pa_memblockq_seek(s->memblockq, delta * (int64_t) s->rtp_context.frame_size, PA_SEEK_RELATIVE, true);
if (now.tv_sec == 0) {
PA_ONCE_BEGIN {
if (pa_memblockq_push(s->memblockq, &chunk) < 0) {
pa_log_warn("Queue overrun");
- pa_memblockq_seek(s->memblockq, (int64_t) chunk.length, PA_SEEK_RELATIVE, TRUE);
+ pa_memblockq_seek(s->memblockq, (int64_t) chunk.length, PA_SEEK_RELATIVE, true);
}
/* pa_log("blocks in q: %u", pa_memblockq_get_nblocks(s->memblockq)); */
pa_log_debug("Requesting rewind due to end of underrun");
pa_sink_input_request_rewind(s->sink_input,
(size_t) (s->sink_input->thread_info.underrun_for == (uint64_t) -1 ? 0 : s->sink_input->thread_info.underrun_for),
- FALSE, TRUE, FALSE);
+ false, true, false);
}
return 1;
pa_make_udp_socket_low_delay(fd);
+#ifdef SO_TIMESTAMP
one = 1;
if (setsockopt(fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one)) < 0) {
pa_log("SO_TIMESTAMP failed: %s", pa_cstrerror(errno));
goto fail;
}
+#else
+ pa_log("SO_TIMESTAMP unsupported on this platform");
+ goto fail;
+#endif
one = 1;
if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0) {
goto fail;
}
+ r = 0;
if (af == AF_INET) {
- struct ip_mreq mr4;
- memset(&mr4, 0, sizeof(mr4));
- mr4.imr_multiaddr = ((const struct sockaddr_in*) sa)->sin_addr;
- r = setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mr4, sizeof(mr4));
+ /* IPv4 multicast addresses are in the 224.0.0.0-239.255.255.255 range */
+ static const uint32_t ipv4_mcast_mask = 0xe0000000;
+
+ if ((ntohl(((const struct sockaddr_in*) sa)->sin_addr.s_addr) & ipv4_mcast_mask) == ipv4_mcast_mask) {
+ struct ip_mreq mr4;
+ memset(&mr4, 0, sizeof(mr4));
+ mr4.imr_multiaddr = ((const struct sockaddr_in*) sa)->sin_addr;
+ r = setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mr4, sizeof(mr4));
+ }
#ifdef HAVE_IPV6
- } else {
- struct ipv6_mreq mr6;
- memset(&mr6, 0, sizeof(mr6));
- mr6.ipv6mr_multiaddr = ((const struct sockaddr_in6*) sa)->sin6_addr;
- r = setsockopt(fd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mr6, sizeof(mr6));
+ } else if (af == AF_INET6) {
+ /* IPv6 multicast addresses have 255 as the most significant byte */
+ if (((const struct sockaddr_in6*) sa)->sin6_addr.s6_addr[0] == 0xff) {
+ struct ipv6_mreq mr6;
+ memset(&mr6, 0, sizeof(mr6));
+ mr6.ipv6mr_multiaddr = ((const struct sockaddr_in6*) sa)->sin6_addr;
+ r = setsockopt(fd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mr6, sizeof(mr6));
+ }
#endif
- }
+ } else
+ pa_assert_not_reached();
if (r < 0) {
pa_log_info("Joining mcast group failed: %s", pa_cstrerror(errno));
s = pa_xnew0(struct session, 1);
s->userdata = u;
- s->first_packet = FALSE;
+ s->first_packet = false;
s->sdp_info = *sdp_info;
s->rtpoll_item = NULL;
- s->intended_latency = LATENCY_USEC;
+ s->intended_latency = u->latency;
s->last_rate_update = pa_timeval_load(&now);
- s->last_latency = LATENCY_USEC;
+ s->last_latency = u->latency;
s->estimated_rate = (double) sink->sample_spec.rate;
s->avg_estimated_rate = (double) sink->sample_spec.rate;
pa_atomic_store(&s->timestamp, (int) now.tv_sec);
goto fail;
pa_sink_input_new_data_init(&data);
- pa_sink_input_new_data_set_sink(&data, sink, FALSE);
+ pa_sink_input_new_data_set_sink(&data, sink, false);
data.driver = __FILE__;
pa_proplist_sets(data.proplist, PA_PROP_MEDIA_ROLE, "stream");
pa_proplist_setf(data.proplist, PA_PROP_MEDIA_NAME,
s->intended_latency = s->sink_latency*2;
s->memblockq = pa_memblockq_new(
+ "module-rtp-recv memblockq",
0,
MEMBLOCKQ_MAXLENGTH,
MEMBLOCKQ_MAXLENGTH,
- pa_frame_size(&s->sink_input->sample_spec),
+ &s->sink_input->sample_spec,
pa_usec_to_bytes(s->intended_latency - s->sink_latency, &s->sink_input->sample_spec),
0,
0,
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);
static void sap_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event_flags_t flags, void *userdata) {
struct userdata *u = userdata;
- pa_bool_t goodbye = FALSE;
+ bool goodbye = false;
pa_sdp_info info;
struct session *s;
return;
if (goodbye) {
-
- if ((s = pa_hashmap_get(u->by_origin, info.origin)))
- session_free(s);
-
+ pa_hashmap_remove_and_free(u->by_origin, info.origin);
pa_sdp_info_destroy(&info);
} else {
k = pa_atomic_load(&s->timestamp);
if (k + DEATH_TIMEOUT < now.tv_sec)
- session_free(s);
+ pa_hashmap_remove_and_free(u->by_origin, s->sdp_info.origin);
}
/* Restart timer */
struct sockaddr *sa;
socklen_t salen;
const char *sap_address;
+ uint32_t latency_msec;
int fd = -1;
pa_assert(m);
goto fail;
}
+ latency_msec = DEFAULT_LATENCY_MSEC;
+ if (pa_modargs_get_value_u32(ma, "latency_msec", &latency_msec) < 0 || latency_msec < 1 || latency_msec > 300000) {
+ pa_log("Invalid latency specification");
+ goto fail;
+ }
+
if ((fd = mcast_socket(sa, salen)) < 0)
goto fail;
u->module = m;
u->core = m->core;
u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
+ u->latency = (pa_usec_t) latency_msec * PA_USEC_PER_MSEC;
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);
+ u->by_origin = pa_hashmap_new_full(pa_idxset_string_hash_func, pa_idxset_string_compare_func, NULL, (pa_free_cb_t) session_free);
u->check_death_event = pa_core_rttime_new(m->core, pa_rtclock_now() + DEATH_TIMEOUT * PA_USEC_PER_SEC, check_death_event_cb, u);
void pa__done(pa_module*m) {
struct userdata *u;
- struct session *s;
pa_assert(m);
pa_sap_context_destroy(&u->sap_context);
- if (u->by_origin) {
- while ((s = pa_hashmap_first(u->by_origin)))
- session_free(s);
-
- pa_hashmap_free(u->by_origin, NULL, NULL);
- }
+ if (u->by_origin)
+ pa_hashmap_free(u->by_origin);
pa_xfree(u->sink_name);
pa_xfree(u);