code.delx.au - pulseaudio/blob - src/modules/module-tunnel-source-new.c

   1 /***
   2     This file is part of PulseAudio.
   3
   4     Copyright 2013 Alexander Couzens
   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 <pulse/context.h>
  27 #include <pulse/timeval.h>
  28 #include <pulse/xmalloc.h>
  29 #include <pulse/stream.h>
  30 #include <pulse/mainloop.h>
  31 #include <pulse/introspect.h>
  32 #include <pulse/error.h>
  33
  34 #include <pulsecore/core.h>
  35 #include <pulsecore/core-util.h>
  36 #include <pulsecore/i18n.h>
  37 #include <pulsecore/source.h>
  38 #include <pulsecore/modargs.h>
  39 #include <pulsecore/log.h>
  40 #include <pulsecore/thread.h>
  41 #include <pulsecore/thread-mq.h>
  42 #include <pulsecore/poll.h>
  43 #include <pulsecore/proplist-util.h>
  44
  45 #include "module-tunnel-source-new-symdef.h"
  46
  47 PA_MODULE_AUTHOR("Alexander Couzens");
  48 PA_MODULE_DESCRIPTION("Create a network source which connects via a stream to a remote PulseAudio server");
  49 PA_MODULE_VERSION(PACKAGE_VERSION);
  50 PA_MODULE_LOAD_ONCE(false);
  51 PA_MODULE_USAGE(
  52         "server=<address> "
  53         "source=<name of the remote source> "
  54         "source_name=<name for the local source> "
  55         "source_properties=<properties for the local source> "
  56         "format=<sample format> "
  57         "channels=<number of channels> "
  58         "rate=<sample rate> "
  59         "channel_map=<channel map> "
  60         "cookie=<cookie file path>"
  61         );
  62
  63 #define TUNNEL_THREAD_FAILED_MAINLOOP 1
  64
  65 static void stream_state_cb(pa_stream *stream, void *userdata);
  66 static void stream_read_cb(pa_stream *s, size_t length, void *userdata);
  67 static void context_state_cb(pa_context *c, void *userdata);
  68 static void source_update_requested_latency_cb(pa_source *s);
  69
  70 struct userdata {
  71     pa_module *module;
  72     pa_source *source;
  73     pa_thread *thread;
  74     pa_thread_mq *thread_mq;
  75     pa_mainloop *thread_mainloop;
  76     pa_mainloop_api *thread_mainloop_api;
  77
  78     pa_context *context;
  79     pa_stream *stream;
  80
  81     bool update_stream_bufferattr_after_connect;
  82     bool connected;
  83     bool new_data;
  84
  85     char *cookie_file;
  86     char *remote_server;
  87     char *remote_source_name;
  88 };
  89
  90 static const char* const valid_modargs[] = {
  91     "source_name",
  92     "source_properties",
  93     "server",
  94     "source",
  95     "format",
  96     "channels",
  97     "rate",
  98     "channel_map",
  99     "cookie",
 100    /* "reconnect", reconnect if server comes back again - unimplemented */
 101     NULL,
 102 };
 103
 104 static void reset_bufferattr(pa_buffer_attr *bufferattr) {
 105     pa_assert(bufferattr);
 106     bufferattr->fragsize = (uint32_t) -1;
 107     bufferattr->minreq = (uint32_t) -1;
 108     bufferattr->maxlength = (uint32_t) -1;
 109     bufferattr->prebuf = (uint32_t) -1;
 110     bufferattr->tlength = (uint32_t) -1;
 111 }
 112
 113 static pa_proplist* tunnel_new_proplist(struct userdata *u) {
 114     pa_proplist *proplist = pa_proplist_new();
 115     pa_assert(proplist);
 116     pa_proplist_sets(proplist, PA_PROP_APPLICATION_NAME, "PulseAudio");
 117     pa_proplist_sets(proplist, PA_PROP_APPLICATION_ID, "org.PulseAudio.PulseAudio");
 118     pa_proplist_sets(proplist, PA_PROP_APPLICATION_VERSION, PACKAGE_VERSION);
 119     pa_init_proplist(proplist);
 120
 121     return proplist;
 122 }
 123
 124 static void stream_read_cb(pa_stream *s, size_t length, void *userdata) {
 125     struct userdata *u = userdata;
 126     u->new_data = true;
 127 }
 128
 129 /* called from io context to read samples from the stream into our source */
 130 static void read_new_samples(struct userdata *u) {
 131     const void *p;
 132     size_t readable = 0;
 133     pa_memchunk memchunk;
 134
 135     pa_assert(u);
 136     u->new_data = false;
 137
 138     pa_memchunk_reset(&memchunk);
 139
 140     if (PA_UNLIKELY(!u->connected || pa_stream_get_state(u->stream) != PA_STREAM_READY))
 141         return;
 142
 143     readable = pa_stream_readable_size(u->stream);
 144     while (readable > 0) {
 145         size_t nbytes = 0;
 146         if (PA_UNLIKELY(pa_stream_peek(u->stream, &p, &nbytes) != 0)) {
 147             pa_log("pa_stream_peek() failed: %s", pa_strerror(pa_context_errno(u->context)));
 148             u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
 149             return;
 150         }
 151
 152         if (PA_LIKELY(p)) {
 153             /* we have valid data */
 154             memchunk.memblock = pa_memblock_new_fixed(u->module->core->mempool, (void *) p, nbytes, true);
 155             memchunk.length = nbytes;
 156             memchunk.index = 0;
 157
 158             pa_source_post(u->source, &memchunk);
 159             pa_memblock_unref_fixed(memchunk.memblock);
 160         } else {
 161             size_t bytes_to_generate = nbytes;
 162
 163             /* we have a hole. generate silence */
 164             memchunk = u->source->silence;
 165             pa_memblock_ref(memchunk.memblock);
 166
 167             while (bytes_to_generate > 0) {
 168                 if (bytes_to_generate < memchunk.length)
 169                     memchunk.length = bytes_to_generate;
 170
 171                 pa_source_post(u->source, &memchunk);
 172                 bytes_to_generate -= memchunk.length;
 173             }
 174
 175             pa_memblock_unref(memchunk.memblock);
 176         }
 177
 178         pa_stream_drop(u->stream);
 179         readable -= nbytes;
 180     }
 181 }
 182
 183 static void thread_func(void *userdata) {
 184     struct userdata *u = userdata;
 185     pa_proplist *proplist;
 186
 187     pa_assert(u);
 188
 189     pa_log_debug("Thread starting up");
 190     pa_thread_mq_install(u->thread_mq);
 191
 192     proplist = tunnel_new_proplist(u);
 193     u->context = pa_context_new_with_proplist(u->thread_mainloop_api,
 194                                               "PulseAudio",
 195                                               proplist);
 196     pa_proplist_free(proplist);
 197
 198     if (!u->context) {
 199         pa_log("Failed to create libpulse context");
 200         goto fail;
 201     }
 202
 203     if (u->cookie_file && pa_context_load_cookie_from_file(u->context, u->cookie_file) != 0) {
 204         pa_log_error("Can not load cookie file!");
 205         goto fail;
 206     }
 207
 208     pa_context_set_state_callback(u->context, context_state_cb, u);
 209     if (pa_context_connect(u->context,
 210                            u->remote_server,
 211                            PA_CONTEXT_NOAUTOSPAWN,
 212                            NULL) < 0) {
 213         pa_log("Failed to connect libpulse context: %s", pa_strerror(pa_context_errno(u->context)));
 214         goto fail;
 215     }
 216
 217     for (;;) {
 218         int ret;
 219
 220         if (pa_mainloop_iterate(u->thread_mainloop, 1, &ret) < 0) {
 221             if (ret == 0)
 222                 goto finish;
 223             else
 224                 goto fail;
 225         }
 226
 227         if (u->new_data)
 228             read_new_samples(u);
 229     }
 230 fail:
 231     pa_asyncmsgq_post(u->thread_mq->outq, PA_MSGOBJECT(u->module->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
 232     pa_asyncmsgq_wait_for(u->thread_mq->inq, PA_MESSAGE_SHUTDOWN);
 233
 234 finish:
 235     if (u->stream) {
 236         pa_stream_disconnect(u->stream);
 237         pa_stream_unref(u->stream);
 238         u->stream = NULL;
 239     }
 240
 241     if (u->context) {
 242         pa_context_disconnect(u->context);
 243         pa_context_unref(u->context);
 244         u->context = NULL;
 245     }
 246
 247     pa_log_debug("Thread shutting down");
 248 }
 249
 250 static void stream_state_cb(pa_stream *stream, void *userdata) {
 251     struct userdata *u = userdata;
 252
 253     pa_assert(u);
 254
 255     switch (pa_stream_get_state(stream)) {
 256         case PA_STREAM_FAILED:
 257             pa_log_error("Stream failed: %s", pa_strerror(pa_context_errno(u->context)));
 258             u->connected = false;
 259             u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
 260             break;
 261         case PA_STREAM_TERMINATED:
 262             pa_log_debug("Stream terminated.");
 263             break;
 264         case PA_STREAM_READY:
 265             /* Only call our requested_latency_cb when requested_latency
 266              * changed between PA_STREAM_CREATING -> PA_STREAM_READY, because
 267              * we don't want to override the initial fragsize set by the server
 268              * without a good reason. */
 269             if (u->update_stream_bufferattr_after_connect)
 270                 source_update_requested_latency_cb(u->source);
 271         case PA_STREAM_UNCONNECTED:
 272         case PA_STREAM_CREATING:
 273             break;
 274     }
 275 }
 276
 277 static void context_state_cb(pa_context *c, void *userdata) {
 278     struct userdata *u = userdata;
 279     pa_assert(u);
 280
 281     switch (pa_context_get_state(c)) {
 282         case PA_CONTEXT_UNCONNECTED:
 283         case PA_CONTEXT_CONNECTING:
 284         case PA_CONTEXT_AUTHORIZING:
 285         case PA_CONTEXT_SETTING_NAME:
 286             break;
 287         case PA_CONTEXT_READY: {
 288             pa_proplist *proplist;
 289             pa_buffer_attr bufferattr;
 290             pa_usec_t requested_latency;
 291             char *username = pa_get_user_name_malloc();
 292             char *hostname = pa_get_host_name_malloc();
 293             /* TODO: old tunnel put here the remote source_name into stream name e.g. 'Null Output for lynxis@lazus' */
 294             char *stream_name = pa_sprintf_malloc(_("Tunnel for %s@%s"), username, hostname);
 295             pa_xfree(username);
 296             pa_xfree(hostname);
 297
 298             pa_log_debug("Connection successful. Creating stream.");
 299             pa_assert(!u->stream);
 300
 301             proplist = tunnel_new_proplist(u);
 302             u->stream = pa_stream_new_with_proplist(u->context,
 303                                                     stream_name,
 304                                                     &u->source->sample_spec,
 305                                                     &u->source->channel_map,
 306                                                     proplist);
 307             pa_proplist_free(proplist);
 308             pa_xfree(stream_name);
 309
 310             if (!u->stream) {
 311                 pa_log_error("Could not create a stream: %s", pa_strerror(pa_context_errno(u->context)));
 312                 u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
 313                 return;
 314             }
 315
 316             requested_latency = pa_source_get_requested_latency_within_thread(u->source);
 317             if (requested_latency == (uint32_t) -1)
 318                 requested_latency = u->source->thread_info.max_latency;
 319
 320             reset_bufferattr(&bufferattr);
 321             bufferattr.fragsize = pa_usec_to_bytes(requested_latency, &u->source->sample_spec);
 322
 323             pa_stream_set_state_callback(u->stream, stream_state_cb, userdata);
 324             pa_stream_set_read_callback(u->stream, stream_read_cb, userdata);
 325             if (pa_stream_connect_record(u->stream,
 326                                          u->remote_source_name,
 327                                          &bufferattr,
 328                                          PA_STREAM_INTERPOLATE_TIMING|PA_STREAM_DONT_MOVE|PA_STREAM_AUTO_TIMING_UPDATE) < 0) {
 329                 pa_log_debug("Could not create stream: %s", pa_strerror(pa_context_errno(u->context)));
 330                 u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
 331             }
 332             u->connected = true;
 333             break;
 334         }
 335         case PA_CONTEXT_FAILED:
 336             pa_log_debug("Context failed with err %s.", pa_strerror(pa_context_errno(u->context)));
 337             u->connected = false;
 338             u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
 339             break;
 340         case PA_CONTEXT_TERMINATED:
 341             pa_log_debug("Context terminated.");
 342             u->connected = false;
 343             u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
 344             break;
 345     }
 346 }
 347
 348 static void source_update_requested_latency_cb(pa_source *s) {
 349     struct userdata *u;
 350     pa_operation *operation;
 351     size_t nbytes;
 352     pa_usec_t block_usec;
 353     pa_buffer_attr bufferattr;
 354
 355     pa_source_assert_ref(s);
 356     pa_assert_se(u = s->userdata);
 357
 358     block_usec = pa_source_get_requested_latency_within_thread(s);
 359     if (block_usec == (pa_usec_t) -1)
 360         block_usec = s->thread_info.max_latency;
 361
 362     nbytes = pa_usec_to_bytes(block_usec, &s->sample_spec);
 363
 364     if (u->stream) {
 365         switch (pa_stream_get_state(u->stream)) {
 366             case PA_STREAM_READY:
 367                 if (pa_stream_get_buffer_attr(u->stream)->fragsize == nbytes)
 368                     break;
 369
 370                 reset_bufferattr(&bufferattr);
 371                 bufferattr.fragsize = nbytes;
 372                 if ((operation = pa_stream_set_buffer_attr(u->stream, &bufferattr, NULL, NULL)))
 373                     pa_operation_unref(operation);
 374                 break;
 375             case PA_STREAM_CREATING:
 376                 /* we have to delay our request until stream is ready */
 377                 u->update_stream_bufferattr_after_connect = true;
 378                 break;
 379             default:
 380                 break;
 381         }
 382     }
 383 }
 384
 385 static int source_process_msg_cb(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 386     struct userdata *u = PA_SOURCE(o)->userdata;
 387
 388     switch (code) {
 389         case PA_SOURCE_MESSAGE_GET_LATENCY: {
 390             int negative;
 391             pa_usec_t remote_latency;
 392
 393             if (!PA_SOURCE_IS_LINKED(u->source->thread_info.state)) {
 394                 *((pa_usec_t*) data) = 0;
 395                 return 0;
 396             }
 397
 398             if (!u->stream) {
 399                 *((pa_usec_t*) data) = 0;
 400                 return 0;
 401             }
 402
 403             if (pa_stream_get_state(u->stream) != PA_STREAM_READY) {
 404                 *((pa_usec_t*) data) = 0;
 405                 return 0;
 406             }
 407
 408             if (pa_stream_get_latency(u->stream, &remote_latency, &negative) < 0) {
 409                 *((pa_usec_t*) data) = 0;
 410                 return 0;
 411             }
 412
 413             if (negative)
 414                 *((pa_usec_t*) data) = 0;
 415             else
 416                 *((pa_usec_t*) data) = remote_latency;
 417
 418             return 0;
 419         }
 420     }
 421     return pa_source_process_msg(o, code, data, offset, chunk);
 422 }
 423
 424 int pa__init(pa_module *m) {
 425     struct userdata *u = NULL;
 426     pa_modargs *ma = NULL;
 427     pa_source_new_data source_data;
 428     pa_sample_spec ss;
 429     pa_channel_map map;
 430     const char *remote_server = NULL;
 431     const char *source_name = NULL;
 432     char *default_source_name = NULL;
 433
 434     pa_assert(m);
 435
 436     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
 437         pa_log("Failed to parse module arguments.");
 438         goto fail;
 439     }
 440
 441     ss = m->core->default_sample_spec;
 442     map = m->core->default_channel_map;
 443     if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0) {
 444         pa_log("Invalid sample format specification or channel map");
 445         goto fail;
 446     }
 447
 448     remote_server = pa_modargs_get_value(ma, "server", NULL);
 449     if (!remote_server) {
 450         pa_log("No server given!");
 451         goto fail;
 452     }
 453
 454     u = pa_xnew0(struct userdata, 1);
 455     u->module = m;
 456     m->userdata = u;
 457     u->remote_server = pa_xstrdup(remote_server);
 458     u->thread_mainloop = pa_mainloop_new();
 459     if (u->thread_mainloop == NULL) {
 460         pa_log("Failed to create mainloop");
 461         goto fail;
 462     }
 463     u->thread_mainloop_api = pa_mainloop_get_api(u->thread_mainloop);
 464     u->cookie_file = pa_xstrdup(pa_modargs_get_value(ma, "cookie", NULL));
 465     u->remote_source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));
 466
 467     u->thread_mq = pa_xnew0(pa_thread_mq, 1);
 468     pa_thread_mq_init_thread_mainloop(u->thread_mq, m->core->mainloop, u->thread_mainloop_api);
 469
 470     /* Create source */
 471     pa_source_new_data_init(&source_data);
 472     source_data.driver = __FILE__;
 473     source_data.module = m;
 474
 475     default_source_name = pa_sprintf_malloc("tunnel-source-new.%s", remote_server);
 476     source_name = pa_modargs_get_value(ma, "source_name", default_source_name);
 477
 478     pa_source_new_data_set_name(&source_data, source_name);
 479     pa_source_new_data_set_sample_spec(&source_data, &ss);
 480     pa_source_new_data_set_channel_map(&source_data, &map);
 481
 482     pa_proplist_sets(source_data.proplist, PA_PROP_DEVICE_CLASS, "sound");
 483     pa_proplist_setf(source_data.proplist,
 484                      PA_PROP_DEVICE_DESCRIPTION,
 485                      _("Tunnel to %s/%s"),
 486                      remote_server,
 487                      pa_strempty(u->remote_source_name));
 488
 489     if (pa_modargs_get_proplist(ma, "source_properties", source_data.proplist, PA_UPDATE_REPLACE) < 0) {
 490         pa_log("Invalid properties");
 491         pa_source_new_data_done(&source_data);
 492         goto fail;
 493     }
 494     if (!(u->source = pa_source_new(m->core, &source_data, PA_SOURCE_LATENCY | PA_SOURCE_DYNAMIC_LATENCY | PA_SOURCE_NETWORK))) {
 495         pa_log("Failed to create source.");
 496         pa_source_new_data_done(&source_data);
 497         goto fail;
 498     }
 499
 500     pa_source_new_data_done(&source_data);
 501     u->source->userdata = u;
 502     u->source->parent.process_msg = source_process_msg_cb;
 503     u->source->update_requested_latency = source_update_requested_latency_cb;
 504
 505     pa_source_set_asyncmsgq(u->source, u->thread_mq->inq);
 506
 507     if (!(u->thread = pa_thread_new("tunnel-source", thread_func, u))) {
 508         pa_log("Failed to create thread.");
 509         goto fail;
 510     }
 511
 512     pa_source_put(u->source);
 513     pa_modargs_free(ma);
 514     pa_xfree(default_source_name);
 515
 516     return 0;
 517
 518 fail:
 519     if (ma)
 520         pa_modargs_free(ma);
 521
 522     if (default_source_name)
 523         pa_xfree(default_source_name);
 524
 525     pa__done(m);
 526
 527     return -1;
 528 }
 529
 530 void pa__done(pa_module *m) {
 531     struct userdata *u;
 532
 533     pa_assert(m);
 534
 535     if (!(u = m->userdata))
 536         return;
 537
 538     if (u->source)
 539         pa_source_unlink(u->source);
 540
 541     if (u->thread) {
 542         pa_asyncmsgq_send(u->thread_mq->inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
 543         pa_thread_free(u->thread);
 544     }
 545
 546     if (u->thread_mq) {
 547         pa_thread_mq_done(u->thread_mq);
 548         pa_xfree(u->thread_mq);
 549     }
 550
 551     if (u->thread_mainloop)
 552         pa_mainloop_free(u->thread_mainloop);
 553
 554     if (u->cookie_file)
 555         pa_xfree(u->cookie_file);
 556
 557     if (u->remote_source_name)
 558         pa_xfree(u->remote_source_name);
 559
 560     if (u->remote_server)
 561         pa_xfree(u->remote_server);
 562
 563     if (u->source)
 564         pa_source_unref(u->source);
 565
 566     pa_xfree(u);
 567 }