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 #include <polypcore/sample-util.h>
  45
  46 #include "module-rtp-recv-symdef.h"
  47
  48 #include "rtp.h"
  49 #include "sdp.h"
  50 #include "sap.h"
  51
  52 PA_MODULE_AUTHOR("Lennart Poettering")
  53 PA_MODULE_DESCRIPTION("Recieve data from a network via RTP/SAP/SDP")
  54 PA_MODULE_VERSION(PACKAGE_VERSION)
  55 PA_MODULE_USAGE(
  56         "sink=<name of the sink> "
  57         "sap_address=<multicast address to listen on> "
  58 )
  59
  60 #define SAP_PORT 9875
  61 #define DEFAULT_SAP_ADDRESS "224.0.0.56"
  62 #define MEMBLOCKQ_MAXLENGTH (1024*170)
  63 #define MAX_SESSIONS 16
  64 #define DEATH_TIMEOUT 20000000
  65
  66 static const char* const valid_modargs[] = {
  67     "sink",
  68     "sap_address",
  69     NULL
  70 };
  71
  72 struct session {
  73     struct userdata *userdata;
  74
  75     pa_sink_input *sink_input;
  76     pa_memblockq *memblockq;
  77
  78     pa_time_event *death_event;
  79
  80     int first_packet;
  81     uint32_t ssrc;
  82     uint32_t offset;
  83
  84     struct pa_sdp_info sdp_info;
  85
  86     pa_rtp_context rtp_context;
  87     pa_io_event* rtp_event;
  88 };
  89
  90 struct userdata {
  91     pa_module *module;
  92     pa_core *core;
  93
  94     pa_sap_context sap_context;
  95     pa_io_event* sap_event;
  96
  97     pa_hashmap *by_origin;
  98
  99     char *sink_name;
 100
 101     int n_sessions;
 102 };
 103
 104 static void session_free(struct session *s, int from_hash);
 105
 106 static int sink_input_peek(pa_sink_input *i, pa_memchunk *chunk) {
 107     struct session *s;
 108     assert(i);
 109     s = i->userdata;
 110
 111     return pa_memblockq_peek(s->memblockq, chunk);
 112 }
 113
 114 static void sink_input_drop(pa_sink_input *i, const pa_memchunk *chunk, size_t length) {
 115     struct session *s;
 116     assert(i);
 117     s = i->userdata;
 118
 119     pa_memblockq_drop(s->memblockq, chunk, length);
 120 }
 121
 122 static void sink_input_kill(pa_sink_input* i) {
 123     struct session *s;
 124     assert(i);
 125     s = i->userdata;
 126
 127     session_free(s, 1);
 128 }
 129
 130 static pa_usec_t sink_input_get_latency(pa_sink_input *i) {
 131     struct session *s;
 132     assert(i);
 133     s = i->userdata;
 134
 135     return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &i->sample_spec);
 136 }
 137
 138 static void rtp_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event_flags_t flags, void *userdata) {
 139     struct session *s = userdata;
 140     pa_memchunk chunk;
 141     int64_t k, j, delta;
 142     struct timeval tv;
 143
 144     assert(m);
 145     assert(e);
 146     assert(s);
 147     assert(fd == s->rtp_context.fd);
 148     assert(flags == PA_IO_EVENT_INPUT);
 149
 150     if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->core->memblock_stat) < 0)
 151         return;
 152
 153     if (s->sdp_info.payload != s->rtp_context.payload) {
 154         pa_memblock_unref(chunk.memblock);
 155         return;
 156     }
 157
 158     if (!s->first_packet) {
 159         s->first_packet = 1;
 160
 161         s->ssrc = s->rtp_context.ssrc;
 162         s->offset = s->rtp_context.timestamp;
 163
 164         if (s->ssrc == s->userdata->core->cookie)
 165             pa_log_warn(__FILE__": WARNING! Detected RTP packet loop!");
 166     } else {
 167         if (s->ssrc != s->rtp_context.ssrc) {
 168             pa_memblock_unref(chunk.memblock);
 169             return;
 170         }
 171     }
 172
 173     /* Check wheter there was a timestamp overflow */
 174     k = (int64_t) s->rtp_context.timestamp - (int64_t) s->offset;
 175     j = (int64_t) 0x100000000LL - (int64_t) s->offset + (int64_t) s->rtp_context.timestamp;
 176
 177     if ((k < 0 ? -k : k) < (j < 0 ? -j : j))
 178         delta = k;
 179     else
 180         delta = j;
 181
 182     pa_memblockq_seek(s->memblockq, delta * s->rtp_context.frame_size, PA_SEEK_RELATIVE);
 183
 184     if (pa_memblockq_push(s->memblockq, &chunk) < 0) {
 185         /* queue overflow, let's flush it and try again */
 186         pa_memblockq_flush(s->memblockq);
 187         pa_memblockq_push(s->memblockq, &chunk);
 188     }
 189
 190     /* The next timestamp we expect */
 191     s->offset = s->rtp_context.timestamp + (chunk.length / s->rtp_context.frame_size);
 192
 193     pa_memblock_unref(chunk.memblock);
 194
 195     /* Reset death timer */
 196     pa_gettimeofday(&tv);
 197     pa_timeval_add(&tv, DEATH_TIMEOUT);
 198     m->time_restart(s->death_event, &tv);
 199 }
 200
 201 static void death_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *tv, void *userdata) {
 202     struct session *s = userdata;
 203
 204     assert(m);
 205     assert(t);
 206     assert(tv);
 207     assert(s);
 208
 209     session_free(s, 1);
 210 }
 211
 212 static int mcast_socket(const struct sockaddr* sa, socklen_t salen) {
 213     int af, fd = -1, r, one;
 214
 215     af = sa->sa_family;
 216     if ((fd = socket(af, SOCK_DGRAM, 0)) < 0) {
 217         pa_log(__FILE__": Failed to create socket: %s", strerror(errno));
 218         goto fail;
 219     }
 220
 221     one = 1;
 222     if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0) {
 223         pa_log(__FILE__": SO_REUSEADDR failed: %s", strerror(errno));
 224         goto fail;
 225     }
 226
 227     if (af == AF_INET) {
 228         struct ip_mreq mr4;
 229         memset(&mr4, 0, sizeof(mr4));
 230         mr4.imr_multiaddr = ((const struct sockaddr_in*) sa)->sin_addr;
 231         r = setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mr4, sizeof(mr4));
 232     } else {
 233         struct ipv6_mreq mr6;
 234         memset(&mr6, 0, sizeof(mr6));
 235         mr6.ipv6mr_multiaddr = ((const struct sockaddr_in6*) sa)->sin6_addr;
 236         r = setsockopt(fd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &mr6, sizeof(mr6));
 237     }
 238
 239     if (r < 0) {
 240         pa_log_info(__FILE__": Joining mcast group failed: %s", strerror(errno));
 241         goto fail;
 242     }
 243
 244     if (bind(fd, sa, salen) < 0) {
 245         pa_log(__FILE__": bind() failed: %s", strerror(errno));
 246         goto fail;
 247     }
 248
 249     return fd;
 250
 251 fail:
 252     if (fd >= 0)
 253         close(fd);
 254
 255     return -1;
 256 }
 257
 258 static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_info) {
 259     struct session *s = NULL;
 260     struct timeval tv;
 261     char *c;
 262     pa_sink *sink;
 263     int fd = -1;
 264     pa_memblock *silence;
 265
 266     if (u->n_sessions >= MAX_SESSIONS) {
 267         pa_log(__FILE__": session limit reached.");
 268         goto fail;
 269     }
 270
 271     if (!(sink = pa_namereg_get(u->core, u->sink_name, PA_NAMEREG_SINK, 1))) {
 272         pa_log(__FILE__": sink does not exist.");
 273         goto fail;
 274     }
 275
 276     s = pa_xnew0(struct session, 1);
 277     s->userdata = u;
 278     s->first_packet = 0;
 279     s->sdp_info = *sdp_info;
 280
 281     if ((fd = mcast_socket((const struct sockaddr*) &sdp_info->sa, sdp_info->salen)) < 0)
 282         goto fail;
 283
 284     c = pa_sprintf_malloc("RTP Stream%s%s%s",
 285                           sdp_info->session_name ? " (" : "",
 286                           sdp_info->session_name ? sdp_info->session_name : "",
 287                           sdp_info->session_name ? ")" : "");
 288
 289     s->sink_input = pa_sink_input_new(sink, __FILE__, c, &sdp_info->sample_spec, NULL, 0, PA_RESAMPLER_INVALID);
 290     pa_xfree(c);
 291
 292     if (!s->sink_input) {
 293         pa_log(__FILE__": failed to create sink input.");
 294         goto fail;
 295     }
 296
 297     s->sink_input->userdata = s;
 298     s->sink_input->owner = u->module;
 299
 300     s->sink_input->peek = sink_input_peek;
 301     s->sink_input->drop = sink_input_drop;
 302     s->sink_input->kill = sink_input_kill;
 303     s->sink_input->get_latency = sink_input_get_latency;
 304
 305     silence = pa_silence_memblock_new(&s->sink_input->sample_spec,
 306                                       (pa_bytes_per_second(&s->sink_input->sample_spec)/128/pa_frame_size(&s->sink_input->sample_spec))*
 307                                       pa_frame_size(&s->sink_input->sample_spec),
 308                                       s->userdata->core->memblock_stat);
 309
 310     s->memblockq = pa_memblockq_new(
 311             0,
 312             MEMBLOCKQ_MAXLENGTH,
 313             MEMBLOCKQ_MAXLENGTH,
 314             pa_frame_size(&s->sink_input->sample_spec),
 315             pa_bytes_per_second(&s->sink_input->sample_spec)/10+1,
 316             0,
 317             silence,
 318             u->core->memblock_stat);
 319
 320     pa_memblock_unref(silence);
 321
 322     s->rtp_event = u->core->mainloop->io_new(u->core->mainloop, fd, PA_IO_EVENT_INPUT, rtp_event_cb, s);
 323
 324     pa_gettimeofday(&tv);
 325     pa_timeval_add(&tv, DEATH_TIMEOUT);
 326     s->death_event = u->core->mainloop->time_new(u->core->mainloop, &tv, death_event_cb, s);
 327
 328     pa_hashmap_put(s->userdata->by_origin, s->sdp_info.origin, s);
 329
 330     pa_rtp_context_init_recv(&s->rtp_context, fd, pa_frame_size(&s->sdp_info.sample_spec));
 331
 332     pa_log_info(__FILE__": Found new session '%s'", s->sdp_info.session_name);
 333
 334     u->n_sessions++;
 335
 336     return s;
 337
 338 fail:
 339     if (s) {
 340         if (fd >= 0)
 341             close(fd);
 342
 343         pa_xfree(s);
 344     }
 345
 346     return NULL;
 347 }
 348
 349 static void session_free(struct session *s, int from_hash) {
 350     assert(s);
 351
 352     pa_log_info(__FILE__": Freeing session '%s'", s->sdp_info.session_name);
 353
 354     s->userdata->core->mainloop->time_free(s->death_event);
 355     s->userdata->core->mainloop->io_free(s->rtp_event);
 356
 357     if (from_hash)
 358         pa_hashmap_remove(s->userdata->by_origin, s->sdp_info.origin);
 359
 360     pa_sink_input_disconnect(s->sink_input);
 361     pa_sink_input_unref(s->sink_input);
 362
 363     pa_memblockq_free(s->memblockq);
 364     pa_sdp_info_destroy(&s->sdp_info);
 365     pa_rtp_context_destroy(&s->rtp_context);
 366
 367     assert(s->userdata->n_sessions >= 1);
 368     s->userdata->n_sessions--;
 369
 370     pa_xfree(s);
 371 }
 372
 373 static void sap_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event_flags_t flags, void *userdata) {
 374     struct userdata *u = userdata;
 375     int goodbye;
 376     pa_sdp_info info;
 377     struct session *s;
 378
 379     assert(m);
 380     assert(e);
 381     assert(u);
 382     assert(fd == u->sap_context.fd);
 383     assert(flags == PA_IO_EVENT_INPUT);
 384
 385     if (pa_sap_recv(&u->sap_context, &goodbye) < 0)
 386         return;
 387
 388     if (!pa_sdp_parse(u->sap_context.sdp_data, &info, goodbye))
 389         return;
 390
 391     if (goodbye) {
 392
 393         if ((s = pa_hashmap_get(u->by_origin, info.origin)))
 394             session_free(s, 1);
 395
 396         pa_sdp_info_destroy(&info);
 397     } else {
 398
 399         if (!(s = pa_hashmap_get(u->by_origin, info.origin))) {
 400             if (!(s = session_new(u, &info)))
 401                 pa_sdp_info_destroy(&info);
 402
 403         } else {
 404             struct timeval tv;
 405
 406             pa_gettimeofday(&tv);
 407             pa_timeval_add(&tv, DEATH_TIMEOUT);
 408             m->time_restart(s->death_event, &tv);
 409
 410             pa_sdp_info_destroy(&info);
 411         }
 412     }
 413 }
 414
 415 int pa__init(pa_core *c, pa_module*m) {
 416     struct userdata *u;
 417     pa_modargs *ma = NULL;
 418     struct sockaddr_in sa4;
 419     struct sockaddr_in6 sa6;
 420     struct sockaddr *sa;
 421     socklen_t salen;
 422     const char *sap_address;
 423     int fd = -1;
 424
 425     assert(c);
 426     assert(m);
 427
 428     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
 429         pa_log(__FILE__": failed to parse module arguments");
 430         goto fail;
 431     }
 432
 433     sap_address = pa_modargs_get_value(ma, "sap_address", DEFAULT_SAP_ADDRESS);
 434
 435     if (inet_pton(AF_INET6, sap_address, &sa6.sin6_addr) > 0) {
 436         sa6.sin6_family = AF_INET6;
 437         sa6.sin6_port = htons(SAP_PORT);
 438         sa = (struct sockaddr*) &sa6;
 439         salen = sizeof(sa6);
 440     } else if (inet_pton(AF_INET, sap_address, &sa4.sin_addr) > 0) {
 441         sa4.sin_family = AF_INET;
 442         sa4.sin_port = htons(SAP_PORT);
 443         sa = (struct sockaddr*) &sa4;
 444         salen = sizeof(sa4);
 445     } else {
 446         pa_log(__FILE__": invalid SAP address '%s'", sap_address);
 447         goto fail;
 448     }
 449
 450     if ((fd = mcast_socket(sa, salen)) < 0)
 451         goto fail;
 452
 453     u = pa_xnew(struct userdata, 1);
 454     m->userdata = u;
 455     u->module = m;
 456     u->core = c;
 457     u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
 458     u->n_sessions = 0;
 459
 460     u->sap_event = c->mainloop->io_new(c->mainloop, fd, PA_IO_EVENT_INPUT, sap_event_cb, u);
 461
 462     u->by_origin = pa_hashmap_new(pa_idxset_string_hash_func, pa_idxset_string_compare_func);
 463
 464     pa_sap_context_init_recv(&u->sap_context, fd);
 465
 466     pa_modargs_free(ma);
 467
 468     return 0;
 469
 470 fail:
 471     if (ma)
 472         pa_modargs_free(ma);
 473
 474     if (fd >= 0)
 475         close(fd);
 476
 477     return -1;
 478 }
 479
 480 static void free_func(void *p, PA_GCC_UNUSED void *userdata) {
 481     session_free(p, 0);
 482 }
 483
 484 void pa__done(pa_core *c, pa_module*m) {
 485     struct userdata *u;
 486     assert(c);
 487     assert(m);
 488
 489     if (!(u = m->userdata))
 490         return;
 491
 492     c->mainloop->io_free(u->sap_event);
 493     pa_sap_context_destroy(&u->sap_context);
 494
 495     pa_hashmap_free(u->by_origin, free_func, NULL);
 496
 497     pa_xfree(u->sink_name);
 498     pa_xfree(u);
 499 }