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

   1 /***
   2   This file is part of PulseAudio.
   3
   4   Copyright 2006 Lennart Poettering
   5
   6   PulseAudio is free software; you can redistribute it and/or modify
   7   it under the terms of the GNU Lesser General Public License as published
   8   by the Free Software Foundation; either version 2.1 of the License,
   9   or (at your option) any later version.
  10
  11   PulseAudio is distributed in the hope that it will be useful, but
  12   WITHOUT ANY WARRANTY; without even the implied warranty of
  13   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  14   General Public License for more details.
  15
  16   You should have received a copy of the GNU Lesser General Public License
  17   along with PulseAudio; if not, write to the Free Software
  18   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  19   USA.
  20 ***/
  21
  22 #ifdef HAVE_CONFIG_H
  23 #include <config.h>
  24 #endif
  25
  26 #include <stdio.h>
  27 #include <sys/socket.h>
  28 #include <netinet/in.h>
  29 #include <errno.h>
  30 #include <unistd.h>
  31
  32 #include <pulse/rtclock.h>
  33 #include <pulse/timeval.h>
  34 #include <pulse/util.h>
  35 #include <pulse/xmalloc.h>
  36
  37 #include <pulsecore/core-error.h>
  38 #include <pulsecore/module.h>
  39 #include <pulsecore/source.h>
  40 #include <pulsecore/source-output.h>
  41 #include <pulsecore/memblockq.h>
  42 #include <pulsecore/log.h>
  43 #include <pulsecore/core-util.h>
  44 #include <pulsecore/modargs.h>
  45 #include <pulsecore/namereg.h>
  46 #include <pulsecore/sample-util.h>
  47 #include <pulsecore/macro.h>
  48 #include <pulsecore/socket-util.h>
  49 #include <pulsecore/arpa-inet.h>
  50
  51 #include "module-rtp-send-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("Read data from source and send it to the network via RTP/SAP/SDP");
  59 PA_MODULE_VERSION(PACKAGE_VERSION);
  60 PA_MODULE_LOAD_ONCE(false);
  61 PA_MODULE_USAGE(
  62         "source=<name of the source> "
  63         "format=<sample format> "
  64         "channels=<number of channels> "
  65         "rate=<sample rate> "
  66         "destination_ip=<destination IP address> "
  67         "source_ip=<source IP address> "
  68         "port=<port number> "
  69         "mtu=<maximum transfer unit> "
  70         "loop=<loopback to local host?> "
  71         "ttl=<ttl value> "
  72         "inhibit_auto_suspend=<always|never|only_with_non_monitor_sources>"
  73 );
  74
  75 #define DEFAULT_PORT 46000
  76 #define DEFAULT_TTL 1
  77 #define SAP_PORT 9875
  78 #define DEFAULT_SOURCE_IP "0.0.0.0"
  79 #define DEFAULT_DESTINATION_IP "224.0.0.56"
  80 #define MEMBLOCKQ_MAXLENGTH (1024*170)
  81 #define DEFAULT_MTU 1280
  82 #define SAP_INTERVAL (5*PA_USEC_PER_SEC)
  83
  84 static const char* const valid_modargs[] = {
  85     "source",
  86     "format",
  87     "channels",
  88     "rate",
  89     "destination", /* Compatbility */
  90     "destination_ip",
  91     "source_ip",
  92     "port",
  93     "mtu" ,
  94     "loop",
  95     "ttl",
  96     "inhibit_auto_suspend",
  97     NULL
  98 };
  99
 100 enum inhibit_auto_suspend {
 101     INHIBIT_AUTO_SUSPEND_ALWAYS,
 102     INHIBIT_AUTO_SUSPEND_NEVER,
 103     INHIBIT_AUTO_SUSPEND_ONLY_WITH_NON_MONITOR_SOURCES
 104 };
 105
 106 struct userdata {
 107     pa_module *module;
 108
 109     pa_source_output *source_output;
 110     pa_memblockq *memblockq;
 111
 112     pa_rtp_context rtp_context;
 113     pa_sap_context sap_context;
 114     size_t mtu;
 115
 116     pa_time_event *sap_event;
 117
 118     enum inhibit_auto_suspend inhibit_auto_suspend;
 119 };
 120
 121 /* Called from I/O thread context */
 122 static int source_output_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 123     struct userdata *u;
 124     pa_assert_se(u = PA_SOURCE_OUTPUT(o)->userdata);
 125
 126     switch (code) {
 127         case PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY:
 128             *((pa_usec_t*) data) = pa_bytes_to_usec(pa_memblockq_get_length(u->memblockq), &u->source_output->sample_spec);
 129
 130             /* Fall through, the default handler will add in the extra
 131              * latency added by the resampler */
 132             break;
 133     }
 134
 135     return pa_source_output_process_msg(o, code, data, offset, chunk);
 136 }
 137
 138 /* Called from I/O thread context */
 139 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
 140     struct userdata *u;
 141     pa_source_output_assert_ref(o);
 142     pa_assert_se(u = o->userdata);
 143
 144     if (pa_memblockq_push(u->memblockq, chunk) < 0) {
 145         pa_log_warn("Failed to push chunk into memblockq.");
 146         return;
 147     }
 148
 149     pa_rtp_send(&u->rtp_context, u->mtu, u->memblockq);
 150 }
 151
 152 static pa_source_output_flags_t get_dont_inhibit_auto_suspend_flag(pa_source *source,
 153                                                                    enum inhibit_auto_suspend inhibit_auto_suspend) {
 154     pa_assert(source);
 155
 156     switch (inhibit_auto_suspend) {
 157         case INHIBIT_AUTO_SUSPEND_ALWAYS:
 158             return 0;
 159
 160         case INHIBIT_AUTO_SUSPEND_NEVER:
 161             return PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND;
 162
 163         case INHIBIT_AUTO_SUSPEND_ONLY_WITH_NON_MONITOR_SOURCES:
 164             return source->monitor_of ? 0 : PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND;
 165     }
 166
 167     pa_assert_not_reached();
 168 }
 169
 170 /* Called from the main thread. */
 171 static void source_output_moving_cb(pa_source_output *o, pa_source *dest) {
 172     struct userdata *u;
 173
 174     pa_assert(o);
 175
 176     u = o->userdata;
 177
 178     if (!dest)
 179         return;
 180
 181     o->flags &= ~PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND;
 182     o->flags |= get_dont_inhibit_auto_suspend_flag(dest, u->inhibit_auto_suspend);
 183 }
 184
 185 /* Called from main context */
 186 static void source_output_kill_cb(pa_source_output* o) {
 187     struct userdata *u;
 188     pa_source_output_assert_ref(o);
 189     pa_assert_se(u = o->userdata);
 190
 191     pa_module_unload_request(u->module, true);
 192
 193     pa_source_output_unlink(u->source_output);
 194     pa_source_output_unref(u->source_output);
 195     u->source_output = NULL;
 196 }
 197
 198 static void sap_event_cb(pa_mainloop_api *m, pa_time_event *t, const struct timeval *tv, void *userdata) {
 199     struct userdata *u = userdata;
 200
 201     pa_assert(m);
 202     pa_assert(t);
 203     pa_assert(u);
 204
 205     pa_sap_send(&u->sap_context, 0);
 206
 207     pa_core_rttime_restart(u->module->core, t, pa_rtclock_now() + SAP_INTERVAL);
 208 }
 209
 210 int pa__init(pa_module*m) {
 211     struct userdata *u;
 212     pa_modargs *ma = NULL;
 213     const char *dst_addr;
 214     const char *src_addr;
 215     uint32_t port = DEFAULT_PORT, mtu;
 216     uint32_t ttl = DEFAULT_TTL;
 217     sa_family_t af;
 218     int fd = -1, sap_fd = -1;
 219     pa_source *s;
 220     pa_sample_spec ss;
 221     pa_channel_map cm;
 222     struct sockaddr_in dst_sa4, dst_sap_sa4, src_sa4, src_sap_sa4;
 223 #ifdef HAVE_IPV6
 224     struct sockaddr_in6 dst_sa6, dst_sap_sa6, src_sa6, src_sap_sa6;
 225 #endif
 226     struct sockaddr_storage sa_dst;
 227     pa_source_output *o = NULL;
 228     uint8_t payload;
 229     char *p;
 230     int r, j;
 231     socklen_t k;
 232     char hn[128], *n;
 233     bool loop = false;
 234     enum inhibit_auto_suspend inhibit_auto_suspend = INHIBIT_AUTO_SUSPEND_ONLY_WITH_NON_MONITOR_SOURCES;
 235     const char *inhibit_auto_suspend_str;
 236     pa_source_output_new_data data;
 237
 238     pa_assert(m);
 239
 240     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
 241         pa_log("Failed to parse module arguments");
 242         goto fail;
 243     }
 244
 245     if (!(s = pa_namereg_get(m->core, pa_modargs_get_value(ma, "source", NULL), PA_NAMEREG_SOURCE))) {
 246         pa_log("Source does not exist.");
 247         goto fail;
 248     }
 249
 250     if (pa_modargs_get_value_boolean(ma, "loop", &loop) < 0) {
 251         pa_log("Failed to parse \"loop\" parameter.");
 252         goto fail;
 253     }
 254
 255     if ((inhibit_auto_suspend_str = pa_modargs_get_value(ma, "inhibit_auto_suspend", NULL))) {
 256         if (pa_streq(inhibit_auto_suspend_str, "always"))
 257             inhibit_auto_suspend = INHIBIT_AUTO_SUSPEND_ALWAYS;
 258         else if (pa_streq(inhibit_auto_suspend_str, "never"))
 259             inhibit_auto_suspend = INHIBIT_AUTO_SUSPEND_NEVER;
 260         else if (pa_streq(inhibit_auto_suspend_str, "only_with_non_monitor_sources"))
 261             inhibit_auto_suspend = INHIBIT_AUTO_SUSPEND_ONLY_WITH_NON_MONITOR_SOURCES;
 262         else {
 263             pa_log("Failed to parse the \"inhibit_auto_suspend\" parameter.");
 264             goto fail;
 265         }
 266     }
 267
 268     ss = s->sample_spec;
 269     pa_rtp_sample_spec_fixup(&ss);
 270     cm = s->channel_map;
 271     if (pa_modargs_get_sample_spec(ma, &ss) < 0) {
 272         pa_log("Failed to parse sample specification");
 273         goto fail;
 274     }
 275
 276     if (!pa_rtp_sample_spec_valid(&ss)) {
 277         pa_log("Specified sample type not compatible with RTP");
 278         goto fail;
 279     }
 280
 281     if (ss.channels != cm.channels)
 282         pa_channel_map_init_auto(&cm, ss.channels, PA_CHANNEL_MAP_AIFF);
 283
 284     payload = pa_rtp_payload_from_sample_spec(&ss);
 285
 286     mtu = (uint32_t) pa_frame_align(DEFAULT_MTU, &ss);
 287
 288     if (pa_modargs_get_value_u32(ma, "mtu", &mtu) < 0 || mtu < 1 || mtu % pa_frame_size(&ss) != 0) {
 289         pa_log("Invalid MTU.");
 290         goto fail;
 291     }
 292
 293     port = DEFAULT_PORT + ((uint32_t) (rand() % 512) << 1);
 294     if (pa_modargs_get_value_u32(ma, "port", &port) < 0 || port < 1 || port > 0xFFFF) {
 295         pa_log("port= expects a numerical argument between 1 and 65535.");
 296         goto fail;
 297     }
 298
 299     if (port & 1)
 300         pa_log_warn("Port number not even as suggested in RFC3550!");
 301
 302     if (pa_modargs_get_value_u32(ma, "ttl", &ttl) < 0 || ttl < 1 || ttl > 0xFF) {
 303         pa_log("ttl= expects a numerical argument between 1 and 255.");
 304         goto fail;
 305     }
 306
 307     src_addr = pa_modargs_get_value(ma, "source_ip", DEFAULT_SOURCE_IP);
 308
 309     if (inet_pton(AF_INET, src_addr, &src_sa4.sin_addr) > 0) {
 310         src_sa4.sin_family = af = AF_INET;
 311         src_sa4.sin_port = htons(0);
 312         memset(&src_sa4.sin_zero, 0, sizeof(src_sa4.sin_zero));
 313         src_sap_sa4 = src_sa4;
 314 #ifdef HAVE_IPV6
 315     } else if (inet_pton(AF_INET6, src_addr, &src_sa6.sin6_addr) > 0) {
 316         src_sa6.sin6_family = af = AF_INET6;
 317         src_sa6.sin6_port = htons(0);
 318         src_sa6.sin6_flowinfo = 0;
 319         src_sa6.sin6_scope_id = 0;
 320         src_sap_sa6 = src_sa6;
 321 #endif
 322     } else {
 323         pa_log("Invalid source address '%s'", src_addr);
 324         goto fail;
 325     }
 326
 327     dst_addr = pa_modargs_get_value(ma, "destination", NULL);
 328     if (dst_addr == NULL)
 329         dst_addr = pa_modargs_get_value(ma, "destination_ip", DEFAULT_DESTINATION_IP);
 330
 331     if (inet_pton(AF_INET, dst_addr, &dst_sa4.sin_addr) > 0) {
 332         dst_sa4.sin_family = af = AF_INET;
 333         dst_sa4.sin_port = htons((uint16_t) port);
 334         memset(&dst_sa4.sin_zero, 0, sizeof(dst_sa4.sin_zero));
 335         dst_sap_sa4 = dst_sa4;
 336         dst_sap_sa4.sin_port = htons(SAP_PORT);
 337 #ifdef HAVE_IPV6
 338     } else if (inet_pton(AF_INET6, dst_addr, &dst_sa6.sin6_addr) > 0) {
 339         dst_sa6.sin6_family = af = AF_INET6;
 340         dst_sa6.sin6_port = htons((uint16_t) port);
 341         dst_sa6.sin6_flowinfo = 0;
 342         dst_sa6.sin6_scope_id = 0;
 343         dst_sap_sa6 = dst_sa6;
 344         dst_sap_sa6.sin6_port = htons(SAP_PORT);
 345 #endif
 346     } else {
 347         pa_log("Invalid destination '%s'", dst_addr);
 348         goto fail;
 349     }
 350
 351     if ((fd = pa_socket_cloexec(af, SOCK_DGRAM, 0)) < 0) {
 352         pa_log("socket() failed: %s", pa_cstrerror(errno));
 353         goto fail;
 354     }
 355
 356     if (af == AF_INET && bind(fd, (struct sockaddr*) &src_sa4, sizeof(src_sa4)) < 0) {
 357         pa_log("bind() failed: %s", pa_cstrerror(errno));
 358         goto fail;
 359 #ifdef HAVE_IPV6
 360     } else if (af == AF_INET6 && bind(fd, (struct sockaddr*) &src_sa6, sizeof(src_sa6)) < 0) {
 361         pa_log("bind() failed: %s", pa_cstrerror(errno));
 362         goto fail;
 363 #endif
 364     }
 365
 366     if (af == AF_INET && connect(fd, (struct sockaddr*) &dst_sa4, sizeof(dst_sa4)) < 0) {
 367         pa_log("connect() failed: %s", pa_cstrerror(errno));
 368         goto fail;
 369 #ifdef HAVE_IPV6
 370     } else if (af == AF_INET6 && connect(fd, (struct sockaddr*) &dst_sa6, sizeof(dst_sa6)) < 0) {
 371         pa_log("connect() failed: %s", pa_cstrerror(errno));
 372         goto fail;
 373 #endif
 374     }
 375
 376     if ((sap_fd = pa_socket_cloexec(af, SOCK_DGRAM, 0)) < 0) {
 377         pa_log("socket() failed: %s", pa_cstrerror(errno));
 378         goto fail;
 379     }
 380
 381     if (af == AF_INET && bind(sap_fd, (struct sockaddr*) &src_sap_sa4, sizeof(src_sap_sa4)) < 0) {
 382         pa_log("bind() failed: %s", pa_cstrerror(errno));
 383         goto fail;
 384 #ifdef HAVE_IPV6
 385     } else if (af == AF_INET6 && bind(sap_fd, (struct sockaddr*) &src_sap_sa6, sizeof(src_sap_sa6)) < 0) {
 386         pa_log("bind() failed: %s", pa_cstrerror(errno));
 387         goto fail;
 388 #endif
 389     }
 390
 391     if (af == AF_INET && connect(sap_fd, (struct sockaddr*) &dst_sap_sa4, sizeof(dst_sap_sa4)) < 0) {
 392         pa_log("connect() failed: %s", pa_cstrerror(errno));
 393         goto fail;
 394 #ifdef HAVE_IPV6
 395     } else if (af == AF_INET6 && connect(sap_fd, (struct sockaddr*) &dst_sap_sa6, sizeof(dst_sap_sa6)) < 0) {
 396         pa_log("connect() failed: %s", pa_cstrerror(errno));
 397         goto fail;
 398 #endif
 399     }
 400
 401     j = !!loop;
 402     if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_LOOP, &j, sizeof(j)) < 0 ||
 403         setsockopt(sap_fd, IPPROTO_IP, IP_MULTICAST_LOOP, &j, sizeof(j)) < 0) {
 404         pa_log("IP_MULTICAST_LOOP failed: %s", pa_cstrerror(errno));
 405         goto fail;
 406     }
 407
 408     if (ttl != DEFAULT_TTL) {
 409         int _ttl = (int) ttl;
 410
 411         if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_TTL, &_ttl, sizeof(_ttl)) < 0) {
 412             pa_log("IP_MULTICAST_TTL failed: %s", pa_cstrerror(errno));
 413             goto fail;
 414         }
 415
 416         if (setsockopt(sap_fd, IPPROTO_IP, IP_MULTICAST_TTL, &_ttl, sizeof(_ttl)) < 0) {
 417             pa_log("IP_MULTICAST_TTL (sap) failed: %s", pa_cstrerror(errno));
 418             goto fail;
 419         }
 420     }
 421
 422     /* If the socket queue is full, let's drop packets */
 423     pa_make_fd_nonblock(fd);
 424     pa_make_udp_socket_low_delay(fd);
 425
 426     pa_source_output_new_data_init(&data);
 427     pa_proplist_sets(data.proplist, PA_PROP_MEDIA_NAME, "RTP Monitor Stream");
 428     pa_proplist_sets(data.proplist, "rtp.source", src_addr);
 429     pa_proplist_sets(data.proplist, "rtp.destination", dst_addr);
 430     pa_proplist_setf(data.proplist, "rtp.mtu", "%lu", (unsigned long) mtu);
 431     pa_proplist_setf(data.proplist, "rtp.port", "%lu", (unsigned long) port);
 432     pa_proplist_setf(data.proplist, "rtp.ttl", "%lu", (unsigned long) ttl);
 433     data.driver = __FILE__;
 434     data.module = m;
 435     pa_source_output_new_data_set_source(&data, s, false);
 436     pa_source_output_new_data_set_sample_spec(&data, &ss);
 437     pa_source_output_new_data_set_channel_map(&data, &cm);
 438     data.flags |= get_dont_inhibit_auto_suspend_flag(s, inhibit_auto_suspend);
 439
 440     pa_source_output_new(&o, m->core, &data);
 441     pa_source_output_new_data_done(&data);
 442
 443     if (!o) {
 444         pa_log("failed to create source output.");
 445         goto fail;
 446     }
 447
 448     o->parent.process_msg = source_output_process_msg;
 449     o->push = source_output_push_cb;
 450     o->moving = source_output_moving_cb;
 451     o->kill = source_output_kill_cb;
 452
 453     pa_log_info("Configured source latency of %llu ms.",
 454                 (unsigned long long) pa_source_output_set_requested_latency(o, pa_bytes_to_usec(mtu, &o->sample_spec)) / PA_USEC_PER_MSEC);
 455
 456     m->userdata = o->userdata = u = pa_xnew(struct userdata, 1);
 457     u->module = m;
 458     u->source_output = o;
 459
 460     u->memblockq = pa_memblockq_new(
 461             "module-rtp-send memblockq",
 462             0,
 463             MEMBLOCKQ_MAXLENGTH,
 464             MEMBLOCKQ_MAXLENGTH,
 465             &ss,
 466             1,
 467             0,
 468             0,
 469             NULL);
 470
 471     u->mtu = mtu;
 472
 473     k = sizeof(sa_dst);
 474     pa_assert_se((r = getsockname(fd, (struct sockaddr*) &sa_dst, &k)) >= 0);
 475
 476     n = pa_sprintf_malloc("PulseAudio RTP Stream on %s", pa_get_fqdn(hn, sizeof(hn)));
 477
 478     if (af == AF_INET) {
 479         p = pa_sdp_build(af,
 480                      (void*) &((struct sockaddr_in*) &sa_dst)->sin_addr,
 481                      (void*) &dst_sa4.sin_addr,
 482                      n, (uint16_t) port, payload, &ss);
 483 #ifdef HAVE_IPV6
 484     } else {
 485         p = pa_sdp_build(af,
 486                      (void*) &((struct sockaddr_in6*) &sa_dst)->sin6_addr,
 487                      (void*) &dst_sa6.sin6_addr,
 488                      n, (uint16_t) port, payload, &ss);
 489 #endif
 490     }
 491
 492     pa_xfree(n);
 493
 494     pa_rtp_context_init_send(&u->rtp_context, fd, m->core->cookie, payload, pa_frame_size(&ss));
 495     pa_sap_context_init_send(&u->sap_context, sap_fd, p);
 496
 497     pa_log_info("RTP stream initialized with mtu %u on %s:%u from %s ttl=%u, SSRC=0x%08x, payload=%u, initial sequence #%u", mtu, dst_addr, port, src_addr, ttl, u->rtp_context.ssrc, payload, u->rtp_context.sequence);
 498     pa_log_info("SDP-Data:\n%s\nEOF", p);
 499
 500     pa_sap_send(&u->sap_context, 0);
 501
 502     u->sap_event = pa_core_rttime_new(m->core, pa_rtclock_now() + SAP_INTERVAL, sap_event_cb, u);
 503     u->inhibit_auto_suspend = inhibit_auto_suspend;
 504
 505     pa_source_output_put(u->source_output);
 506
 507     pa_modargs_free(ma);
 508
 509     return 0;
 510
 511 fail:
 512     if (ma)
 513         pa_modargs_free(ma);
 514
 515     if (fd >= 0)
 516         pa_close(fd);
 517
 518     if (sap_fd >= 0)
 519         pa_close(sap_fd);
 520
 521     if (o) {
 522         pa_source_output_unlink(o);
 523         pa_source_output_unref(o);
 524     }
 525
 526     return -1;
 527 }
 528
 529 void pa__done(pa_module*m) {
 530     struct userdata *u;
 531     pa_assert(m);
 532
 533     if (!(u = m->userdata))
 534         return;
 535
 536     if (u->sap_event)
 537         m->core->mainloop->time_free(u->sap_event);
 538
 539     if (u->source_output) {
 540         pa_source_output_unlink(u->source_output);
 541         pa_source_output_unref(u->source_output);
 542     }
 543
 544     pa_rtp_context_destroy(&u->rtp_context);
 545
 546     pa_sap_send(&u->sap_context, 1);
 547     pa_sap_context_destroy(&u->sap_context);
 548
 549     if (u->memblockq)
 550         pa_memblockq_free(u->memblockq);
 551
 552     pa_xfree(u);
 553 }