src/modules/rtp/module-rtp-recv.c

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