diff options
Diffstat (limited to 'src/modules/rtp')
| -rw-r--r-- | src/modules/rtp/Makefile | 13 | ||||
| -rw-r--r-- | src/modules/rtp/headerlist.c | 186 | ||||
| -rw-r--r-- | src/modules/rtp/headerlist.h | 46 | ||||
| -rw-r--r-- | src/modules/rtp/module-rtp-recv.c | 649 | ||||
| -rw-r--r-- | src/modules/rtp/module-rtp-send.c | 320 | ||||
| -rw-r--r-- | src/modules/rtp/rtp.c | 228 | ||||
| -rw-r--r-- | src/modules/rtp/rtp.h | 19 | ||||
| -rw-r--r-- | src/modules/rtp/rtsp_client.c | 530 | ||||
| -rw-r--r-- | src/modules/rtp/rtsp_client.h | 71 | ||||
| -rw-r--r-- | src/modules/rtp/sap.c | 131 | ||||
| -rw-r--r-- | src/modules/rtp/sap.h | 16 | ||||
| -rw-r--r-- | src/modules/rtp/sdp.c | 119 | ||||
| -rw-r--r-- | src/modules/rtp/sdp.h | 12 |
13 files changed, 1792 insertions, 548 deletions
diff --git a/src/modules/rtp/Makefile b/src/modules/rtp/Makefile deleted file mode 100644 index 316beb72..00000000 --- a/src/modules/rtp/Makefile +++ /dev/null @@ -1,13 +0,0 @@ -# This is a dirty trick just to ease compilation with emacs -# -# This file is not intended to be distributed or anything -# -# So: don't touch it, even better ignore it! - -all: - $(MAKE) -C ../.. - -clean: - $(MAKE) -C ../.. clean - -.PHONY: all clean diff --git a/src/modules/rtp/headerlist.c b/src/modules/rtp/headerlist.c new file mode 100644 index 00000000..0fef835b --- /dev/null +++ b/src/modules/rtp/headerlist.c @@ -0,0 +1,186 @@ +/*** + This file is part of PulseAudio. + + Copyright 2008 Colin Guthrie + Copyright 2007 Lennart Poettering + + PulseAudio is free software; you can redistribute it and/or modify + it under the terms of the GNU Lesser General Public License as + published by the Free Software Foundation; either version 2.1 of the + License, or (at your option) any later version. + + PulseAudio is distributed in the hope that it will be useful, but + WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public + License along with PulseAudio; if not, write to the Free Software + Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 + USA. +***/ + +#ifdef HAVE_CONFIG_H +#include <config.h> +#endif + +#include <string.h> + +#include <pulse/xmalloc.h> + +#include <pulsecore/hashmap.h> +#include <pulsecore/strbuf.h> +#include <pulsecore/core-util.h> + +#include "headerlist.h" + +struct header { + char *key; + void *value; + size_t nbytes; +}; + +#define MAKE_HASHMAP(p) ((pa_hashmap*) (p)) +#define MAKE_HEADERLIST(p) ((pa_headerlist*) (p)) + +static void header_free(struct header *hdr) { + pa_assert(hdr); + + pa_xfree(hdr->key); + pa_xfree(hdr->value); + pa_xfree(hdr); +} + +pa_headerlist* pa_headerlist_new(void) { + return MAKE_HEADERLIST(pa_hashmap_new(pa_idxset_string_hash_func, pa_idxset_string_compare_func)); +} + +void pa_headerlist_free(pa_headerlist* p) { + struct header *hdr; + + while ((hdr = pa_hashmap_steal_first(MAKE_HASHMAP(p)))) + header_free(hdr); + + pa_hashmap_free(MAKE_HASHMAP(p), NULL, NULL); +} + +int pa_headerlist_puts(pa_headerlist *p, const char *key, const char *value) { + struct header *hdr; + pa_bool_t add = FALSE; + + pa_assert(p); + pa_assert(key); + + if (!(hdr = pa_hashmap_get(MAKE_HASHMAP(p), key))) { + hdr = pa_xnew(struct header, 1); + hdr->key = pa_xstrdup(key); + add = TRUE; + } else + pa_xfree(hdr->value); + + hdr->value = pa_xstrdup(value); + hdr->nbytes = strlen(value)+1; + + if (add) + pa_hashmap_put(MAKE_HASHMAP(p), hdr->key, hdr); + + return 0; +} + +int pa_headerlist_putsappend(pa_headerlist *p, const char *key, const char *value) { + struct header *hdr; + pa_bool_t add = FALSE; + + pa_assert(p); + pa_assert(key); + + if (!(hdr = pa_hashmap_get(MAKE_HASHMAP(p), key))) { + hdr = pa_xnew(struct header, 1); + hdr->key = pa_xstrdup(key); + hdr->value = pa_xstrdup(value); + add = TRUE; + } else { + void *newval = pa_sprintf_malloc("%s%s", (char*)hdr->value, value); + pa_xfree(hdr->value); + hdr->value = newval; + } + hdr->nbytes = strlen(hdr->value)+1; + + if (add) + pa_hashmap_put(MAKE_HASHMAP(p), hdr->key, hdr); + + return 0; +} + +const char *pa_headerlist_gets(pa_headerlist *p, const char *key) { + struct header *hdr; + + pa_assert(p); + pa_assert(key); + + if (!(hdr = pa_hashmap_get(MAKE_HASHMAP(p), key))) + return NULL; + + if (hdr->nbytes <= 0) + return NULL; + + if (((char*) hdr->value)[hdr->nbytes-1] != 0) + return NULL; + + if (strlen((char*) hdr->value) != hdr->nbytes-1) + return NULL; + + return (char*) hdr->value; +} + +int pa_headerlist_remove(pa_headerlist *p, const char *key) { + struct header *hdr; + + pa_assert(p); + pa_assert(key); + + if (!(hdr = pa_hashmap_remove(MAKE_HASHMAP(p), key))) + return -1; + + header_free(hdr); + return 0; +} + +const char *pa_headerlist_iterate(pa_headerlist *p, void **state) { + struct header *hdr; + + if (!(hdr = pa_hashmap_iterate(MAKE_HASHMAP(p), state, NULL))) + return NULL; + + return hdr->key; +} + +char *pa_headerlist_to_string(pa_headerlist *p) { + const char *key; + void *state = NULL; + pa_strbuf *buf; + + pa_assert(p); + + buf = pa_strbuf_new(); + + while ((key = pa_headerlist_iterate(p, &state))) { + + const char *v; + + if ((v = pa_headerlist_gets(p, key))) + pa_strbuf_printf(buf, "%s: %s\r\n", key, v); + } + + return pa_strbuf_tostring_free(buf); +} + +int pa_headerlist_contains(pa_headerlist *p, const char *key) { + pa_assert(p); + pa_assert(key); + + if (!(pa_hashmap_get(MAKE_HASHMAP(p), key))) + return 0; + + return 1; +} diff --git a/src/modules/rtp/headerlist.h b/src/modules/rtp/headerlist.h new file mode 100644 index 00000000..4b9c6433 --- /dev/null +++ b/src/modules/rtp/headerlist.h @@ -0,0 +1,46 @@ +#ifndef foopulseheaderlisthfoo +#define foopulseheaderlisthfoo + +/*** + This file is part of PulseAudio. + + Copyright 2008 Colin Guthrie + Copyright 2007 Lennart Poettering + + PulseAudio is free software; you can redistribute it and/or modify + it under the terms of the GNU Lesser General Public License as + published by the Free Software Foundation; either version 2.1 of the + License, or (at your option) any later version. + + PulseAudio is distributed in the hope that it will be useful, but + WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public + License along with PulseAudio; if not, write to the Free Software + Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 + USA. +***/ + +#include <pulsecore/macro.h> + +typedef struct pa_headerlist pa_headerlist; + +pa_headerlist* pa_headerlist_new(void); +void pa_headerlist_free(pa_headerlist* p); + +int pa_headerlist_puts(pa_headerlist *p, const char *key, const char *value); +int pa_headerlist_putsappend(pa_headerlist *p, const char *key, const char *value); + +const char *pa_headerlist_gets(pa_headerlist *p, const char *key); + +int pa_headerlist_remove(pa_headerlist *p, const char *key); + +const char *pa_headerlist_iterate(pa_headerlist *p, void **state); + +char *pa_headerlist_to_string(pa_headerlist *p); + +int pa_headerlist_contains(pa_headerlist *p, const char *key); + +#endif diff --git a/src/modules/rtp/module-rtp-recv.c b/src/modules/rtp/module-rtp-recv.c index 0359a43b..7025c15a 100644 --- a/src/modules/rtp/module-rtp-recv.c +++ b/src/modules/rtp/module-rtp-recv.c @@ -1,17 +1,19 @@ /*** This file is part of PulseAudio. - + + Copyright 2006 Lennart Poettering + PulseAudio is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published - by the Free Software Foundation; either version 2 of the License, + by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version. - + PulseAudio is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. - + You should have received a copy of the GNU Lesser General Public License along with PulseAudio; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 @@ -22,15 +24,15 @@ #include <config.h> #endif -#include <assert.h> #include <stdio.h> #include <sys/socket.h> #include <netinet/in.h> -#include <arpa/inet.h> #include <errno.h> #include <string.h> #include <unistd.h> +#include <math.h> +#include <pulse/rtclock.h> #include <pulse/timeval.h> #include <pulse/xmalloc.h> @@ -41,10 +43,17 @@ #include <pulsecore/sink-input.h> #include <pulsecore/memblockq.h> #include <pulsecore/log.h> +#include <pulsecore/core-rtclock.h> #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 <pulsecore/atomic.h> +#include <pulsecore/once.h> +#include <pulsecore/poll.h> +#include <pulsecore/arpa-inet.h> #include "module-rtp-recv-symdef.h" @@ -52,19 +61,22 @@ #include "sdp.h" #include "sap.h" -PA_MODULE_AUTHOR("Lennart Poettering") -PA_MODULE_DESCRIPTION("Recieve data from a network via RTP/SAP/SDP") -PA_MODULE_VERSION(PACKAGE_VERSION) +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_USAGE( "sink=<name of the sink> " "sap_address=<multicast address to listen on> " -) +); #define SAP_PORT 9875 #define DEFAULT_SAP_ADDRESS "224.0.0.56" -#define MEMBLOCKQ_MAXLENGTH (1024*170) +#define MEMBLOCKQ_MAXLENGTH (1024*1024*40) #define MAX_SESSIONS 16 -#define DEATH_TIMEOUT 20000000 +#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", @@ -74,20 +86,30 @@ 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; + + pa_usec_t intended_latency; + pa_usec_t sink_latency; + + pa_usec_t last_rate_update; + pa_usec_t last_latency; + double estimated_rate; + double avg_estimated_rate; }; struct userdata { @@ -97,83 +119,135 @@ struct userdata { 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); + +/* 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; -static int sink_input_peek(pa_sink_input *i, pa_memchunk *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); +} + +/* Called from I/O thread context */ +static int sink_input_pop_cb(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); + + if (pa_memblockq_peek(s->memblockq, chunk) < 0) + return -1; - return pa_memblockq_peek(s->memblockq, chunk); + pa_memblockq_drop(s->memblockq, chunk->length); + + return 0; } -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_process_rewind_cb(pa_sink_input *i, size_t nbytes) { struct session *s; - assert(i); - s = i->userdata; - pa_memblockq_drop(s->memblockq, chunk, length); + pa_sink_input_assert_ref(i); + pa_assert_se(s = i->userdata); + + pa_memblockq_rewind(s->memblockq, nbytes); } +/* Called from I/O thread context */ +static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) { + struct session *s; + + pa_sink_input_assert_ref(i); + pa_assert_se(s = i->userdata); + + pa_memblockq_set_maxrewind(s->memblockq, nbytes); +} + +/* 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); - session_free(s, 1); + session_free(s); } -static pa_usec_t sink_input_get_latency(pa_sink_input *i) { +/* Called from IO context */ +static void sink_input_suspend_within_thread(pa_sink_input* i, pa_bool_t b) { 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); + if (b) + pa_memblockq_flush_read(s->memblockq); + else + s->first_packet = FALSE; } -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; - - assert(m); - 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->memblock_stat) < 0) - return; + struct timeval now = { 0, 0 }; + struct session *s; + struct pollfd *p; + + pa_assert_se(s = pa_rtpoll_item_get_userdata(i)); + + 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 (s->sdp_info.payload != s->rtp_context.payload) { + if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->module->core->mempool, &now) < 0) + return 0; + + if (s->sdp_info.payload != s->rtp_context.payload || + !PA_SINK_IS_OPENED(s->sink_input->sink->thread_info.state)) { 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(__FILE__": 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; } } - /* Check wheter there was a timestamp overflow */ + /* Check whether there was a timestamp overflow */ k = (int64_t) s->rtp_context.timestamp - (int64_t) s->offset; j = (int64_t) 0x100000000LL - (int64_t) s->offset + (int64_t) s->rtp_context.timestamp; @@ -181,76 +255,217 @@ static void rtp_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event delta = k; else delta = j; - - pa_memblockq_seek(s->memblockq, delta * s->rtp_context.frame_size, PA_SEEK_RELATIVE); + + 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 { + pa_log_warn("Using artificial time instead of timestamp"); + } PA_ONCE_END; + pa_rtclock_get(&now); + } else + pa_rtclock_from_wallclock(&now); if (pa_memblockq_push(s->memblockq, &chunk) < 0) { - /* queue overflow, let's flush it and try again */ - pa_memblockq_flush(s->memblockq); - pa_memblockq_push(s->memblockq, &chunk); + pa_log_warn("Queue overrun"); + pa_memblockq_seek(s->memblockq, (int64_t) chunk.length, PA_SEEK_RELATIVE, TRUE); } - - /* The next timestamp we expect */ - s->offset = s->rtp_context.timestamp + (chunk.length / s->rtp_context.frame_size); - + +/* pa_log("blocks in q: %u", pa_memblockq_get_nblocks(s->memblockq)); */ + pa_memblock_unref(chunk.memblock); - /* Reset death timer */ - pa_gettimeofday(&tv); - pa_timeval_add(&tv, DEATH_TIMEOUT); - m->time_restart(s->death_event, &tv); + /* The next timestamp we expect */ + s->offset = s->rtp_context.timestamp + (uint32_t) (chunk.length / s->rtp_context.frame_size); + + pa_atomic_store(&s->timestamp, (int) now.tv_sec); + + if (s->last_rate_update + RATE_UPDATE_INTERVAL < pa_timeval_load(&now)) { + pa_usec_t wi, ri, render_delay, sink_delay = 0, latency; + uint32_t base_rate = s->sink_input->sink->sample_spec.rate; + uint32_t current_rate = s->sink_input->sample_spec.rate; + uint32_t new_rate; + double estimated_rate, alpha = 0.02; + + pa_log_debug("Updating sample rate"); + + wi = pa_bytes_to_usec((uint64_t) pa_memblockq_get_write_index(s->memblockq), &s->sink_input->sample_spec); + ri = pa_bytes_to_usec((uint64_t) pa_memblockq_get_read_index(s->memblockq), &s->sink_input->sample_spec); + + pa_log_debug("wi=%lu ri=%lu", (unsigned long) wi, (unsigned long) ri); + + sink_delay = pa_sink_get_latency_within_thread(s->sink_input->sink); + render_delay = pa_bytes_to_usec(pa_memblockq_get_length(s->sink_input->thread_info.render_memblockq), &s->sink_input->sink->sample_spec); + + if (ri > render_delay+sink_delay) + ri -= render_delay+sink_delay; + else + ri = 0; + + if (wi < ri) + latency = 0; + else + latency = wi - ri; + + pa_log_debug("Write index deviates by %0.2f ms, expected %0.2f ms", (double) latency/PA_USEC_PER_MSEC, (double) s->intended_latency/PA_USEC_PER_MSEC); + + /* The buffer is filling with some unknown rate R̂ samples/second. If the rate of reading in + * the last T seconds was Rⁿ, then the increase in buffer latency ΔLⁿ = Lⁿ - Lⁿ⁻ⁱ in that + * same period is ΔLⁿ = (TR̂ - TRⁿ) / R̂, giving the estimated target rate + * T + * R̂ = ─────────────── Rⁿ . (1) + * T - (Lⁿ - Lⁿ⁻ⁱ) + * + * Setting the sample rate to R̂ results in the latency being constant (if the estimate of R̂ + * is correct). But there is also the requirement to keep the buffer at a predefined target + * latency L̂. So instead of setting Rⁿ⁺ⁱ to R̂ immediately, the strategy will be to reduce R + * from Rⁿ⁺ⁱ to R̂ in a steps of T seconds, where Rⁿ⁺ⁱ is chosen such that in the total time + * aT the latency is reduced from Lⁿ to L̂. This strategy translates to the requirements + * ₐ R̂ - Rⁿ⁺ʲ a-j+1 j-1 + * Σ T ────────── = L̂ - Lⁿ with Rⁿ⁺ʲ = ───── Rⁿ⁺ⁱ + ───── R̂ . + * ʲ⁼ⁱ R̂ a a + * Solving for Rⁿ⁺ⁱ gives + * T - ²∕ₐ₊₁(L̂ - Lⁿ) + * Rⁿ⁺ⁱ = ───────────────── R̂ . (2) + * T + * In the code below a = 7 is used. + * + * Equation (1) is not directly used in (2), but instead an exponentially weighted average + * of the estimated rate R̂ is used. This average R̅ is defined as + * R̅ⁿ = α R̂ⁿ + (1-α) R̅ⁿ⁻ⁱ . + * Because it is difficult to find a fixed value for the coefficient α such that the + * averaging is without significant lag but oscillations are filtered out, a heuristic is + * used. When the successive estimates R̂ⁿ do not change much then α→1, but when there is a + * sudden spike in the estimated rate α→0, such that the deviation is given little weight. + */ + estimated_rate = (double) current_rate * (double) RATE_UPDATE_INTERVAL / (double) (RATE_UPDATE_INTERVAL + s->last_latency - latency); + if (fabs(s->estimated_rate - s->avg_estimated_rate) > 1) { + double ratio = (estimated_rate + s->estimated_rate - 2*s->avg_estimated_rate) / (s->estimated_rate - s->avg_estimated_rate); + alpha = PA_CLAMP(2 * (ratio + fabs(ratio)) / (4 + ratio*ratio), 0.02, 0.8); + } + s->avg_estimated_rate = alpha * estimated_rate + (1-alpha) * s->avg_estimated_rate; + s->estimated_rate = estimated_rate; + pa_log_debug("Estimated target rate: %.0f Hz, using average of %.0f Hz (α=%.3f)", estimated_rate, s->avg_estimated_rate, alpha); + new_rate = (uint32_t) ((double) (RATE_UPDATE_INTERVAL + latency/4 - s->intended_latency/4) / (double) RATE_UPDATE_INTERVAL * s->avg_estimated_rate); + s->last_latency = latency; + + if (new_rate < (uint32_t) (base_rate*0.8) || new_rate > (uint32_t) (base_rate*1.25)) { + pa_log_warn("Sample rates too different, not adjusting (%u vs. %u).", base_rate, new_rate); + new_rate = base_rate; + } else { + if (base_rate < new_rate + 20 && new_rate < base_rate + 20) + new_rate = base_rate; + /* Do the adjustment in small steps; 2‰ can be considered inaudible */ + if (new_rate < (uint32_t) (current_rate*0.998) || new_rate > (uint32_t) (current_rate*1.002)) { + pa_log_info("New rate of %u Hz not within 2‰ of %u Hz, forcing smaller adjustment", new_rate, current_rate); + new_rate = PA_CLAMP(new_rate, (uint32_t) (current_rate*0.998), (uint32_t) (current_rate*1.002)); + } + } + s->sink_input->sample_spec.rate = new_rate; + + pa_assert(pa_sample_spec_valid(&s->sink_input->sample_spec)); + + pa_resampler_set_input_rate(s->sink_input->thread_info.resampler, s->sink_input->sample_spec.rate); + + pa_log_debug("Updated sampling rate to %lu Hz.", (unsigned long) s->sink_input->sample_spec.rate); + + s->last_rate_update = pa_timeval_load(&now); + } + + if (pa_memblockq_is_readable(s->memblockq) && + s->sink_input->thread_info.underrun_for > 0) { + 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); + } + + 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; - - assert(m); - assert(t); - assert(tv); - assert(s); +/* 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); - session_free(s, 1); + pa_assert(!s->rtpoll_item); + s->rtpoll_item = pa_rtpoll_item_new(i->sink->thread_info.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) { 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(__FILE__": Failed to create socket: %s", pa_cstrerror(errno)); + if ((fd = pa_socket_cloexec(af, SOCK_DGRAM, 0)) < 0) { + pa_log("Failed to create socket: %s", pa_cstrerror(errno)); + goto fail; + } + + pa_make_udp_socket_low_delay(fd); + + one = 1; + if (setsockopt(fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one)) < 0) { + pa_log("SO_TIMESTAMP failed: %s", pa_cstrerror(errno)); goto fail; } one = 1; if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0) { - pa_log(__FILE__": SO_REUSEADDR failed: %s", pa_cstrerror(errno)); + pa_log("SO_REUSEADDR failed: %s", pa_cstrerror(errno)); goto fail; } - + 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)); +#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)); +#endif } if (r < 0) { - pa_log_info(__FILE__": Joining mcast group failed: %s", pa_cstrerror(errno)); + pa_log_info("Joining mcast group failed: %s", pa_cstrerror(errno)); goto fail; } - + if (bind(fd, sa, salen) < 0) { - pa_log(__FILE__": bind() failed: %s", pa_cstrerror(errno)); + pa_log("bind() failed: %s", pa_cstrerror(errno)); goto fail; } return fd; - + fail: if (fd >= 0) close(fd); @@ -260,130 +475,150 @@ 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_memchunk 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(__FILE__": 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(__FILE__": sink does not exist."); + + if (!(sink = pa_namereg_get(u->module->core, u->sink_name, PA_NAMEREG_SINK))) { + pa_log("Sink does not exist."); goto fail; } + pa_rtclock_get(&now); + 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; + s->intended_latency = LATENCY_USEC; + s->last_rate_update = pa_timeval_load(&now); + s->last_latency = LATENCY_USEC; + 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); if ((fd = mcast_socket((const struct sockaddr*) &sdp_info->sa, sdp_info->salen)) < 0) goto fail; - c = pa_sprintf_malloc("RTP Stream%s%s%s", - sdp_info->session_name ? " (" : "", - sdp_info->session_name ? sdp_info->session_name : "", - sdp_info->session_name ? ")" : ""); + pa_sink_input_new_data_init(&data); + 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, + "RTP Stream%s%s%s", + sdp_info->session_name ? " (" : "", + sdp_info->session_name ? sdp_info->session_name : "", + sdp_info->session_name ? ")" : ""); + + if (sdp_info->session_name) + pa_proplist_sets(data.proplist, "rtp.session", sdp_info->session_name); + pa_proplist_sets(data.proplist, "rtp.origin", sdp_info->origin); + pa_proplist_setf(data.proplist, "rtp.payload", "%u", (unsigned) sdp_info->payload); + data.module = u->module; + pa_sink_input_new_data_set_sample_spec(&data, &sdp_info->sample_spec); + data.flags = PA_SINK_INPUT_VARIABLE_RATE; + + pa_sink_input_new(&s->sink_input, u->module->core, &data); + pa_sink_input_new_data_done(&data); - s->sink_input = pa_sink_input_new(sink, __FILE__, c, &sdp_info->sample_spec, NULL, NULL, 0, PA_RESAMPLER_INVALID); - pa_xfree(c); - if (!s->sink_input) { - pa_log(__FILE__": failed to create sink input."); + pa_log("Failed to create sink input."); goto fail; } s->sink_input->userdata = s; - s->sink_input->owner = u->module; - s->sink_input->peek = sink_input_peek; - s->sink_input->drop = sink_input_drop; + s->sink_input->parent.process_msg = sink_input_process_msg; + s->sink_input->pop = sink_input_pop_cb; + s->sink_input->process_rewind = sink_input_process_rewind_cb; + s->sink_input->update_max_rewind = sink_input_update_max_rewind_cb; 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; + s->sink_input->suspend_within_thread = sink_input_suspend_within_thread; + + pa_sink_input_get_silence(s->sink_input, &silence); + + s->sink_latency = pa_sink_input_set_requested_latency(s->sink_input, s->intended_latency/2); + + if (s->intended_latency < s->sink_latency*2) + s->intended_latency = s->sink_latency*2; - silence = pa_silence_memblock_new(&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), - s->userdata->core->memblock_stat); - s->memblockq = pa_memblockq_new( 0, MEMBLOCKQ_MAXLENGTH, MEMBLOCKQ_MAXLENGTH, pa_frame_size(&s->sink_input->sample_spec), - pa_bytes_per_second(&s->sink_input->sample_spec)/10+1, + pa_usec_to_bytes(s->intended_latency - s->sink_latency, &s->sink_input->sample_spec), 0, - silence, - u->core->memblock_stat); + 0, + &silence); - pa_memblock_unref(silence); + pa_memblock_unref(silence.memblock); - 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_sink_input_put(s->sink_input); - pa_log_info(__FILE__": Found new session '%s'", s->sdp_info.session_name); + pa_log_info("New session '%s'", s->sdp_info.session_name); - u->n_sessions++; - 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); - - pa_log_info(__FILE__": 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); +static void session_free(struct session *s) { + pa_assert(s); - if (from_hash) - pa_hashmap_remove(s->userdata->by_origin, s->sdp_info.origin); + pa_log_info("Freeing session '%s'", s->sdp_info.session_name); - 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); } 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; - int goodbye; + pa_bool_t goodbye = FALSE; 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; @@ -394,78 +629,107 @@ 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 { if (!(s = pa_hashmap_get(u->by_origin, info.origin))) { - if (!(s = session_new(u, &info))) + if (!session_new(u, &info)) 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, (int) 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 *tv, void *userdata) { + struct session *s, *n; + struct userdata *u = userdata; + struct timeval now; + + pa_assert(m); + pa_assert(t); + 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_core_rttime_restart(u->module->core, t, pa_rtclock_now() + DEATH_TIMEOUT * PA_USEC_PER_SEC); +} + +int pa__init(pa_module*m) { struct userdata *u; pa_modargs *ma = NULL; struct sockaddr_in sa4; +#ifdef HAVE_IPV6 struct sockaddr_in6 sa6; +#endif struct sockaddr *sa; socklen_t salen; const char *sap_address; int fd = -1; - - assert(c); - assert(m); + + pa_assert(m); if (!(ma = pa_modargs_new(m->argument, valid_modargs))) { - pa_log(__FILE__": failed to parse module arguments"); + pa_log("failed to parse module arguments"); goto fail; } sap_address = pa_modargs_get_value(ma, "sap_address", DEFAULT_SAP_ADDRESS); - - if (inet_pton(AF_INET6, sap_address, &sa6.sin6_addr) > 0) { - sa6.sin6_family = AF_INET6; - sa6.sin6_port = htons(SAP_PORT); - sa = (struct sockaddr*) &sa6; - salen = sizeof(sa6); - } else if (inet_pton(AF_INET, sap_address, &sa4.sin_addr) > 0) { + + if (inet_pton(AF_INET, sap_address, &sa4.sin_addr) > 0) { sa4.sin_family = AF_INET; sa4.sin_port = htons(SAP_PORT); sa = (struct sockaddr*) &sa4; salen = sizeof(sa4); +#ifdef HAVE_IPV6 + } else if (inet_pton(AF_INET6, sap_address, &sa6.sin6_addr) > 0) { + sa6.sin6_family = AF_INET6; + sa6.sin6_port = htons(SAP_PORT); + sa = (struct sockaddr*) &sa6; + salen = sizeof(sa6); +#endif } else { - pa_log(__FILE__": invalid SAP address '%s'", sap_address); + pa_log("Invalid SAP address '%s'", sap_address); goto fail; } if ((fd = mcast_socket(sa, salen)) < 0) goto fail; - u = pa_xnew(struct userdata, 1); - m->userdata = u; + m->userdata = u = pa_xnew(struct userdata, 1); u->module = m; - u->core = c; + u->core = m->core; 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); - + + u->check_death_event = pa_core_rttime_new(m->core, pa_rtclock_now() + DEATH_TIMEOUT * PA_USEC_PER_SEC, check_death_event_cb, u); + pa_modargs_free(ma); return 0; @@ -475,28 +739,35 @@ fail: pa_modargs_free(ma); if (fd >= 0) - close(fd); - - return -1; -} + pa_close(fd); -static void free_func(void *p, PA_GCC_UNUSED void *userdata) { - session_free(p, 0); + return -1; } -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_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 5c47047f..7131629c 100644 --- a/src/modules/rtp/module-rtp-send.c +++ b/src/modules/rtp/module-rtp-send.c @@ -1,18 +1,18 @@ -/* $Id$ */ - /*** This file is part of PulseAudio. - + + Copyright 2006 Lennart Poettering + PulseAudio is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published - by the Free Software Foundation; either version 2 of the License, + by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version. - + PulseAudio is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. - + You should have received a copy of the GNU Lesser General Public License along with PulseAudio; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 @@ -23,22 +23,19 @@ #include <config.h> #endif -#include <assert.h> #include <stdio.h> #include <sys/socket.h> #include <netinet/in.h> -#include <arpa/inet.h> #include <errno.h> -#include <string.h> #include <unistd.h> +#include <pulse/rtclock.h> #include <pulse/timeval.h> #include <pulse/util.h> #include <pulse/xmalloc.h> #include <pulsecore/core-error.h> #include <pulsecore/module.h> -#include <pulsecore/llist.h> #include <pulsecore/source.h> #include <pulsecore/source-output.h> #include <pulsecore/memblockq.h> @@ -46,6 +43,10 @@ #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 <pulsecore/arpa-inet.h> #include "module-rtp-send-symdef.h" @@ -53,9 +54,10 @@ #include "sdp.h" #include "sap.h" -PA_MODULE_AUTHOR("Lennart Poettering") -PA_MODULE_DESCRIPTION("Read data from source and send it to the network via RTP/SAP/SDP") -PA_MODULE_VERSION(PACKAGE_VERSION) +PA_MODULE_AUTHOR("Lennart Poettering"); +PA_MODULE_DESCRIPTION("Read data from source and send it to the network via RTP/SAP/SDP"); +PA_MODULE_VERSION(PACKAGE_VERSION); +PA_MODULE_LOAD_ONCE(FALSE); PA_MODULE_USAGE( "source=<name of the source> " "format=<sample format> " @@ -64,15 +66,17 @@ PA_MODULE_USAGE( "destination=<destination IP address> " "port=<port number> " "mtu=<maximum transfer unit> " - "loop=<loopback to local host?>" -) + "loop=<loopback to local host?> " + "ttl=<ttl value>" +); #define DEFAULT_PORT 46000 +#define DEFAULT_TTL 1 #define SAP_PORT 9875 #define DEFAULT_DESTINATION "224.0.0.56" #define MEMBLOCKQ_MAXLENGTH (1024*170) #define DEFAULT_MTU 1280 -#define SAP_INTERVAL 5000000 +#define SAP_INTERVAL (5*PA_USEC_PER_SEC) static const char* const valid_modargs[] = { "source", @@ -81,13 +85,14 @@ static const char* const valid_modargs[] = { "rate", "destination", "port", + "mtu" , "loop", + "ttl", NULL }; struct userdata { pa_module *module; - pa_core *core; pa_source_output *source_output; pa_memblockq *memblockq; @@ -99,91 +104,101 @@ 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(__FILE__": 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_module_unload_request(u->module, TRUE); - 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(u); pa_sap_send(&u->sap_context, 0); - pa_gettimeofday(&next); - pa_timeval_add(&next, SAP_INTERVAL); - m->time_restart(t, &next); + pa_core_rttime_restart(u->module->core, t, pa_rtclock_now() + SAP_INTERVAL); } -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; uint32_t port = DEFAULT_PORT, mtu; - int af, fd = -1, sap_fd = -1; + uint32_t ttl = DEFAULT_TTL; + sa_family_t af; + int fd = -1, sap_fd = -1; pa_source *s; pa_sample_spec ss; pa_channel_map cm; struct sockaddr_in sa4, sap_sa4; +#ifdef HAVE_IPV6 struct sockaddr_in6 sa6, sap_sa6; +#endif struct sockaddr_storage sa_dst; pa_source_output *o = NULL; uint8_t payload; char *p; - int r; + int r, j; socklen_t k; - struct timeval tv; char hn[128], *n; - int loop = 0; - - assert(c); - assert(m); + pa_bool_t loop = FALSE; + pa_source_output_new_data data; + + pa_assert(m); if (!(ma = pa_modargs_new(m->argument, valid_modargs))) { - pa_log(__FILE__": 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(__FILE__": source does not exist."); + if (!(s = pa_namereg_get(m->core, pa_modargs_get_value(ma, "source", NULL), PA_NAMEREG_SOURCE))) { + pa_log("Source does not exist."); goto fail; } if (pa_modargs_get_value_boolean(ma, "loop", &loop) < 0) { - pa_log(__FILE__": failed to parse \"loop\" parameter."); + pa_log("Failed to parse \"loop\" parameter."); goto fail; } @@ -191,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(__FILE__": failed to parse sample specification"); + pa_log("Failed to parse sample specification"); goto fail; } if (!pa_rtp_sample_spec_valid(&ss)) { - pa_log(__FILE__": specified sample type not compatible with RTP"); + pa_log("Specified sample type not compatible with RTP"); goto fail; } @@ -205,83 +220,133 @@ 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 = (uint32_t) 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(__FILE__": invalid mtu."); + pa_log("Invalid MTU."); goto fail; } - port = DEFAULT_PORT + ((rand() % 512) << 1); + port = DEFAULT_PORT + ((uint32_t) (rand() % 512) << 1); if (pa_modargs_get_value_u32(ma, "port", &port) < 0 || port < 1 || port > 0xFFFF) { - pa_log(__FILE__": port= expects a numerical argument between 1 and 65535."); + pa_log("port= expects a numerical argument between 1 and 65535."); goto fail; } if (port & 1) - pa_log_warn(__FILE__": WARNING: port number not even as suggested in RFC3550!"); + pa_log_warn("Port number not even as suggested in RFC3550!"); + + if (pa_modargs_get_value_u32(ma, "ttl", &ttl) < 0 || ttl < 1 || ttl > 0xFF) { + pa_log("ttl= expects a numerical argument between 1 and 255."); + goto fail; + } dest = pa_modargs_get_value(ma, "destination", DEFAULT_DESTINATION); - if (inet_pton(AF_INET6, dest, &sa6.sin6_addr) > 0) { - sa6.sin6_family = af = AF_INET6; - sa6.sin6_port = htons(port); - sap_sa6 = sa6; - sap_sa6.sin6_port = htons(SAP_PORT); - } else if (inet_pton(AF_INET, dest, &sa4.sin_addr) > 0) { + if (inet_pton(AF_INET, dest, &sa4.sin_addr) > 0) { sa4.sin_family = af = AF_INET; - sa4.sin_port = htons(port); + sa4.sin_port = htons((uint16_t) port); sap_sa4 = sa4; sap_sa4.sin_port = htons(SAP_PORT); +#ifdef HAVE_IPV6 + } else if (inet_pton(AF_INET6, dest, &sa6.sin6_addr) > 0) { + sa6.sin6_family = af = AF_INET6; + sa6.sin6_port = htons((uint16_t) port); + sap_sa6 = sa6; + sap_sa6.sin6_port = htons(SAP_PORT); +#endif } else { - pa_log(__FILE__": invalid destination '%s'", dest); + pa_log("Invalid destination '%s'", dest); goto fail; } - - if ((fd = socket(af, SOCK_DGRAM, 0)) < 0) { - pa_log(__FILE__": socket() failed: %s", pa_cstrerror(errno)); + + if ((fd = pa_socket_cloexec(af, SOCK_DGRAM, 0)) < 0) { + pa_log("socket() failed: %s", pa_cstrerror(errno)); goto fail; } - if (connect(fd, af == AF_INET ? (struct sockaddr*) &sa4 : (struct sockaddr*) &sa6, af == AF_INET ? sizeof(sa4) : sizeof(sa6)) < 0) { - pa_log(__FILE__": connect() failed: %s", pa_cstrerror(errno)); + if (af == AF_INET && connect(fd, (struct sockaddr*) &sa4, sizeof(sa4)) < 0) { + pa_log("connect() failed: %s", pa_cstrerror(errno)); goto fail; +#ifdef HAVE_IPV6 + } else if (af == AF_INET6 && connect(fd, (struct sockaddr*) &sa6, sizeof(sa6)) < 0) { + pa_log("connect() failed: %s", pa_cstrerror(errno)); + goto fail; +#endif } - if ((sap_fd = socket(af, SOCK_DGRAM, 0)) < 0) { - pa_log(__FILE__": socket() failed: %s", pa_cstrerror(errno)); + if ((sap_fd = pa_socket_cloexec(af, SOCK_DGRAM, 0)) < 0) { + pa_log("socket() failed: %s", pa_cstrerror(errno)); goto fail; } - if (connect(sap_fd, af == AF_INET ? (struct sockaddr*) &sap_sa4 : (struct sockaddr*) &sap_sa6, af == AF_INET ? sizeof(sap_sa4) : sizeof(sap_sa6)) < 0) { - pa_log(__FILE__": connect() failed: %s", pa_cstrerror(errno)); + if (af == AF_INET && connect(sap_fd, (struct sockaddr*) &sap_sa4, sizeof(sap_sa4)) < 0) { + pa_log("connect() failed: %s", pa_cstrerror(errno)); + goto fail; +#ifdef HAVE_IPV6 + } else if (af == AF_INET6 && connect(sap_fd, (struct sockaddr*) &sap_sa6, sizeof(sap_sa6)) < 0) { + pa_log("connect() failed: %s", pa_cstrerror(errno)); goto fail; +#endif } - if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_LOOP, &loop, sizeof(loop)) < 0 || - setsockopt(sap_fd, IPPROTO_IP, IP_MULTICAST_LOOP, &loop, sizeof(loop)) < 0) { - pa_log(__FILE__": IP_MULTICAST_LOOP failed: %s", pa_cstrerror(errno)); + j = !!loop; + if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_LOOP, &j, sizeof(j)) < 0 || + setsockopt(sap_fd, IPPROTO_IP, IP_MULTICAST_LOOP, &j, sizeof(j)) < 0) { + pa_log("IP_MULTICAST_LOOP failed: %s", pa_cstrerror(errno)); goto fail; } - - if (!(o = pa_source_output_new(s, __FILE__, "RTP Monitor Stream", &ss, &cm, PA_RESAMPLER_INVALID))) { - pa_log(__FILE__": failed to create source output."); + + if (ttl != DEFAULT_TTL) { + int _ttl = (int) ttl; + + if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_TTL, &_ttl, sizeof(_ttl)) < 0) { + pa_log("IP_MULTICAST_TTL failed: %s", pa_cstrerror(errno)); + goto fail; + } + + if (setsockopt(sap_fd, IPPROTO_IP, IP_MULTICAST_TTL, &_ttl, sizeof(_ttl)) < 0) { + pa_log("IP_MULTICAST_TTL (sap) failed: %s", pa_cstrerror(errno)); + 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_source_output_new_data_init(&data); + pa_proplist_sets(data.proplist, PA_PROP_MEDIA_NAME, "RTP Monitor Stream"); + pa_proplist_sets(data.proplist, "rtp.destination", dest); + pa_proplist_setf(data.proplist, "rtp.mtu", "%lu", (unsigned long) mtu); + pa_proplist_setf(data.proplist, "rtp.port", "%lu", (unsigned long) port); + pa_proplist_setf(data.proplist, "rtp.ttl", "%lu", (unsigned long) ttl); + data.driver = __FILE__; + data.module = m; + pa_source_output_new_data_set_source(&data, s, FALSE); + pa_source_output_new_data_set_sample_spec(&data, &ss); + pa_source_output_new_data_set_channel_map(&data, &cm); + data.flags = PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND; + + pa_source_output_new(&o, m->core, &data); + pa_source_output_new_data_done(&data); + + if (!o) { + 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; - o->owner = m; - - u = pa_xnew(struct userdata, 1); - m->userdata = u; - o->userdata = u; + pa_log_info("Configured source latency of %llu ms.", + (unsigned long long) pa_source_output_set_requested_latency(o, pa_bytes_to_usec(mtu, &o->sample_spec)) / PA_USEC_PER_MSEC); + + m->userdata = o->userdata = u = pa_xnew(struct userdata, 1); u->module = m; - u->core = c; u->source_output = o; - + u->memblockq = pa_memblockq_new( 0, MEMBLOCKQ_MAXLENGTH, @@ -289,35 +354,43 @@ int pa__init(pa_core *c, pa_module*m) { pa_frame_size(&ss), 1, 0, - NULL, - c->memblock_stat); + 0, + NULL); 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))); - - p = pa_sdp_build(af, - af == AF_INET ? (void*) &((struct sockaddr_in*) &sa_dst)->sin_addr : (void*) &((struct sockaddr_in6*) &sa_dst)->sin6_addr, - af == AF_INET ? (void*) &sa4.sin_addr : (void*) &sa6.sin6_addr, - n, port, payload, &ss); + + if (af == AF_INET) { + p = pa_sdp_build(af, + (void*) &((struct sockaddr_in*) &sa_dst)->sin_addr, + (void*) &sa4.sin_addr, + n, (uint16_t) port, payload, &ss); +#ifdef HAVE_IPV6 + } else { + p = pa_sdp_build(af, + (void*) &((struct sockaddr_in6*) &sa_dst)->sin6_addr, + (void*) &sa6.sin6_addr, + n, (uint16_t) port, payload, &ss); +#endif + } 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(__FILE__": 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(__FILE__": SDP-Data:\n%s\n"__FILE__": EOF", p); + pa_log_info("RTP stream initialized with mtu %u on %s:%u ttl=%u, SSRC=0x%08x, payload=%u, initial sequence #%u", mtu, dest, port, ttl, u->rtp_context.ssrc, payload, u->rtp_context.sequence); + 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); + u->sap_event = pa_core_rttime_new(m->core, pa_rtclock_now() + SAP_INTERVAL, sap_event_cb, u); + + pa_source_output_put(u->source_output); pa_modargs_free(ma); @@ -328,31 +401,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); } @@ -361,7 +434,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 ee037d42..05c736a7 100644 --- a/src/modules/rtp/rtp.c +++ b/src/modules/rtp/rtp.c @@ -1,18 +1,18 @@ -/* $Id$ */ - /*** This file is part of PulseAudio. - + + Copyright 2006 Lennart Poettering + PulseAudio is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published - by the Free Software Foundation; either version 2 of the License, + by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version. - + PulseAudio is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. - + You should have received a copy of the GNU Lesser General Public License along with PulseAudio; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 @@ -23,12 +23,9 @@ #include <config.h> #endif -#include <assert.h> -#include <fcntl.h> #include <stdlib.h> #include <string.h> #include <errno.h> -#include <arpa/inet.h> #include <unistd.h> #include <sys/ioctl.h> @@ -36,22 +33,31 @@ #include <sys/filio.h> #endif +#ifdef HAVE_SYS_UIO_H +#include <sys/uio.h> +#endif + #include <pulsecore/core-error.h> #include <pulsecore/log.h> +#include <pulsecore/macro.h> +#include <pulsecore/core-util.h> +#include <pulsecore/arpa-inet.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()); c->timestamp = 0; c->ssrc = ssrc ? ssrc : (uint32_t) (rand()*rand()); - c->payload = payload & 127; + c->payload = (uint8_t) (payload & 127U); c->frame_size = frame_size; - + + pa_memchunk_reset(&c->memchunk); + return c; } @@ -61,40 +67,43 @@ 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; - - assert(c); - assert(size > 0); - assert(q); + size_t n = 0; + + pa_assert(c); + pa_assert(size > 0); + pa_assert(q); if (pa_memblockq_get_length(q) < size) return 0; - + for (;;) { 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; + ssize_t k; + int i; if (n > 0) { header[0] = htonl(((uint32_t) 2 << 30) | ((uint32_t) c->payload << 16) | ((uint32_t) c->sequence)); @@ -103,37 +112,38 @@ int pa_rtp_send(pa_rtp_context *c, size_t size, pa_memblockq *q) { iov[0].iov_base = (void*)header; iov[0].iov_len = sizeof(header); - + m.msg_name = NULL; m.msg_namelen = 0; m.msg_iov = iov; - m.msg_iovlen = iov_idx; + m.msg_iovlen = (size_t) iov_idx; m.msg_control = NULL; m.msg_controllen = 0; m.msg_flags = 0; - + 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 += (unsigned) (n/c->frame_size); + if (k < 0) { - if (errno != EAGAIN) /* If the queue is full, just ignore it */ - pa_log(__FILE__": sendmsg() failed: %s", pa_cstrerror(errno)); + if (errno != EAGAIN && errno != EINTR) /* If the queue is full, just ignore it */ + pa_log("sendmsg() failed: %s", pa_cstrerror(errno)); return -1; } - + if (r < 0 || pa_memblockq_get_length(q) < size) break; n = 0; - skip = 0; iov_idx = 1; } } @@ -142,97 +152,143 @@ 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; + + pa_memchunk_reset(&c->memchunk); return c; } -int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_memblock_stat *st) { +int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool, struct timeval *tstamp) { int size; struct msghdr m; + struct cmsghdr *cm; struct iovec iov; uint32_t header; - int cc; + unsigned cc; ssize_t r; - - assert(c); - assert(chunk); + uint8_t aux[1024]; + pa_bool_t found_tstamp = FALSE; + + pa_assert(c); + pa_assert(chunk); - chunk->memblock = NULL; + pa_memchunk_reset(chunk); if (ioctl(c->fd, FIONREAD, &size) < 0) { - pa_log(__FILE__": FIONREAD failed: %s", pa_cstrerror(errno)); + pa_log_warn("FIONREAD failed: %s", pa_cstrerror(errno)); goto fail; } - if (!size) + if (size <= 0) return 0; - chunk->memblock = pa_memblock_new(size, st); + if (c->memchunk.length < (unsigned) size) { + size_t l; + + if (c->memchunk.memblock) + pa_memblock_unref(c->memchunk.memblock); + + l = PA_MAX((size_t) size, pa_mempool_block_size_max(pool)); - iov.iov_base = chunk->memblock->data; - iov.iov_len = size; + c->memchunk.memblock = pa_memblock_new(pool, l); + c->memchunk.index = 0; + c->memchunk.length = pa_memblock_get_length(c->memchunk.memblock); + } + + pa_assert(c->memchunk.length >= (size_t) size); + + chunk->memblock = pa_memblock_ref(c->memchunk.memblock); + chunk->index = c->memchunk.index; + + iov.iov_base = (uint8_t*) pa_memblock_acquire(chunk->memblock) + chunk->index; + iov.iov_len = (size_t) size; m.msg_name = NULL; m.msg_namelen = 0; m.msg_iov = &iov; m.msg_iovlen = 1; - m.msg_control = NULL; - m.msg_controllen = 0; + m.msg_control = aux; + m.msg_controllen = sizeof(aux); m.msg_flags = 0; - - if ((r = recvmsg(c->fd, &m, 0)) != size) { - pa_log(__FILE__": 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(__FILE__": 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(__FILE__": Unsupported RTP version."); + pa_log_warn("Unsupported RTP version."); goto fail; } if ((header >> 29) & 1) { - pa_log(__FILE__": RTP padding not supported."); + pa_log_warn("RTP padding not supported."); goto fail; } if ((header >> 28) & 1) { - pa_log(__FILE__": RTP header extensions not supported."); + pa_log_warn("RTP header extensions not supported."); goto fail; } cc = (header >> 24) & 0xF; - c->payload = (header >> 16) & 127; - c->sequence = header & 0xFFFF; + c->payload = (uint8_t) ((header >> 16) & 127U); + c->sequence = (uint16_t) (header & 0xFFFFU); - if (12 + cc*4 > size) { - pa_log(__FILE__": RTP packet too short. (CSRC)"); + if (12 + cc*4 > (unsigned) size) { + pa_log_warn("RTP packet too short. (CSRC)"); goto fail; } - chunk->index = 12 + cc*4; - chunk->length = size - chunk->index; + chunk->index += 12 + cc*4; + chunk->length = (size_t) size - 12 + cc*4; if (chunk->length % c->frame_size != 0) { - pa_log(__FILE__": Vad RTP packet size."); + pa_log_warn("Bad RTP packet size."); goto fail; } - + + c->memchunk.index = chunk->index + chunk->length; + c->memchunk.length = pa_memblock_get_length(c->memchunk.memblock) - c->memchunk.index; + + if (c->memchunk.length <= 0) { + pa_memblock_unref(c->memchunk.memblock); + pa_memchunk_reset(&c->memchunk); + } + + for (cm = CMSG_FIRSTHDR(&m); cm; cm = CMSG_NXTHDR(&m, cm)) + if (cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SO_TIMESTAMP) { + memcpy(tstamp, CMSG_DATA(cm), sizeof(struct timeval)); + found_tstamp = TRUE; + break; + } + + if (!found_tstamp) { + pa_log_warn("Couldn't find SO_TIMESTAMP data in auxiliary recvmsg() data!"); + memset(tstamp, 0, sizeof(tstamp)); + } + return 0; fail: @@ -243,7 +299,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; @@ -253,12 +309,12 @@ uint8_t pa_rtp_payload_from_sample_spec(const pa_sample_spec *ss) { return 10; if (ss->format == PA_SAMPLE_S16BE && ss->rate == 44100 && ss->channels == 1) return 11; - + return 127; } 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: @@ -278,7 +334,7 @@ pa_sample_spec *pa_rtp_sample_spec_from_payload(uint8_t payload, pa_sample_spec ss->format = PA_SAMPLE_S16BE; ss->rate = 44100; break; - + case 11: ss->channels = 1; ss->format = PA_SAMPLE_S16BE; @@ -293,17 +349,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; @@ -316,9 +372,12 @@ 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); + + pa_assert_se(pa_close(c->fd) == 0); - close(c->fd); + if (c->memchunk.memblock) + pa_memblock_unref(c->memchunk.memblock); } const char* pa_rtp_format_to_string(pa_sample_format_t f) { @@ -337,8 +396,8 @@ 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; else if (!strcmp(s, "L8")) @@ -350,4 +409,3 @@ pa_sample_format_t pa_rtp_string_to_format(const char *s) { else return PA_SAMPLE_INVALID; } - diff --git a/src/modules/rtp/rtp.h b/src/modules/rtp/rtp.h index 35fbbd35..e975e750 100644 --- a/src/modules/rtp/rtp.h +++ b/src/modules/rtp/rtp.h @@ -1,21 +1,21 @@ #ifndef foortphfoo #define foortphfoo -/* $Id$ */ - /*** This file is part of PulseAudio. - + + Copyright 2006 Lennart Poettering + PulseAudio is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published - by the Free Software Foundation; either version 2 of the License, + by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version. - + PulseAudio is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. - + You should have received a copy of the GNU Lesser General Public License along with PulseAudio; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 @@ -35,13 +35,18 @@ typedef struct pa_rtp_context { uint32_t ssrc; uint8_t payload; size_t frame_size; + + pa_memchunk memchunk; } pa_rtp_context; pa_rtp_context* pa_rtp_context_init_send(pa_rtp_context *c, int fd, uint32_t ssrc, uint8_t payload, size_t frame_size); + +/* If the memblockq doesn't have a silence memchunk set, then the caller must + * guarantee that the current read index doesn't point to a hole. */ 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); -int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_memblock_stat *st); +int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool, struct timeval *tstamp); void pa_rtp_context_destroy(pa_rtp_context *c); diff --git a/src/modules/rtp/rtsp_client.c b/src/modules/rtp/rtsp_client.c new file mode 100644 index 00000000..ecf85b89 --- /dev/null +++ b/src/modules/rtp/rtsp_client.c @@ -0,0 +1,530 @@ +/*** + This file is part of PulseAudio. + + Copyright 2008 Colin Guthrie + + PulseAudio is free software; you can redistribute it and/or modify + it under the terms of the GNU Lesser General Public License as published + by the Free Software Foundation; either version 2.1 of the License, + or (at your option) any later version. + + PulseAudio is distributed in the hope that it will be useful, but + WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with PulseAudio; if not, write to the Free Software + Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 + USA. +***/ + +#ifdef HAVE_CONFIG_H +#include <config.h> +#endif + +#include <stdlib.h> +#include <string.h> +#include <errno.h> +#include <unistd.h> +#include <sys/ioctl.h> +#include <netinet/in.h> + +#ifdef HAVE_SYS_FILIO_H +#include <sys/filio.h> +#endif + +#include <pulse/xmalloc.h> + +#include <pulsecore/core-error.h> +#include <pulsecore/core-util.h> +#include <pulsecore/log.h> +#include <pulsecore/macro.h> +#include <pulsecore/strbuf.h> +#include <pulsecore/ioline.h> +#include <pulsecore/arpa-inet.h> + +#include "rtsp_client.h" + +struct pa_rtsp_client { + pa_mainloop_api *mainloop; + char *hostname; + uint16_t port; + + pa_socket_client *sc; + pa_ioline *ioline; + + pa_rtsp_cb_t callback; + + void *userdata; + const char *useragent; + + pa_rtsp_state state; + uint8_t waiting; + + pa_headerlist* headers; + char *last_header; + pa_strbuf *header_buffer; + pa_headerlist* response_headers; + + char *localip; + char *url; + uint16_t rtp_port; + uint32_t cseq; + char *session; + char *transport; +}; + +pa_rtsp_client* pa_rtsp_client_new(pa_mainloop_api *mainloop, const char* hostname, uint16_t port, const char* useragent) { + pa_rtsp_client *c; + + pa_assert(mainloop); + pa_assert(hostname); + pa_assert(port > 0); + + c = pa_xnew0(pa_rtsp_client, 1); + c->mainloop = mainloop; + c->hostname = pa_xstrdup(hostname); + c->port = port; + c->headers = pa_headerlist_new(); + + if (useragent) + c->useragent = useragent; + else + c->useragent = "PulseAudio RTSP Client"; + + return c; +} + + +void pa_rtsp_client_free(pa_rtsp_client* c) { + pa_assert(c); + + if (c->sc) + pa_socket_client_unref(c->sc); + + pa_rtsp_disconnect(c); + + pa_xfree(c->hostname); + pa_xfree(c->url); + pa_xfree(c->localip); + pa_xfree(c->session); + pa_xfree(c->transport); + pa_xfree(c->last_header); + if (c->header_buffer) + pa_strbuf_free(c->header_buffer); + if (c->response_headers) + pa_headerlist_free(c->response_headers); + pa_headerlist_free(c->headers); + + pa_xfree(c); +} + + +static void headers_read(pa_rtsp_client *c) { + char* token; + char delimiters[] = ";"; + + pa_assert(c); + pa_assert(c->response_headers); + pa_assert(c->callback); + + /* Deal with a SETUP response */ + if (STATE_SETUP == c->state) { + const char* token_state = NULL; + const char* pc = NULL; + c->session = pa_xstrdup(pa_headerlist_gets(c->response_headers, "Session")); + c->transport = pa_xstrdup(pa_headerlist_gets(c->response_headers, "Transport")); + + if (!c->session || !c->transport) { + pa_log("Invalid SETUP response."); + return; + } + + /* Now parse out the server port component of the response. */ + while ((token = pa_split(c->transport, delimiters, &token_state))) { + if ((pc = strstr(token, "="))) { + if (0 == strncmp(token, "server_port", 11)) { + pa_atou(pc+1, (uint32_t*)(&c->rtp_port)); + pa_xfree(token); + break; + } + } + pa_xfree(token); + } + if (0 == c->rtp_port) { + /* Error no server_port in response */ + pa_log("Invalid SETUP response (no port number)."); + return; + } + } + + /* Call our callback */ + c->callback(c, c->state, c->response_headers, c->userdata); +} + + +static void line_callback(pa_ioline *line, const char *s, void *userdata) { + char *delimpos; + char *s2, *s2p; + + pa_rtsp_client *c = userdata; + pa_assert(line); + pa_assert(c); + pa_assert(c->callback); + + if (!s) { + /* Keep the ioline/iochannel open as they will be freed automatically */ + c->ioline = NULL; + c->callback(c, STATE_DISCONNECTED, NULL, c->userdata); + return; + } + + s2 = pa_xstrdup(s); + /* Trim trailing carriage returns */ + s2p = s2 + strlen(s2) - 1; + while (s2p >= s2 && '\r' == *s2p) { + *s2p = '\0'; + s2p -= 1; + } + if (c->waiting && 0 == strcmp("RTSP/1.0 200 OK", s2)) { + c->waiting = 0; + if (c->response_headers) + pa_headerlist_free(c->response_headers); + c->response_headers = pa_headerlist_new(); + goto exit; + } + if (c->waiting) { + pa_log_warn("Unexpected response: %s", s2); + goto exit;; + } + if (!strlen(s2)) { + /* End of headers */ + /* We will have a header left from our looping iteration, so add it in :) */ + if (c->last_header) { + char *tmp = pa_strbuf_tostring_free(c->header_buffer); + /* This is not a continuation header so let's dump it into our proplist */ + pa_headerlist_puts(c->response_headers, c->last_header, tmp); + pa_xfree(tmp); + pa_xfree(c->last_header); + c->last_header = NULL; + c->header_buffer = NULL; + } + + pa_log_debug("Full response received. Dispatching"); + headers_read(c); + c->waiting = 1; + goto exit; + } + + /* Read and parse a header (we know it's not empty) */ + /* TODO: Move header reading into the headerlist. */ + + /* If the first character is a space, it's a continuation header */ + if (c->last_header && ' ' == s2[0]) { + pa_assert(c->header_buffer); + + /* Add this line to the buffer (sans the space. */ + pa_strbuf_puts(c->header_buffer, &(s2[1])); + goto exit; + } + + if (c->last_header) { + char *tmp = pa_strbuf_tostring_free(c->header_buffer); + /* This is not a continuation header so let's dump the full + header/value into our proplist */ + pa_headerlist_puts(c->response_headers, c->last_header, tmp); + pa_xfree(tmp); + pa_xfree(c->last_header); + c->last_header = NULL; + c->header_buffer = NULL; + } + + delimpos = strstr(s2, ":"); + if (!delimpos) { + pa_log_warn("Unexpected response when expecting header: %s", s); + goto exit; + } + + pa_assert(!c->header_buffer); + pa_assert(!c->last_header); + + c->header_buffer = pa_strbuf_new(); + if (strlen(delimpos) > 1) { + /* Cut our line off so we can copy the header name out */ + *delimpos++ = '\0'; + + /* Trim the front of any spaces */ + while (' ' == *delimpos) + ++delimpos; + + pa_strbuf_puts(c->header_buffer, delimpos); + } else { + /* Cut our line off so we can copy the header name out */ + *delimpos = '\0'; + } + + /* Save the header name */ + c->last_header = pa_xstrdup(s2); + exit: + pa_xfree(s2); +} + + +static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata) { + pa_rtsp_client *c = userdata; + union { + struct sockaddr sa; + struct sockaddr_in in; + struct sockaddr_in6 in6; + } sa; + socklen_t sa_len = sizeof(sa); + + pa_assert(sc); + pa_assert(c); + pa_assert(STATE_CONNECT == c->state); + pa_assert(c->sc == sc); + pa_socket_client_unref(c->sc); + c->sc = NULL; + + if (!io) { + pa_log("Connection failed: %s", pa_cstrerror(errno)); + return; + } + pa_assert(!c->ioline); + + c->ioline = pa_ioline_new(io); + pa_ioline_set_callback(c->ioline, line_callback, c); + + /* Get the local IP address for use externally */ + if (0 == getsockname(pa_iochannel_get_recv_fd(io), &sa.sa, &sa_len)) { + char buf[INET6_ADDRSTRLEN]; + const char *res = NULL; + + if (AF_INET == sa.sa.sa_family) { + if ((res = inet_ntop(sa.sa.sa_family, &sa.in.sin_addr, buf, sizeof(buf)))) { + c->localip = pa_xstrdup(res); + } + } else if (AF_INET6 == sa.sa.sa_family) { + if ((res = inet_ntop(AF_INET6, &sa.in6.sin6_addr, buf, sizeof(buf)))) { + c->localip = pa_sprintf_malloc("[%s]", res); + } + } + } + pa_log_debug("Established RTSP connection from local ip %s", c->localip); + + if (c->callback) + c->callback(c, c->state, NULL, c->userdata); +} + +int pa_rtsp_connect(pa_rtsp_client *c) { + pa_assert(c); + pa_assert(!c->sc); + + pa_xfree(c->session); + c->session = NULL; + + if (!(c->sc = pa_socket_client_new_string(c->mainloop, TRUE, c->hostname, c->port))) { + pa_log("failed to connect to server '%s:%d'", c->hostname, c->port); + return -1; + } + + pa_socket_client_set_callback(c->sc, on_connection, c); + c->waiting = 1; + c->state = STATE_CONNECT; + return 0; +} + +void pa_rtsp_set_callback(pa_rtsp_client *c, pa_rtsp_cb_t callback, void *userdata) { + pa_assert(c); + + c->callback = callback; + c->userdata = userdata; +} + +void pa_rtsp_disconnect(pa_rtsp_client *c) { + pa_assert(c); + + if (c->ioline) + pa_ioline_close(c->ioline); + c->ioline = NULL; +} + + +const char* pa_rtsp_localip(pa_rtsp_client* c) { + pa_assert(c); + + return c->localip; +} + +uint32_t pa_rtsp_serverport(pa_rtsp_client* c) { + pa_assert(c); + + return c->rtp_port; +} + +void pa_rtsp_set_url(pa_rtsp_client* c, const char* url) { + pa_assert(c); + + c->url = pa_xstrdup(url); +} + +void pa_rtsp_add_header(pa_rtsp_client *c, const char* key, const char* value) { + pa_assert(c); + pa_assert(key); + pa_assert(value); + + pa_headerlist_puts(c->headers, key, value); +} + +void pa_rtsp_remove_header(pa_rtsp_client *c, const char* key) { + pa_assert(c); + pa_assert(key); + + pa_headerlist_remove(c->headers, key); +} + +static int rtsp_exec(pa_rtsp_client* c, const char* cmd, + const char* content_type, const char* content, + int expect_response, + pa_headerlist* headers) { + pa_strbuf* buf; + char* hdrs; + + pa_assert(c); + pa_assert(c->url); + pa_assert(cmd); + pa_assert(c->ioline); + + pa_log_debug("Sending command: %s", cmd); + + buf = pa_strbuf_new(); + pa_strbuf_printf(buf, "%s %s RTSP/1.0\r\nCSeq: %d\r\n", cmd, c->url, ++c->cseq); + if (c->session) + pa_strbuf_printf(buf, "Session: %s\r\n", c->session); + + /* Add the headers */ + if (headers) { + hdrs = pa_headerlist_to_string(headers); + pa_strbuf_puts(buf, hdrs); + pa_xfree(hdrs); + } + + if (content_type && content) { + pa_strbuf_printf(buf, "Content-Type: %s\r\nContent-Length: %d\r\n", + content_type, (int)strlen(content)); + } + + pa_strbuf_printf(buf, "User-Agent: %s\r\n", c->useragent); + + if (c->headers) { + hdrs = pa_headerlist_to_string(c->headers); + pa_strbuf_puts(buf, hdrs); + pa_xfree(hdrs); + } + + pa_strbuf_puts(buf, "\r\n"); + + if (content_type && content) { + pa_strbuf_puts(buf, content); + } + + /* Our packet is created... now we can send it :) */ + hdrs = pa_strbuf_tostring_free(buf); + /*pa_log_debug("Submitting request:"); + pa_log_debug(hdrs);*/ + pa_ioline_puts(c->ioline, hdrs); + pa_xfree(hdrs); + + return 0; +} + + +int pa_rtsp_announce(pa_rtsp_client *c, const char* sdp) { + pa_assert(c); + if (!sdp) + return -1; + + c->state = STATE_ANNOUNCE; + return rtsp_exec(c, "ANNOUNCE", "application/sdp", sdp, 1, NULL); +} + + +int pa_rtsp_setup(pa_rtsp_client* c) { + pa_headerlist* headers; + int rv; + + pa_assert(c); + + headers = pa_headerlist_new(); + pa_headerlist_puts(headers, "Transport", "RTP/AVP/TCP;unicast;interleaved=0-1;mode=record"); + + c->state = STATE_SETUP; + rv = rtsp_exec(c, "SETUP", NULL, NULL, 1, headers); + pa_headerlist_free(headers); + return rv; +} + + +int pa_rtsp_record(pa_rtsp_client* c, uint16_t* seq, uint32_t* rtptime) { + pa_headerlist* headers; + int rv; + char *info; + + pa_assert(c); + if (!c->session) { + /* No session in progress */ + return -1; + } + + /* Todo: Generate these values randomly as per spec */ + *seq = *rtptime = 0; + + headers = pa_headerlist_new(); + pa_headerlist_puts(headers, "Range", "npt=0-"); + info = pa_sprintf_malloc("seq=%u;rtptime=%u", *seq, *rtptime); + pa_headerlist_puts(headers, "RTP-Info", info); + pa_xfree(info); + + c->state = STATE_RECORD; + rv = rtsp_exec(c, "RECORD", NULL, NULL, 1, headers); + pa_headerlist_free(headers); + return rv; +} + + +int pa_rtsp_teardown(pa_rtsp_client *c) { + pa_assert(c); + + c->state = STATE_TEARDOWN; + return rtsp_exec(c, "TEARDOWN", NULL, NULL, 0, NULL); +} + + +int pa_rtsp_setparameter(pa_rtsp_client *c, const char* param) { + pa_assert(c); + if (!param) + return -1; + + c->state = STATE_SET_PARAMETER; + return rtsp_exec(c, "SET_PARAMETER", "text/parameters", param, 1, NULL); +} + + +int pa_rtsp_flush(pa_rtsp_client *c, uint16_t seq, uint32_t rtptime) { + pa_headerlist* headers; + int rv; + char *info; + + pa_assert(c); + + headers = pa_headerlist_new(); + info = pa_sprintf_malloc("seq=%u;rtptime=%u", seq, rtptime); + pa_headerlist_puts(headers, "RTP-Info", info); + pa_xfree(info); + + c->state = STATE_FLUSH; + rv = rtsp_exec(c, "FLUSH", NULL, NULL, 1, headers); + pa_headerlist_free(headers); + return rv; +} diff --git a/src/modules/rtp/rtsp_client.h b/src/modules/rtp/rtsp_client.h new file mode 100644 index 00000000..a56b9324 --- /dev/null +++ b/src/modules/rtp/rtsp_client.h @@ -0,0 +1,71 @@ +#ifndef foortspclienthfoo +#define foortspclienthfoo + +/*** + This file is part of PulseAudio. + + Copyright 2008 Colin Guthrie + + PulseAudio is free software; you can redistribute it and/or modify + it under the terms of the GNU Lesser General Public License as published + by the Free Software Foundation; either version 2.1 of the License, + or (at your option) any later version. + + PulseAudio is distributed in the hope that it will be useful, but + WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with PulseAudio; if not, write to the Free Software + Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 + USA. +***/ + +#include <inttypes.h> +#include <sys/socket.h> +#include <sys/types.h> +#include <netdb.h> + +#include <pulsecore/socket-client.h> +#include <pulse/mainloop-api.h> + +#include "headerlist.h" + +typedef struct pa_rtsp_client pa_rtsp_client; +typedef enum { + STATE_CONNECT, + STATE_ANNOUNCE, + STATE_SETUP, + STATE_RECORD, + STATE_FLUSH, + STATE_TEARDOWN, + STATE_SET_PARAMETER, + STATE_DISCONNECTED +} pa_rtsp_state; +typedef void (*pa_rtsp_cb_t)(pa_rtsp_client *c, pa_rtsp_state state, pa_headerlist* hl, void *userdata); + +pa_rtsp_client* pa_rtsp_client_new(pa_mainloop_api *mainloop, const char* hostname, uint16_t port, const char* useragent); +void pa_rtsp_client_free(pa_rtsp_client* c); + +int pa_rtsp_connect(pa_rtsp_client* c); +void pa_rtsp_set_callback(pa_rtsp_client *c, pa_rtsp_cb_t callback, void *userdata); + +void pa_rtsp_disconnect(pa_rtsp_client* c); + +const char* pa_rtsp_localip(pa_rtsp_client* c); +uint32_t pa_rtsp_serverport(pa_rtsp_client* c); +void pa_rtsp_set_url(pa_rtsp_client* c, const char* url); +void pa_rtsp_add_header(pa_rtsp_client *c, const char* key, const char* value); +void pa_rtsp_remove_header(pa_rtsp_client *c, const char* key); + +int pa_rtsp_announce(pa_rtsp_client* c, const char* sdp); + +int pa_rtsp_setup(pa_rtsp_client* c); +int pa_rtsp_record(pa_rtsp_client* c, uint16_t* seq, uint32_t* rtptime); +int pa_rtsp_teardown(pa_rtsp_client* c); + +int pa_rtsp_setparameter(pa_rtsp_client* c, const char* param); +int pa_rtsp_flush(pa_rtsp_client* c, uint16_t seq, uint32_t rtptime); + +#endif diff --git a/src/modules/rtp/sap.c b/src/modules/rtp/sap.c index 615b8e4e..4d8bf668 100644 --- a/src/modules/rtp/sap.c +++ b/src/modules/rtp/sap.c @@ -1,18 +1,18 @@ -/* $Id$ */ - /*** This file is part of PulseAudio. - + + Copyright 2006 Lennart Poettering + PulseAudio is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published - by the Free Software Foundation; either version 2 of the License, + by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version. - + PulseAudio is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. - + You should have received a copy of the GNU Lesser General Public License along with PulseAudio; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 @@ -23,13 +23,10 @@ #include <config.h> #endif -#include <assert.h> -#include <time.h> #include <stdlib.h> #include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> -#include <arpa/inet.h> #include <errno.h> #include <string.h> #include <unistd.h> @@ -39,11 +36,17 @@ #include <sys/filio.h> #endif +#ifdef HAVE_SYS_UIO_H +#include <sys/uio.h> +#endif + #include <pulse/xmalloc.h> #include <pulsecore/core-error.h> #include <pulsecore/core-util.h> #include <pulsecore/log.h> +#include <pulsecore/macro.h> +#include <pulsecore/arpa-inet.h> #include "sap.h" #include "sdp.h" @@ -51,57 +54,70 @@ #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; c->msg_id_hash = (uint16_t) (rand()*rand()); - - return c; + + return 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); } -int pa_sap_send(pa_sap_context *c, int goodbye) { +int pa_sap_send(pa_sap_context *c, pa_bool_t goodbye) { uint32_t header; struct sockaddr_storage sa_buf; struct sockaddr *sa = (struct sockaddr*) &sa_buf; socklen_t salen = sizeof(sa_buf); struct iovec iov[4]; struct msghdr m; - int k; + ssize_t k; if (getsockname(c->fd, sa, &salen) < 0) { pa_log("getsockname() failed: %s\n", pa_cstrerror(errno)); return -1; } - assert(sa->sa_family == AF_INET || sa->sa_family == AF_INET6); - +#ifdef HAVE_IPV6 + pa_assert(sa->sa_family == AF_INET || sa->sa_family == AF_INET6); +#else + pa_assert(sa->sa_family == AF_INET); +#endif + header = htonl(((uint32_t) 1 << 29) | +#ifdef HAVE_IPV6 (sa->sa_family == AF_INET6 ? (uint32_t) 1 << 28 : 0) | +#endif (goodbye ? (uint32_t) 1 << 26 : 0) | (c->msg_id_hash)); iov[0].iov_base = &header; iov[0].iov_len = sizeof(header); - iov[1].iov_base = sa->sa_family == AF_INET ? (void*) &((struct sockaddr_in*) sa)->sin_addr : (void*) &((struct sockaddr_in6*) sa)->sin6_addr; - iov[1].iov_len = sa->sa_family == AF_INET ? 4 : 16; + if (sa->sa_family == AF_INET) { + iov[1].iov_base = (void*) &((struct sockaddr_in*) sa)->sin_addr; + iov[1].iov_len = 4U; +#ifdef HAVE_IPV6 + } else { + iov[1].iov_base = (void*) &((struct sockaddr_in6*) sa)->sin6_addr; + iov[1].iov_len = 16U; +#endif + } iov[2].iov_base = (char*) MIME_TYPE; iov[2].iov_len = sizeof(MIME_TYPE); iov[3].iov_base = c->sdp_data; iov[3].iov_len = strlen(c->sdp_data); - + m.msg_name = NULL; m.msg_namelen = 0; m.msg_iov = iov; @@ -109,47 +125,44 @@ int pa_sap_send(pa_sap_context *c, int goodbye) { m.msg_control = NULL; m.msg_controllen = 0; 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; + return (int) 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; return c; } -int pa_sap_recv(pa_sap_context *c, int *goodbye) { +int pa_sap_recv(pa_sap_context *c, pa_bool_t *goodbye) { struct msghdr m; struct iovec iov; - int size, k; + int size; char *buf = NULL, *e; uint32_t header; - int six, ac; + unsigned six, ac, k; ssize_t r; - - assert(c); - assert(goodbye); + + pa_assert(c); + pa_assert(goodbye); if (ioctl(c->fd, FIONREAD, &size) < 0) { - pa_log(__FILE__": FIONREAD failed: %s", pa_cstrerror(errno)); + pa_log_warn("FIONREAD failed: %s", pa_cstrerror(errno)); goto fail; } - if (!size) - return 0; - - buf = pa_xnew(char, size+1); + buf = pa_xnew(char, (unsigned) size+1); buf[size] = 0; - + iov.iov_base = buf; - iov.iov_len = size; + iov.iov_len = (size_t) size; m.msg_name = NULL; m.msg_namelen = 0; @@ -158,14 +171,14 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) { m.msg_control = NULL; m.msg_controllen = 0; m.msg_flags = 0; - + if ((r = recvmsg(c->fd, &m, 0)) != size) { - pa_log(__FILE__": 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(__FILE__": SAP packet too short."); + pa_log_warn("SAP packet too short."); goto fail; } @@ -173,48 +186,48 @@ int pa_sap_recv(pa_sap_context *c, int *goodbye) { header = ntohl(header); if (header >> 29 != 1) { - pa_log(__FILE__": Unsupported SAP version."); + pa_log_warn("Unsupported SAP version."); goto fail; } if ((header >> 25) & 1) { - pa_log(__FILE__": Encrypted SAP not supported."); + pa_log_warn("Encrypted SAP not supported."); goto fail; } if ((header >> 24) & 1) { - pa_log(__FILE__": Compressed SAP not supported."); + pa_log_warn("Compressed SAP not supported."); goto fail; } - six = (header >> 28) & 1; - ac = (header >> 16) & 0xFF; + six = (header >> 28) & 1U; + ac = (header >> 16) & 0xFFU; - k = 4 + (six ? 16 : 4) + ac*4; - if (size < k) { - pa_log(__FILE__": SAP packet too short (AD)."); + k = 4 + (six ? 16U : 4U) + ac*4U; + if ((unsigned) size < k) { + pa_log_warn("SAP packet too short (AD)."); goto fail; } e = buf + k; - size -= k; + size -= (int) k; if ((unsigned) size >= sizeof(MIME_TYPE) && !strcmp(e, MIME_TYPE)) { e += sizeof(MIME_TYPE); - size -= sizeof(MIME_TYPE); + size -= (int) sizeof(MIME_TYPE); } else if ((unsigned) size < sizeof(PA_SDP_HEADER)-1 || strncmp(e, PA_SDP_HEADER, sizeof(PA_SDP_HEADER)-1)) { - pa_log(__FILE__": Invalid SDP header."); + pa_log_warn("Invalid SDP header."); goto fail; } if (c->sdp_data) pa_xfree(c->sdp_data); - - c->sdp_data = pa_xstrndup(e, size); + + c->sdp_data = pa_xstrndup(e, (unsigned) size); pa_xfree(buf); - + *goodbye = !!((header >> 26) & 1); - + return 0; fail: diff --git a/src/modules/rtp/sap.h b/src/modules/rtp/sap.h index 987403e3..ae4ad426 100644 --- a/src/modules/rtp/sap.h +++ b/src/modules/rtp/sap.h @@ -1,21 +1,21 @@ #ifndef foosaphfoo #define foosaphfoo -/* $Id$ */ - /*** This file is part of PulseAudio. - + + Copyright 2006 Lennart Poettering + PulseAudio is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published - by the Free Software Foundation; either version 2 of the License, + by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version. - + PulseAudio is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. - + You should have received a copy of the GNU Lesser General Public License along with PulseAudio; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 @@ -38,9 +38,9 @@ typedef struct pa_sap_context { pa_sap_context* pa_sap_context_init_send(pa_sap_context *c, int fd, char *sdp_data); void pa_sap_context_destroy(pa_sap_context *c); -int pa_sap_send(pa_sap_context *c, int goodbye); +int pa_sap_send(pa_sap_context *c, pa_bool_t goodbye); pa_sap_context* pa_sap_context_init_recv(pa_sap_context *c, int fd); -int pa_sap_recv(pa_sap_context *c, int *goodbye); +int pa_sap_recv(pa_sap_context *c, pa_bool_t *goodbye); #endif diff --git a/src/modules/rtp/sdp.c b/src/modules/rtp/sdp.c index e707e4bb..3e61d9b8 100644 --- a/src/modules/rtp/sdp.c +++ b/src/modules/rtp/sdp.c @@ -1,18 +1,18 @@ -/* $Id$ */ - /*** This file is part of PulseAudio. - + + Copyright 2006 Lennart Poettering + PulseAudio is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published - by the Free Software Foundation; either version 2 of the License, + by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version. - + PulseAudio is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. - + You should have received a copy of the GNU Lesser General Public License along with PulseAudio; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 @@ -23,47 +23,48 @@ #include <config.h> #endif -#include <assert.h> #include <time.h> #include <stdlib.h> #include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> -#include <arpa/inet.h> #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 <pulsecore/arpa-inet.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]; - const char *u, *f, *a; - - assert(src); - assert(dst); - assert(af == AF_INET || af == AF_INET6); - - f = pa_rtp_format_to_string(ss->format); - assert(f); - - if (!(u = getenv("USER"))) - if (!(u = getenv("USERNAME"))) - 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); - + char buf_src[64], buf_dst[64], un[64]; + const char *u, *f; + + pa_assert(src); + pa_assert(dst); + +#ifdef HAVE_IPV6 + pa_assert(af == AF_INET || af == AF_INET6); +#else + pa_assert(af == AF_INET); +#endif + + pa_assert_se(f = pa_rtp_format_to_string(ss->format)); + + if (!(u = pa_get_user_name(un, sizeof(un)))) + u = "-"; + + ntp = (uint32_t) time(NULL) + 2208988800U; + + pa_assert_se(inet_ntop(af, src, buf_src, sizeof(buf_src))); + pa_assert_se(inet_ntop(af, dst, buf_dst, sizeof(buf_dst))); + return pa_sprintf_malloc( PA_SDP_HEADER "o=%s %lu 0 IN %s %s\n" @@ -84,8 +85,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; @@ -103,10 +104,10 @@ static pa_sample_spec *parse_sdp_sample_spec(pa_sample_spec *ss, char *c) { return NULL; if (sscanf(c, "%u/%u", &rate, &channels) == 2) { - ss->rate = rate; - ss->channels = channels; + ss->rate = (uint32_t) rate; + ss->channels = (uint8_t) channels; } else if (sscanf(c, "%u", &rate) == 2) { - ss->rate = rate; + ss->rate = (uint32_t) rate; ss->channels = 1; } else return NULL; @@ -119,17 +120,17 @@ static pa_sample_spec *parse_sdp_sample_spec(pa_sample_spec *ss, char *c) { pa_sdp_info *pa_sdp_parse(const char *t, pa_sdp_info *i, int is_goodbye) { uint16_t port = 0; - int ss_valid = 0; + pa_bool_t ss_valid = FALSE; + + pa_assert(t); + pa_assert(i); - assert(t); - assert(i); - i->origin = i->session_name = NULL; i->salen = 0; i->payload = 255; - + if (!pa_startswith(t, PA_SDP_HEADER)) { - pa_log(__FILE__": Failed to parse SDP data: invalid header."); + pa_log("Failed to parse SDP data: invalid header."); goto fail; } @@ -141,7 +142,7 @@ pa_sdp_info *pa_sdp_parse(const char *t, pa_sdp_info *i, int is_goodbye) { l = strcspn(t, "\n"); if (l <= 2) { - pa_log(__FILE__": Failed to parse SDP data: line too short: >%s<.", t); + pa_log("Failed to parse SDP data: line too short: >%s<.", t); goto fail; } @@ -154,49 +155,51 @@ pa_sdp_info *pa_sdp_parse(const char *t, pa_sdp_info *i, int is_goodbye) { size_t k; k = l-8 > sizeof(a) ? sizeof(a) : l-8; - + pa_strlcpy(a, t+9, k); a[strcspn(a, "/")] = 0; if (inet_pton(AF_INET, a, &((struct sockaddr_in*) &i->sa)->sin_addr) <= 0) { - pa_log(__FILE__": Failed to parse SDP data: bad address: >%s<.", a); + pa_log("Failed to parse SDP data: bad address: >%s<.", a); goto fail; } ((struct sockaddr_in*) &i->sa)->sin_family = AF_INET; ((struct sockaddr_in*) &i->sa)->sin_port = 0; i->salen = sizeof(struct sockaddr_in); +#ifdef HAVE_IPV6 } else if (pa_startswith(t, "c=IN IP6 ")) { char a[64]; size_t k; k = l-8 > sizeof(a) ? sizeof(a) : l-8; - + pa_strlcpy(a, t+9, k); a[strcspn(a, "/")] = 0; if (inet_pton(AF_INET6, a, &((struct sockaddr_in6*) &i->sa)->sin6_addr) <= 0) { - pa_log(__FILE__": Failed to parse SDP data: bad address: >%s<.", a); + pa_log("Failed to parse SDP data: bad address: >%s<.", a); goto fail; } ((struct sockaddr_in6*) &i->sa)->sin6_family = AF_INET6; ((struct sockaddr_in6*) &i->sa)->sin6_port = 0; i->salen = sizeof(struct sockaddr_in6); +#endif } else if (pa_startswith(t, "m=audio ")) { if (i->payload > 127) { int _port, _payload; - + if (sscanf(t+8, "%i RTP/AVP %i", &_port, &_payload) == 2) { if (_port <= 0 || _port > 0xFFFF) { - pa_log(__FILE__": Failed to parse SDP data: invalid port %i.", _port); + pa_log("Failed to parse SDP data: invalid port %i.", _port); goto fail; } if (_payload < 0 || _payload > 127) { - pa_log(__FILE__": Failed to parse SDP data: invalid payload %i.", _payload); + pa_log("Failed to parse SDP data: invalid payload %i.", _payload); goto fail; } @@ -204,7 +207,7 @@ pa_sdp_info *pa_sdp_parse(const char *t, pa_sdp_info *i, int is_goodbye) { i->payload = (uint8_t) _payload; if (pa_rtp_sample_spec_from_payload(i->payload, &i->sample_spec)) - ss_valid = 1; + ss_valid = TRUE; } } } else if (pa_startswith(t, "a=rtpmap:")) { @@ -216,28 +219,28 @@ pa_sdp_info *pa_sdp_parse(const char *t, pa_sdp_info *i, int is_goodbye) { if (sscanf(t+9, "%i %64c", &_payload, c) == 2) { if (_payload < 0 || _payload > 127) { - pa_log(__FILE__": Failed to parse SDP data: invalid payload %i.", _payload); + pa_log("Failed to parse SDP data: invalid payload %i.", _payload); goto fail; } if (_payload == i->payload) { c[strcspn(c, "\n")] = 0; - + if (parse_sdp_sample_spec(&i->sample_spec, c)) - ss_valid = 1; + ss_valid = TRUE; } } } } - + t += l; - + if (*t == '\n') t++; } if (!i->origin || (!is_goodbye && (!i->salen || i->payload > 127 || !ss_valid || port == 0))) { - pa_log(__FILE__": Failed to parse SDP data: missing data."); + pa_log("Failed to parse SDP data: missing data."); goto fail; } @@ -245,7 +248,7 @@ pa_sdp_info *pa_sdp_parse(const char *t, pa_sdp_info *i, int is_goodbye) { ((struct sockaddr_in*) &i->sa)->sin_port = htons(port); else ((struct sockaddr_in6*) &i->sa)->sin6_port = htons(port); - + return i; fail: @@ -256,7 +259,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); diff --git a/src/modules/rtp/sdp.h b/src/modules/rtp/sdp.h index b95ca633..4cb3b203 100644 --- a/src/modules/rtp/sdp.h +++ b/src/modules/rtp/sdp.h @@ -1,21 +1,21 @@ #ifndef foosdphfoo #define foosdphfoo -/* $Id$ */ - /*** This file is part of PulseAudio. - + + Copyright 2006 Lennart Poettering + PulseAudio is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published - by the Free Software Foundation; either version 2 of the License, + by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version. - + PulseAudio is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. - + You should have received a copy of the GNU Lesser General Public License along with PulseAudio; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 |