code.delx.au - pulseaudio/blob - src/modules/rtp/module-rtp-recv.c

   1
   2 /***
   3   This file is part of polypaudio.
   4
   5   polypaudio is free software; you can redistribute it and/or modify
   6   it under the terms of the GNU Lesser General Public License as published
   7   by the Free Software Foundation; either version 2 of the License,
   8   or (at your option) any later version.
   9
  10   polypaudio is distributed in the hope that it will be useful, but
  11   WITHOUT ANY WARRANTY; without even the implied warranty of
  12   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  13   General Public License for more details.
  14
  15   You should have received a copy of the GNU Lesser General Public License
  16   along with polypaudio; if not, write to the Free Software
  17   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  18   USA.
  19 ***/
  20
  21 #ifdef HAVE_CONFIG_H
  22 #include <config.h>
  23 #endif
  24
  25 #include <assert.h>
  26 #include <stdio.h>
  27 #include <sys/socket.h>
  28 #include <netinet/in.h>
  29 #include <arpa/inet.h>
  30 #include <errno.h>
  31 #include <string.h>
  32 #include <unistd.h>
  33
  34 #include <polypcore/module.h>
  35 #include <polypcore/llist.h>
  36 #include <polypcore/sink.h>
  37 #include <polypcore/sink-input.h>
  38 #include <polypcore/memblockq.h>
  39 #include <polypcore/log.h>
  40 #include <polypcore/util.h>
  41 #include <polypcore/xmalloc.h>
  42 #include <polypcore/modargs.h>
  43 #include <polypcore/namereg.h>
  44
  45 #include "module-rtp-recv-symdef.h"
  46
  47 #include "rtp.h"
  48 #include "sdp.h"
  49 #include "sap.h"
  50
  51 PA_MODULE_AUTHOR("Lennart Poettering")
  52 PA_MODULE_DESCRIPTION("Recieve data from a network via RTP/SAP/SDP")
  53 PA_MODULE_VERSION(PACKAGE_VERSION)
  54 PA_MODULE_USAGE(
  55         "sink=<name of the sink> "
  56         "sap_address=<multicast address to listen on> "
  57 )
  58
  59 #define SAP_PORT 9875
  60 #define DEFAULT_SAP_ADDRESS "224.0.1.2"
  61 #define MEMBLOCKQ_MAXLENGTH (1024*170)
  62 #define MAX_SESSIONS 16
  63 #define DEATH_TIMEOUT 20000000
  64
  65 static const char* const valid_modargs[] = {
  66     "sink",
  67     "sap_address",
  68     NULL
  69 };
  70
  71 struct session {
  72     struct userdata *userdata;
  73
  74     pa_sink_input *sink_input;
  75     pa_memblockq *memblockq;
  76
  77     pa_time_event *death_event;
  78
  79     int first_packet;
  80     uint32_t ssrc;
  81     uint32_t offset;
  82
  83     struct pa_sdp_info sdp_info;
  84
  85     pa_rtp_context rtp_context;
  86     pa_io_event* rtp_event;
  87 };
  88
  89 struct userdata {
  90     pa_module *module;
  91     pa_core *core;
  92
  93     pa_sap_context sap_context;
  94     pa_io_event* sap_event;
  95
  96     pa_hashmap *by_origin;
  97
  98     char *sink_name;
  99 };
 100
 101 static void session_free(struct session *s, int from_hash);
 102
 103 static int sink_input_peek(pa_sink_input *i, pa_memchunk *chunk) {
 104     struct session *s;
 105     assert(i);
 106     s = i->userdata;
 107
 108     return pa_memblockq_peek(s->memblockq, chunk);
 109 }
 110
 111 static void sink_input_drop(pa_sink_input *i, const pa_memchunk *chunk, size_t length) {
 112     struct session *s;
 113     assert(i);
 114     s = i->userdata;
 115
 116     pa_memblockq_drop(s->memblockq, chunk, length);
 117 }
 118
 119 static void sink_input_kill(pa_sink_input* i) {
 120     struct session *s;
 121     assert(i);
 122     s = i->userdata;
 123
 124     session_free(s, 1);
 125 }
 126
 127 static pa_usec_t sink_input_get_latency(pa_sink_input *i) {
 128     struct session *s;
 129     assert(i);
 130     s = i->userdata;
 131
 132     return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &i->sample_spec);
 133 }
 134
 135 static void rtp_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event_flags_t flags, void *userdata) {
 136     struct session *s = userdata;
 137     pa_memchunk chunk;
 138     int64_t k, j, delta;
 139     struct timeval tv;
 140
 141     assert(m);
 142     assert(e);
 143     assert(s);
 144     assert(fd == s->rtp_context.fd);
 145     assert(flags == PA_IO_EVENT_INPUT);
 146
 147     if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->core->memblock_stat) < 0)
 148         return;
 149
 150     if (s->sdp_info.payload != s->rtp_context.payload) {
 151         pa_memblock_unref(chunk.memblock);
 152         return;
 153     }
 154
 155     if (!s->first_packet) {
 156         s->first_packet = 1;
 157
 158         s->ssrc = s->rtp_context.ssrc;
 159         s->offset = s->rtp_context.timestamp;
 160     } else {
 161         if (s->ssrc != s->rtp_context.ssrc) {
 162             pa_memblock_unref(chunk.memblock);
 163             return;
 164         }
 165     }
 166
 167     /* Check wheter there was a timestamp overflow */
 168     k = (int64_t) s->rtp_context.timestamp - (int64_t) s->offset;
 169     j = (int64_t) 0x100000000 - (int64_t) s->offset + (int64_t) s->rtp_context.timestamp;
 170
 171     if ((k < 0 ? -k : k) < (j < 0 ? -j : j))
 172         delta = k;
 173     else
 174         delta = j;
 175
 176     pa_memblockq_seek(s->memblockq, delta * s->rtp_context.frame_size, PA_SEEK_RELATIVE);
 177     pa_memblockq_push(s->memblockq, &chunk);
 178
 179     /* The next timestamp we expect */
 180     s->offset = s->rtp_context.timestamp + (chunk.length / s->rtp_context.frame_size);
 181
 182     pa_memblock_unref(chunk.memblock);
 183
 184     /* Reset death timer */
 185     pa_gettimeofday(&tv);
 186     pa_timeval_add(&tv, DEATH_TIMEOUT);
 187     m->time_restart(s->death_event, &tv);
 188 }
 189
 190 static void death_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *tv, void *userdata) {
 191     struct session *s = userdata;
 192
 193     assert(m);
 194     assert(t);
 195     assert(tv);
 196     assert(s);
 197
 198     session_free(s, 1);
 199 }
 200
 201 static int mcast_socket(const struct sockaddr* sa, socklen_t salen) {
 202     int af, fd = -1, r, one;
 203
 204     af = sa->sa_family;
 205     if ((fd = socket(af, SOCK_DGRAM, 0)) < 0) {
 206         pa_log(__FILE__": Failed to create socket: %s", strerror(errno));
 207         goto fail;
 208     }
 209
 210     one = 1;
 211     if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0) {
 212         pa_log(__FILE__": SO_REUSEADDR failed: %s", strerror(errno));
 213         goto fail;
 214     }
 215
 216     if (af == AF_INET) {
 217         struct ip_mreq mr4;
 218         memset(&mr4, 0, sizeof(mr4));
 219         mr4.imr_multiaddr = ((const struct sockaddr_in*) sa)->sin_addr;
 220         r = setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mr4, sizeof(mr4));
 221     } else {
 222         struct ipv6_mreq mr6;
 223         memset(&mr6, 0, sizeof(mr6));
 224         mr6.ipv6mr_multiaddr = ((const struct sockaddr_in6*) sa)->sin6_addr;
 225         r = setsockopt(fd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &mr6, sizeof(mr6));
 226     }
 227
 228     if (r < 0) {
 229         pa_log_info(__FILE__": Joining mcast group failed: %s", strerror(errno));
 230         goto fail;
 231     }
 232
 233     if (bind(fd, sa, salen) < 0) {
 234         pa_log(__FILE__": bind() failed: %s", strerror(errno));
 235         goto fail;
 236     }
 237
 238     return fd;
 239
 240 fail:
 241     if (fd >= 0)
 242         close(fd);
 243
 244     return -1;
 245 }
 246
 247 static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_info) {
 248     struct session *s = NULL;
 249     struct timeval tv;
 250     char *c;
 251     pa_sink *sink;
 252     int fd = -1;
 253
 254     if (!(sink = pa_namereg_get(u->core, u->sink_name, PA_NAMEREG_SINK, 1))) {
 255         pa_log(__FILE__": sink does not exist.");
 256         goto fail;
 257     }
 258
 259     s = pa_xnew0(struct session, 1);
 260     s->userdata = u;
 261     s->first_packet = 0;
 262     s->sdp_info = *sdp_info;
 263
 264     if ((fd = mcast_socket((const struct sockaddr*) &sdp_info->sa, sdp_info->salen)) < 0)
 265         goto fail;
 266
 267     c = pa_sprintf_malloc("RTP Stream%s%s%s",
 268                           sdp_info->session_name ? " (" : "",
 269                           sdp_info->session_name ? sdp_info->session_name : "",
 270                           sdp_info->session_name ? ")" : "");
 271
 272     s->sink_input = pa_sink_input_new(sink, __FILE__, c, &sdp_info->sample_spec, NULL, 0, PA_RESAMPLER_INVALID);
 273     pa_xfree(c);
 274
 275     if (!s->sink_input) {
 276         pa_log(__FILE__": failed to create sink input.");
 277         goto fail;
 278     }
 279
 280     s->sink_input->userdata = s;
 281     s->sink_input->owner = u->module;
 282
 283     s->sink_input->peek = sink_input_peek;
 284     s->sink_input->drop = sink_input_drop;
 285     s->sink_input->kill = sink_input_kill;
 286     s->sink_input->get_latency = sink_input_get_latency;
 287
 288     s->memblockq = pa_memblockq_new(
 289             0,
 290             MEMBLOCKQ_MAXLENGTH,
 291             MEMBLOCKQ_MAXLENGTH,
 292             pa_frame_size(&s->sink_input->sample_spec),
 293             1,
 294             0,
 295             NULL,
 296             u->core->memblock_stat);
 297
 298     s->rtp_event = u->core->mainloop->io_new(u->core->mainloop, fd, PA_IO_EVENT_INPUT, rtp_event_cb, s);
 299
 300     pa_gettimeofday(&tv);
 301     pa_timeval_add(&tv, DEATH_TIMEOUT);
 302     s->death_event = u->core->mainloop->time_new(u->core->mainloop, &tv, death_event_cb, s);
 303
 304     pa_hashmap_put(s->userdata->by_origin, s->sdp_info.origin, s);
 305
 306     pa_rtp_context_init_recv(&s->rtp_context, fd, pa_frame_size(&s->sdp_info.sample_spec));
 307
 308     pa_log_info(__FILE__": Found new session '%s'", s->sdp_info.session_name);
 309
 310     return s;
 311
 312 fail:
 313     if (s) {
 314         if (fd >= 0)
 315             close(fd);
 316
 317         pa_xfree(s);
 318     }
 319
 320     return NULL;
 321 }
 322
 323 static void session_free(struct session *s, int from_hash) {
 324     assert(s);
 325
 326     pa_log_info(__FILE__": Freeing session '%s'", s->sdp_info.session_name);
 327
 328     s->userdata->core->mainloop->time_free(s->death_event);
 329     s->userdata->core->mainloop->io_free(s->rtp_event);
 330
 331     if (from_hash)
 332         pa_hashmap_remove(s->userdata->by_origin, s->sdp_info.origin);
 333
 334     pa_sink_input_disconnect(s->sink_input);
 335     pa_sink_input_unref(s->sink_input);
 336
 337     pa_memblockq_free(s->memblockq);
 338     pa_sdp_info_destroy(&s->sdp_info);
 339     pa_rtp_context_destroy(&s->rtp_context);
 340
 341     pa_xfree(s);
 342 }
 343
 344 static void sap_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event_flags_t flags, void *userdata) {
 345     struct userdata *u = userdata;
 346     int goodbye;
 347     pa_sdp_info info;
 348     struct session *s;
 349
 350     assert(m);
 351     assert(e);
 352     assert(u);
 353     assert(fd == u->sap_context.fd);
 354     assert(flags == PA_IO_EVENT_INPUT);
 355
 356     if (pa_sap_recv(&u->sap_context, &goodbye) < 0)
 357         return;
 358
 359     if (!pa_sdp_parse(u->sap_context.sdp_data, &info, goodbye))
 360         return;
 361
 362     if (goodbye) {
 363
 364         if ((s = pa_hashmap_get(u->by_origin, info.origin)))
 365             session_free(s, 1);
 366
 367         pa_sdp_info_destroy(&info);
 368     } else {
 369
 370         if (!(s = pa_hashmap_get(u->by_origin, info.origin))) {
 371             if (!(s = session_new(u, &info)))
 372                 pa_sdp_info_destroy(&info);
 373
 374         } else {
 375             struct timeval tv;
 376
 377             pa_gettimeofday(&tv);
 378             pa_timeval_add(&tv, DEATH_TIMEOUT);
 379             m->time_restart(s->death_event, &tv);
 380
 381             pa_sdp_info_destroy(&info);
 382         }
 383     }
 384 }
 385
 386 int pa__init(pa_core *c, pa_module*m) {
 387     struct userdata *u;
 388     pa_modargs *ma = NULL;
 389     struct sockaddr_in sa4;
 390     struct sockaddr_in6 sa6;
 391     struct sockaddr *sa;
 392     socklen_t salen;
 393     const char *sap_address;
 394     int fd = -1;
 395
 396     assert(c);
 397     assert(m);
 398
 399     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
 400         pa_log(__FILE__": failed to parse module arguments");
 401         goto fail;
 402     }
 403
 404     sap_address = pa_modargs_get_value(ma, "sap_address", DEFAULT_SAP_ADDRESS);
 405
 406     if (inet_pton(AF_INET6, sap_address, &sa6.sin6_addr) > 0) {
 407         sa6.sin6_family = AF_INET6;
 408         sa6.sin6_port = htons(SAP_PORT);
 409         sa = (struct sockaddr*) &sa6;
 410         salen = sizeof(sa6);
 411     } else if (inet_pton(AF_INET, sap_address, &sa4.sin_addr) > 0) {
 412         sa4.sin_family = AF_INET;
 413         sa4.sin_port = htons(SAP_PORT);
 414         sa = (struct sockaddr*) &sa4;
 415         salen = sizeof(sa4);
 416     } else {
 417         pa_log(__FILE__": invalid SAP address '%s'", sap_address);
 418         goto fail;
 419     }
 420
 421     if ((fd = mcast_socket(sa, salen)) < 0)
 422         goto fail;
 423
 424     u = pa_xnew(struct userdata, 1);
 425     m->userdata = u;
 426     u->module = m;
 427     u->core = c;
 428     u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
 429
 430     u->sap_event = c->mainloop->io_new(c->mainloop, fd, PA_IO_EVENT_INPUT, sap_event_cb, u);
 431
 432     u->by_origin = pa_hashmap_new(pa_idxset_string_hash_func, pa_idxset_string_compare_func);
 433
 434     pa_sap_context_init_recv(&u->sap_context, fd);
 435
 436     pa_modargs_free(ma);
 437
 438     return 0;
 439
 440 fail:
 441     if (ma)
 442         pa_modargs_free(ma);
 443
 444     if (fd >= 0)
 445         close(fd);
 446
 447     return -1;
 448 }
 449
 450 static void free_func(void *p, void *userdata) {
 451     session_free(p, 0);
 452 }
 453
 454 void pa__done(pa_core *c, pa_module*m) {
 455     struct userdata *u;
 456     assert(c);
 457     assert(m);
 458
 459     if (!(u = m->userdata))
 460         return;
 461
 462     c->mainloop->io_free(u->sap_event);
 463     pa_sap_context_destroy(&u->sap_context);
 464
 465     pa_hashmap_free(u->by_origin, free_func, NULL);
 466
 467     pa_xfree(u->sink_name);
 468     pa_xfree(u);
 469 }