src/modules/module-loopback.c

   1 /***
   2     This file is part of PulseAudio.
   3
   4     Copyright 2009 Intel Corporation
   5     Contributor: Pierre-Louis Bossart <pierre-louis.bossart@intel.com>
   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.1 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 <stdio.h>
  28
  29 #include <pulse/xmalloc.h>
  30
  31 #include <pulsecore/sink-input.h>
  32 #include <pulsecore/module.h>
  33 #include <pulsecore/modargs.h>
  34 #include <pulsecore/namereg.h>
  35 #include <pulsecore/log.h>
  36 #include <pulsecore/core-util.h>
  37
  38 #include <pulse/rtclock.h>
  39 #include <pulse/timeval.h>
  40
  41 #include "module-loopback-symdef.h"
  42
  43 PA_MODULE_AUTHOR("Pierre-Louis Bossart");
  44 PA_MODULE_DESCRIPTION("Loopback from source to sink");
  45 PA_MODULE_VERSION(PACKAGE_VERSION);
  46 PA_MODULE_LOAD_ONCE(FALSE);
  47 PA_MODULE_USAGE(
  48         "source=<source to connect to> "
  49         "sink=<sink to connect to> "
  50         "adjust_time=<how often to readjust rates in s> "
  51         "latency_msec=<latency in ms> "
  52         "format=<sample format> "
  53         "rate=<sample rate> "
  54         "channels=<number of channels> "
  55         "channel_map=<channel map> "
  56         "sink_input_properties=<proplist> "
  57         "source_output_properties=<proplist> "
  58         "source_dont_move=<boolean> "
  59         "sink_dont_move=<boolean> "
  60         "remix=<remix channels?> ");
  61
  62 #define DEFAULT_LATENCY_MSEC 200
  63
  64 #define MEMBLOCKQ_MAXLENGTH (1024*1024*16)
  65
  66 #define DEFAULT_ADJUST_TIME_USEC (10*PA_USEC_PER_SEC)
  67
  68 struct userdata {
  69     pa_core *core;
  70     pa_module *module;
  71
  72     pa_sink_input *sink_input;
  73     pa_source_output *source_output;
  74
  75     pa_asyncmsgq *asyncmsgq;
  76     pa_memblockq *memblockq;
  77
  78     pa_rtpoll_item *rtpoll_item_read, *rtpoll_item_write;
  79
  80     pa_time_event *time_event;
  81     pa_usec_t adjust_time;
  82
  83     int64_t recv_counter;
  84     int64_t send_counter;
  85
  86     size_t skip;
  87     pa_usec_t latency;
  88
  89     pa_bool_t in_pop;
  90     size_t min_memblockq_length;
  91
  92     struct {
  93         int64_t send_counter;
  94         size_t source_output_buffer;
  95         pa_usec_t source_latency;
  96
  97         int64_t recv_counter;
  98         size_t sink_input_buffer;
  99         pa_usec_t sink_latency;
 100
 101         size_t min_memblockq_length;
 102         size_t max_request;
 103     } latency_snapshot;
 104 };
 105
 106 static const char* const valid_modargs[] = {
 107     "source",
 108     "sink",
 109     "adjust_time",
 110     "latency_msec",
 111     "format",
 112     "rate",
 113     "channels",
 114     "channel_map",
 115     "sink_input_properties",
 116     "source_output_properties",
 117     "source_dont_move",
 118     "sink_dont_move",
 119     "remix",
 120     NULL,
 121 };
 122
 123 enum {
 124     SINK_INPUT_MESSAGE_POST = PA_SINK_INPUT_MESSAGE_MAX,
 125     SINK_INPUT_MESSAGE_REWIND,
 126     SINK_INPUT_MESSAGE_LATENCY_SNAPSHOT,
 127     SINK_INPUT_MESSAGE_MAX_REQUEST_CHANGED
 128 };
 129
 130 enum {
 131     SOURCE_OUTPUT_MESSAGE_LATENCY_SNAPSHOT
 132 };
 133
 134 static void enable_adjust_timer(struct userdata *u, bool enable);
 135
 136 /* Called from main context */
 137 static void teardown(struct userdata *u) {
 138     pa_assert(u);
 139     pa_assert_ctl_context();
 140
 141     if (u->asyncmsgq)
 142         pa_asyncmsgq_flush(u->asyncmsgq, 0);
 143
 144     u->adjust_time = 0;
 145     enable_adjust_timer(u, false);
 146
 147     if (u->sink_input)
 148         pa_sink_input_unlink(u->sink_input);
 149
 150     if (u->source_output)
 151         pa_source_output_unlink(u->source_output);
 152
 153     if (u->sink_input) {
 154         u->sink_input->parent.process_msg = pa_sink_input_process_msg;
 155         pa_sink_input_unref(u->sink_input);
 156         u->sink_input = NULL;
 157     }
 158
 159     if (u->source_output) {
 160         u->source_output->parent.process_msg = pa_source_output_process_msg;
 161         pa_source_output_unref(u->source_output);
 162         u->source_output = NULL;
 163     }
 164 }
 165
 166 /* Called from main context */
 167 static void adjust_rates(struct userdata *u) {
 168     size_t buffer, fs;
 169     uint32_t old_rate, base_rate, new_rate;
 170     pa_usec_t buffer_latency;
 171
 172     pa_assert(u);
 173     pa_assert_ctl_context();
 174
 175     pa_asyncmsgq_send(u->source_output->source->asyncmsgq, PA_MSGOBJECT(u->source_output), SOURCE_OUTPUT_MESSAGE_LATENCY_SNAPSHOT, NULL, 0, NULL);
 176     pa_asyncmsgq_send(u->sink_input->sink->asyncmsgq, PA_MSGOBJECT(u->sink_input), SINK_INPUT_MESSAGE_LATENCY_SNAPSHOT, NULL, 0, NULL);
 177
 178     buffer =
 179         u->latency_snapshot.sink_input_buffer +
 180         u->latency_snapshot.source_output_buffer;
 181
 182     if (u->latency_snapshot.recv_counter <= u->latency_snapshot.send_counter)
 183         buffer += (size_t) (u->latency_snapshot.send_counter - u->latency_snapshot.recv_counter);
 184     else
 185         buffer += PA_CLIP_SUB(buffer, (size_t) (u->latency_snapshot.recv_counter - u->latency_snapshot.send_counter));
 186
 187     buffer_latency = pa_bytes_to_usec(buffer, &u->sink_input->sample_spec);
 188
 189     pa_log_debug("Loopback overall latency is %0.2f ms + %0.2f ms + %0.2f ms = %0.2f ms",
 190                 (double) u->latency_snapshot.sink_latency / PA_USEC_PER_MSEC,
 191                 (double) buffer_latency / PA_USEC_PER_MSEC,
 192                 (double) u->latency_snapshot.source_latency / PA_USEC_PER_MSEC,
 193                 ((double) u->latency_snapshot.sink_latency + buffer_latency + u->latency_snapshot.source_latency) / PA_USEC_PER_MSEC);
 194
 195     pa_log_debug("Should buffer %zu bytes, buffered at minimum %zu bytes",
 196                 u->latency_snapshot.max_request*2,
 197                 u->latency_snapshot.min_memblockq_length);
 198
 199     fs = pa_frame_size(&u->sink_input->sample_spec);
 200     old_rate = u->sink_input->sample_spec.rate;
 201     base_rate = u->source_output->sample_spec.rate;
 202
 203     if (u->latency_snapshot.min_memblockq_length < u->latency_snapshot.max_request*2)
 204         new_rate = base_rate - (((u->latency_snapshot.max_request*2 - u->latency_snapshot.min_memblockq_length) / fs) *PA_USEC_PER_SEC)/u->adjust_time;
 205     else
 206         new_rate = base_rate + (((u->latency_snapshot.min_memblockq_length - u->latency_snapshot.max_request*2) / fs) *PA_USEC_PER_SEC)/u->adjust_time;
 207
 208     if (new_rate < (uint32_t) (base_rate*0.8) || new_rate > (uint32_t) (base_rate*1.25)) {
 209         pa_log_warn("Sample rates too different, not adjusting (%u vs. %u).", base_rate, new_rate);
 210         new_rate = base_rate;
 211     } else {
 212         if (base_rate < new_rate + 20 && new_rate < base_rate + 20)
 213           new_rate = base_rate;
 214         /* Do the adjustment in small steps; 2‰ can be considered inaudible */
 215         if (new_rate < (uint32_t) (old_rate*0.998) || new_rate > (uint32_t) (old_rate*1.002)) {
 216             pa_log_info("New rate of %u Hz not within 2‰ of %u Hz, forcing smaller adjustment", new_rate, old_rate);
 217             new_rate = PA_CLAMP(new_rate, (uint32_t) (old_rate*0.998), (uint32_t) (old_rate*1.002));
 218         }
 219     }
 220
 221     pa_sink_input_set_rate(u->sink_input, new_rate);
 222     pa_log_debug("[%s] Updated sampling rate to %lu Hz.", u->sink_input->sink->name, (unsigned long) new_rate);
 223
 224     pa_core_rttime_restart(u->core, u->time_event, pa_rtclock_now() + u->adjust_time);
 225 }
 226
 227 /* Called from main context */
 228 static void time_callback(pa_mainloop_api *a, pa_time_event *e, const struct timeval *t, void *userdata) {
 229     struct userdata *u = userdata;
 230
 231     pa_assert(u);
 232     pa_assert(a);
 233     pa_assert(u->time_event == e);
 234
 235     adjust_rates(u);
 236 }
 237
 238 /* Called from main context */
 239 static void enable_adjust_timer(struct userdata *u, bool enable) {
 240     if (enable) {
 241         if (u->time_event || u->adjust_time <= 0)
 242             return;
 243
 244         u->time_event = pa_core_rttime_new(u->module->core, pa_rtclock_now() + u->adjust_time, time_callback, u);
 245     } else {
 246         if (!u->time_event)
 247             return;
 248
 249         u->core->mainloop->time_free(u->time_event);
 250         u->time_event = NULL;
 251     }
 252 }
 253
 254 /* Called from main context */
 255 static void update_adjust_timer(struct userdata *u) {
 256     if (u->sink_input->state == PA_SINK_INPUT_CORKED || u->source_output->state == PA_SOURCE_OUTPUT_CORKED)
 257         enable_adjust_timer(u, false);
 258     else
 259         enable_adjust_timer(u, true);
 260 }
 261
 262 /* Called from input thread context */
 263 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
 264     struct userdata *u;
 265     pa_memchunk copy;
 266
 267     pa_source_output_assert_ref(o);
 268     pa_source_output_assert_io_context(o);
 269     pa_assert_se(u = o->userdata);
 270
 271     if (u->skip > chunk->length) {
 272         u->skip -= chunk->length;
 273         return;
 274     }
 275
 276     if (u->skip > 0) {
 277         copy = *chunk;
 278         copy.index += u->skip;
 279         copy.length -= u->skip;
 280         u->skip = 0;
 281
 282         chunk = &copy;
 283     }
 284
 285     pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->sink_input), SINK_INPUT_MESSAGE_POST, NULL, 0, chunk, NULL);
 286     u->send_counter += (int64_t) chunk->length;
 287 }
 288
 289 /* Called from input thread context */
 290 static void source_output_process_rewind_cb(pa_source_output *o, size_t nbytes) {
 291     struct userdata *u;
 292
 293     pa_source_output_assert_ref(o);
 294     pa_source_output_assert_io_context(o);
 295     pa_assert_se(u = o->userdata);
 296
 297     pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->sink_input), SINK_INPUT_MESSAGE_REWIND, NULL, (int64_t) nbytes, NULL, NULL);
 298     u->send_counter -= (int64_t) nbytes;
 299 }
 300
 301 /* Called from output thread context */
 302 static int source_output_process_msg_cb(pa_msgobject *obj, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 303     struct userdata *u = PA_SOURCE_OUTPUT(obj)->userdata;
 304
 305     switch (code) {
 306
 307         case SOURCE_OUTPUT_MESSAGE_LATENCY_SNAPSHOT: {
 308             size_t length;
 309
 310             length = pa_memblockq_get_length(u->source_output->thread_info.delay_memblockq);
 311
 312             u->latency_snapshot.send_counter = u->send_counter;
 313             u->latency_snapshot.source_output_buffer = u->source_output->thread_info.resampler ? pa_resampler_result(u->source_output->thread_info.resampler, length) : length;
 314             u->latency_snapshot.source_latency = pa_source_get_latency_within_thread(u->source_output->source);
 315
 316             return 0;
 317         }
 318     }
 319
 320     return pa_source_output_process_msg(obj, code, data, offset, chunk);
 321 }
 322
 323 /* Called from output thread context */
 324 static void source_output_attach_cb(pa_source_output *o) {
 325     struct userdata *u;
 326
 327     pa_source_output_assert_ref(o);
 328     pa_source_output_assert_io_context(o);
 329     pa_assert_se(u = o->userdata);
 330
 331     u->rtpoll_item_write = pa_rtpoll_item_new_asyncmsgq_write(
 332             o->source->thread_info.rtpoll,
 333             PA_RTPOLL_LATE,
 334             u->asyncmsgq);
 335 }
 336
 337 /* Called from output thread context */
 338 static void source_output_detach_cb(pa_source_output *o) {
 339     struct userdata *u;
 340
 341     pa_source_output_assert_ref(o);
 342     pa_source_output_assert_io_context(o);
 343     pa_assert_se(u = o->userdata);
 344
 345     if (u->rtpoll_item_write) {
 346         pa_rtpoll_item_free(u->rtpoll_item_write);
 347         u->rtpoll_item_write = NULL;
 348     }
 349 }
 350
 351 /* Called from output thread context */
 352 static void source_output_state_change_cb(pa_source_output *o, pa_source_output_state_t state) {
 353     struct userdata *u;
 354
 355     pa_source_output_assert_ref(o);
 356     pa_source_output_assert_io_context(o);
 357     pa_assert_se(u = o->userdata);
 358
 359     if (PA_SOURCE_OUTPUT_IS_LINKED(state) && o->thread_info.state == PA_SOURCE_OUTPUT_INIT) {
 360
 361         u->skip = pa_usec_to_bytes(PA_CLIP_SUB(pa_source_get_latency_within_thread(o->source),
 362                                                u->latency),
 363                                    &o->sample_spec);
 364
 365         pa_log_info("Skipping %lu bytes", (unsigned long) u->skip);
 366     }
 367 }
 368
 369 /* Called from main thread */
 370 static void source_output_kill_cb(pa_source_output *o) {
 371     struct userdata *u;
 372
 373     pa_source_output_assert_ref(o);
 374     pa_assert_ctl_context();
 375     pa_assert_se(u = o->userdata);
 376
 377     teardown(u);
 378     pa_module_unload_request(u->module, TRUE);
 379 }
 380
 381 /* Called from main thread */
 382 static pa_bool_t source_output_may_move_to_cb(pa_source_output *o, pa_source *dest) {
 383     struct userdata *u;
 384
 385     pa_source_output_assert_ref(o);
 386     pa_assert_ctl_context();
 387     pa_assert_se(u = o->userdata);
 388
 389     return dest != u->sink_input->sink->monitor_source;
 390 }
 391
 392 /* Called from main thread */
 393 static void source_output_moving_cb(pa_source_output *o, pa_source *dest) {
 394     pa_proplist *p;
 395     const char *n;
 396     struct userdata *u;
 397
 398     if (!dest)
 399         return;
 400
 401     pa_source_output_assert_ref(o);
 402     pa_assert_ctl_context();
 403     pa_assert_se(u = o->userdata);
 404
 405     p = pa_proplist_new();
 406     pa_proplist_setf(p, PA_PROP_MEDIA_NAME, "Loopback of %s", pa_strnull(pa_proplist_gets(dest->proplist, PA_PROP_DEVICE_DESCRIPTION)));
 407
 408     if ((n = pa_proplist_gets(dest->proplist, PA_PROP_DEVICE_ICON_NAME)))
 409         pa_proplist_sets(p, PA_PROP_MEDIA_ICON_NAME, n);
 410
 411     pa_sink_input_update_proplist(u->sink_input, PA_UPDATE_REPLACE, p);
 412     pa_proplist_free(p);
 413 }
 414
 415 /* Called from main thread */
 416 static void source_output_suspend_cb(pa_source_output *o, pa_bool_t suspended) {
 417     struct userdata *u;
 418
 419     pa_source_output_assert_ref(o);
 420     pa_assert_ctl_context();
 421     pa_assert_se(u = o->userdata);
 422
 423     pa_sink_input_cork(u->sink_input, suspended);
 424
 425     update_adjust_timer(u);
 426 }
 427
 428 /* Called from output thread context */
 429 static void update_min_memblockq_length(struct userdata *u) {
 430     size_t length;
 431
 432     pa_assert(u);
 433     pa_sink_input_assert_io_context(u->sink_input);
 434
 435     length = pa_memblockq_get_length(u->memblockq);
 436
 437     if (u->min_memblockq_length == (size_t) -1 ||
 438         length < u->min_memblockq_length)
 439         u->min_memblockq_length = length;
 440 }
 441
 442 /* Called from output thread context */
 443 static int sink_input_pop_cb(pa_sink_input *i, size_t nbytes, pa_memchunk *chunk) {
 444     struct userdata *u;
 445
 446     pa_sink_input_assert_ref(i);
 447     pa_sink_input_assert_io_context(i);
 448     pa_assert_se(u = i->userdata);
 449     pa_assert(chunk);
 450
 451     u->in_pop = TRUE;
 452     while (pa_asyncmsgq_process_one(u->asyncmsgq) > 0)
 453         ;
 454     u->in_pop = FALSE;
 455
 456     if (pa_memblockq_peek(u->memblockq, chunk) < 0) {
 457         pa_log_info("Could not peek into queue");
 458         return -1;
 459     }
 460
 461     chunk->length = PA_MIN(chunk->length, nbytes);
 462     pa_memblockq_drop(u->memblockq, chunk->length);
 463
 464     update_min_memblockq_length(u);
 465
 466     return 0;
 467 }
 468
 469 /* Called from output thread context */
 470 static void sink_input_process_rewind_cb(pa_sink_input *i, size_t nbytes) {
 471     struct userdata *u;
 472
 473     pa_sink_input_assert_ref(i);
 474     pa_sink_input_assert_io_context(i);
 475     pa_assert_se(u = i->userdata);
 476
 477     pa_memblockq_rewind(u->memblockq, nbytes);
 478 }
 479
 480 /* Called from output thread context */
 481 static int sink_input_process_msg_cb(pa_msgobject *obj, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 482     struct userdata *u = PA_SINK_INPUT(obj)->userdata;
 483
 484     switch (code) {
 485
 486         case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
 487             pa_usec_t *r = data;
 488
 489             pa_sink_input_assert_io_context(u->sink_input);
 490
 491             *r = pa_bytes_to_usec(pa_memblockq_get_length(u->memblockq), &u->sink_input->sample_spec);
 492
 493             /* Fall through, the default handler will add in the extra
 494              * latency added by the resampler */
 495             break;
 496         }
 497
 498         case SINK_INPUT_MESSAGE_POST:
 499
 500             pa_sink_input_assert_io_context(u->sink_input);
 501
 502             if (PA_SINK_IS_OPENED(u->sink_input->sink->thread_info.state))
 503                 pa_memblockq_push_align(u->memblockq, chunk);
 504             else
 505                 pa_memblockq_flush_write(u->memblockq, TRUE);
 506
 507             update_min_memblockq_length(u);
 508
 509             /* Is this the end of an underrun? Then let's start things
 510              * right-away */
 511             if (!u->in_pop &&
 512                 u->sink_input->thread_info.underrun_for > 0 &&
 513                 pa_memblockq_is_readable(u->memblockq)) {
 514
 515                 pa_log_debug("Requesting rewind due to end of underrun.");
 516                 pa_sink_input_request_rewind(u->sink_input,
 517                                              (size_t) (u->sink_input->thread_info.underrun_for == (size_t) -1 ? 0 : u->sink_input->thread_info.underrun_for),
 518                                              FALSE, TRUE, FALSE);
 519             }
 520
 521             u->recv_counter += (int64_t) chunk->length;
 522
 523             return 0;
 524
 525         case SINK_INPUT_MESSAGE_REWIND:
 526
 527             pa_sink_input_assert_io_context(u->sink_input);
 528
 529             if (PA_SINK_IS_OPENED(u->sink_input->sink->thread_info.state))
 530                 pa_memblockq_seek(u->memblockq, -offset, PA_SEEK_RELATIVE, TRUE);
 531             else
 532                 pa_memblockq_flush_write(u->memblockq, TRUE);
 533
 534             u->recv_counter -= offset;
 535
 536             update_min_memblockq_length(u);
 537
 538             return 0;
 539
 540         case SINK_INPUT_MESSAGE_LATENCY_SNAPSHOT: {
 541             size_t length;
 542
 543             update_min_memblockq_length(u);
 544
 545             length = pa_memblockq_get_length(u->sink_input->thread_info.render_memblockq);
 546
 547             u->latency_snapshot.recv_counter = u->recv_counter;
 548             u->latency_snapshot.sink_input_buffer =
 549                 pa_memblockq_get_length(u->memblockq) +
 550                 (u->sink_input->thread_info.resampler ? pa_resampler_request(u->sink_input->thread_info.resampler, length) : length);
 551             u->latency_snapshot.sink_latency = pa_sink_get_latency_within_thread(u->sink_input->sink);
 552
 553             u->latency_snapshot.max_request = pa_sink_input_get_max_request(u->sink_input);
 554
 555             u->latency_snapshot.min_memblockq_length = u->min_memblockq_length;
 556             u->min_memblockq_length = (size_t) -1;
 557
 558             return 0;
 559         }
 560
 561         case SINK_INPUT_MESSAGE_MAX_REQUEST_CHANGED: {
 562             /* This message is sent from the IO thread to the main
 563              * thread! So don't be confused. All the user cases above
 564              * are executed in thread context, but this one is not! */
 565
 566             pa_assert_ctl_context();
 567
 568             if (u->time_event)
 569                 adjust_rates(u);
 570             return 0;
 571         }
 572     }
 573
 574     return pa_sink_input_process_msg(obj, code, data, offset, chunk);
 575 }
 576
 577 /* Called from output thread context */
 578 static void sink_input_attach_cb(pa_sink_input *i) {
 579     struct userdata *u;
 580
 581     pa_sink_input_assert_ref(i);
 582     pa_sink_input_assert_io_context(i);
 583     pa_assert_se(u = i->userdata);
 584
 585     u->rtpoll_item_read = pa_rtpoll_item_new_asyncmsgq_read(
 586             i->sink->thread_info.rtpoll,
 587             PA_RTPOLL_LATE,
 588             u->asyncmsgq);
 589
 590     pa_memblockq_set_prebuf(u->memblockq, pa_sink_input_get_max_request(i)*2);
 591     pa_memblockq_set_maxrewind(u->memblockq, pa_sink_input_get_max_rewind(i));
 592
 593     u->min_memblockq_length = (size_t) -1;
 594 }
 595
 596 /* Called from output thread context */
 597 static void sink_input_detach_cb(pa_sink_input *i) {
 598     struct userdata *u;
 599
 600     pa_sink_input_assert_ref(i);
 601     pa_sink_input_assert_io_context(i);
 602     pa_assert_se(u = i->userdata);
 603
 604     if (u->rtpoll_item_read) {
 605         pa_rtpoll_item_free(u->rtpoll_item_read);
 606         u->rtpoll_item_read = NULL;
 607     }
 608 }
 609
 610 /* Called from output thread context */
 611 static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) {
 612     struct userdata *u;
 613
 614     pa_sink_input_assert_ref(i);
 615     pa_sink_input_assert_io_context(i);
 616     pa_assert_se(u = i->userdata);
 617
 618     pa_memblockq_set_maxrewind(u->memblockq, nbytes);
 619 }
 620
 621 /* Called from output thread context */
 622 static void sink_input_update_max_request_cb(pa_sink_input *i, size_t nbytes) {
 623     struct userdata *u;
 624
 625     pa_sink_input_assert_ref(i);
 626     pa_sink_input_assert_io_context(i);
 627     pa_assert_se(u = i->userdata);
 628
 629     pa_memblockq_set_prebuf(u->memblockq, nbytes*2);
 630     pa_log_info("Max request changed");
 631     pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(u->sink_input), SINK_INPUT_MESSAGE_MAX_REQUEST_CHANGED, NULL, 0, NULL, NULL);
 632 }
 633
 634 /* Called from main thread */
 635 static void sink_input_kill_cb(pa_sink_input *i) {
 636     struct userdata *u;
 637
 638     pa_sink_input_assert_ref(i);
 639     pa_assert_ctl_context();
 640     pa_assert_se(u = i->userdata);
 641
 642     teardown(u);
 643     pa_module_unload_request(u->module, TRUE);
 644 }
 645
 646 /* Called from main thread */
 647 static void sink_input_moving_cb(pa_sink_input *i, pa_sink *dest) {
 648     struct userdata *u;
 649     pa_proplist *p;
 650     const char *n;
 651
 652     if (!dest)
 653         return;
 654
 655     pa_sink_input_assert_ref(i);
 656     pa_assert_ctl_context();
 657     pa_assert_se(u = i->userdata);
 658
 659     p = pa_proplist_new();
 660     pa_proplist_setf(p, PA_PROP_MEDIA_NAME, "Loopback to %s", pa_strnull(pa_proplist_gets(dest->proplist, PA_PROP_DEVICE_DESCRIPTION)));
 661
 662     if ((n = pa_proplist_gets(dest->proplist, PA_PROP_DEVICE_ICON_NAME)))
 663         pa_proplist_sets(p, PA_PROP_MEDIA_ICON_NAME, n);
 664
 665     pa_source_output_update_proplist(u->source_output, PA_UPDATE_REPLACE, p);
 666     pa_proplist_free(p);
 667 }
 668
 669 /* Called from main thread */
 670 static pa_bool_t sink_input_may_move_to_cb(pa_sink_input *i, pa_sink *dest) {
 671     struct userdata *u;
 672
 673     pa_sink_input_assert_ref(i);
 674     pa_assert_ctl_context();
 675     pa_assert_se(u = i->userdata);
 676
 677     if (!u->source_output->source->monitor_of)
 678         return TRUE;
 679
 680     return dest != u->source_output->source->monitor_of;
 681 }
 682
 683 /* Called from main thread */
 684 static void sink_input_suspend_cb(pa_sink_input *i, pa_bool_t suspended) {
 685     struct userdata *u;
 686
 687     pa_sink_input_assert_ref(i);
 688     pa_assert_ctl_context();
 689     pa_assert_se(u = i->userdata);
 690
 691     pa_source_output_cork(u->source_output, suspended);
 692
 693     update_adjust_timer(u);
 694 }
 695
 696 int pa__init(pa_module *m) {
 697     pa_modargs *ma = NULL;
 698     struct userdata *u;
 699     pa_sink *sink = NULL;
 700     pa_sink_input_new_data sink_input_data;
 701     pa_bool_t sink_dont_move;
 702     pa_source *source = NULL;
 703     pa_source_output_new_data source_output_data;
 704     pa_bool_t source_dont_move;
 705     uint32_t latency_msec;
 706     pa_sample_spec ss;
 707     pa_channel_map map;
 708     bool format_set = false;
 709     bool rate_set = false;
 710     bool channels_set = false;
 711     pa_memchunk silence;
 712     uint32_t adjust_time_sec;
 713     const char *n;
 714     pa_bool_t remix = TRUE;
 715
 716     pa_assert(m);
 717
 718     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
 719         pa_log("Failed to parse module arguments");
 720         goto fail;
 721     }
 722
 723     n = pa_modargs_get_value(ma, "source", NULL);
 724     if (n && !(source = pa_namereg_get(m->core, n, PA_NAMEREG_SOURCE))) {
 725         pa_log("No such source.");
 726         goto fail;
 727     }
 728
 729     n = pa_modargs_get_value(ma, "sink", NULL);
 730     if (n && !(sink = pa_namereg_get(m->core, n, PA_NAMEREG_SINK))) {
 731         pa_log("No such sink.");
 732         goto fail;
 733     }
 734
 735     if (pa_modargs_get_value_boolean(ma, "remix", &remix) < 0) {
 736         pa_log("Invalid boolean remix parameter");
 737         goto fail;
 738     }
 739
 740     if (sink) {
 741         ss = sink->sample_spec;
 742         map = sink->channel_map;
 743         format_set = true;
 744         rate_set = true;
 745         channels_set = true;
 746     } else if (source) {
 747         ss = source->sample_spec;
 748         map = source->channel_map;
 749         format_set = true;
 750         rate_set = true;
 751         channels_set = true;
 752     } else {
 753         /* FIXME: Dummy stream format, needed because pa_sink_input_new()
 754          * requires valid sample spec and channel map even when all the FIX_*
 755          * stream flags are specified. pa_sink_input_new() should be changed
 756          * to ignore the sample spec and channel map when the FIX_* flags are
 757          * present. */
 758         ss.format = PA_SAMPLE_U8;
 759         ss.rate = 8000;
 760         ss.channels = 1;
 761         map.channels = 1;
 762         map.map[0] = PA_CHANNEL_POSITION_MONO;
 763     }
 764
 765     if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0) {
 766         pa_log("Invalid sample format specification or channel map");
 767         goto fail;
 768     }
 769
 770     if (pa_modargs_get_value(ma, "format", NULL))
 771         format_set = true;
 772
 773     if (pa_modargs_get_value(ma, "rate", NULL))
 774         rate_set = true;
 775
 776     if (pa_modargs_get_value(ma, "channels", NULL) || pa_modargs_get_value(ma, "channel_map", NULL))
 777         channels_set = true;
 778
 779     latency_msec = DEFAULT_LATENCY_MSEC;
 780     if (pa_modargs_get_value_u32(ma, "latency_msec", &latency_msec) < 0 || latency_msec < 1 || latency_msec > 2000) {
 781         pa_log("Invalid latency specification");
 782         goto fail;
 783     }
 784
 785     m->userdata = u = pa_xnew0(struct userdata, 1);
 786     u->core = m->core;
 787     u->module = m;
 788     u->latency = (pa_usec_t) latency_msec * PA_USEC_PER_MSEC;
 789
 790     adjust_time_sec = DEFAULT_ADJUST_TIME_USEC / PA_USEC_PER_SEC;
 791     if (pa_modargs_get_value_u32(ma, "adjust_time", &adjust_time_sec) < 0) {
 792         pa_log("Failed to parse adjust_time value");
 793         goto fail;
 794     }
 795
 796     if (adjust_time_sec != DEFAULT_ADJUST_TIME_USEC / PA_USEC_PER_SEC)
 797         u->adjust_time = adjust_time_sec * PA_USEC_PER_SEC;
 798     else
 799         u->adjust_time = DEFAULT_ADJUST_TIME_USEC;
 800
 801     pa_sink_input_new_data_init(&sink_input_data);
 802     sink_input_data.driver = __FILE__;
 803     sink_input_data.module = m;
 804
 805     if (sink)
 806         pa_sink_input_new_data_set_sink(&sink_input_data, sink, FALSE);
 807
 808     if (pa_modargs_get_proplist(ma, "sink_input_properties", sink_input_data.proplist, PA_UPDATE_REPLACE) < 0) {
 809         pa_log("Failed to parse the sink_input_properties value.");
 810         pa_sink_input_new_data_done(&sink_input_data);
 811         goto fail;
 812     }
 813
 814     if (!pa_proplist_contains(sink_input_data.proplist, PA_PROP_MEDIA_ROLE))
 815         pa_proplist_sets(sink_input_data.proplist, PA_PROP_MEDIA_ROLE, "abstract");
 816
 817     pa_sink_input_new_data_set_sample_spec(&sink_input_data, &ss);
 818     pa_sink_input_new_data_set_channel_map(&sink_input_data, &map);
 819     sink_input_data.flags = PA_SINK_INPUT_VARIABLE_RATE | PA_SINK_INPUT_START_CORKED;
 820
 821     if (!remix)
 822         sink_input_data.flags |= PA_SINK_INPUT_NO_REMIX;
 823
 824     if (!format_set)
 825         sink_input_data.flags |= PA_SINK_INPUT_FIX_FORMAT;
 826
 827     if (!rate_set)
 828         sink_input_data.flags |= PA_SINK_INPUT_FIX_RATE;
 829
 830     if (!channels_set)
 831         sink_input_data.flags |= PA_SINK_INPUT_FIX_CHANNELS;
 832
 833     sink_dont_move = FALSE;
 834     if (pa_modargs_get_value_boolean(ma, "sink_dont_move", &sink_dont_move) < 0) {
 835         pa_log("sink_dont_move= expects a boolean argument.");
 836         goto fail;
 837     }
 838
 839     if (sink_dont_move)
 840         sink_input_data.flags |= PA_SINK_INPUT_DONT_MOVE;
 841
 842     pa_sink_input_new(&u->sink_input, m->core, &sink_input_data);
 843     pa_sink_input_new_data_done(&sink_input_data);
 844
 845     if (!u->sink_input)
 846         goto fail;
 847
 848     u->sink_input->parent.process_msg = sink_input_process_msg_cb;
 849     u->sink_input->pop = sink_input_pop_cb;
 850     u->sink_input->process_rewind = sink_input_process_rewind_cb;
 851     u->sink_input->kill = sink_input_kill_cb;
 852     u->sink_input->attach = sink_input_attach_cb;
 853     u->sink_input->detach = sink_input_detach_cb;
 854     u->sink_input->update_max_rewind = sink_input_update_max_rewind_cb;
 855     u->sink_input->update_max_request = sink_input_update_max_request_cb;
 856     u->sink_input->may_move_to = sink_input_may_move_to_cb;
 857     u->sink_input->moving = sink_input_moving_cb;
 858     u->sink_input->suspend = sink_input_suspend_cb;
 859     u->sink_input->userdata = u;
 860
 861     pa_sink_input_set_requested_latency(u->sink_input, u->latency/3);
 862
 863     pa_source_output_new_data_init(&source_output_data);
 864     source_output_data.driver = __FILE__;
 865     source_output_data.module = m;
 866     if (source)
 867         pa_source_output_new_data_set_source(&source_output_data, source, FALSE);
 868
 869     if (pa_modargs_get_proplist(ma, "source_output_properties", source_output_data.proplist, PA_UPDATE_REPLACE) < 0) {
 870         pa_log("Failed to parse the source_output_properties value.");
 871         pa_source_output_new_data_done(&source_output_data);
 872         goto fail;
 873     }
 874
 875     if (!pa_proplist_contains(source_output_data.proplist, PA_PROP_MEDIA_ROLE))
 876         pa_proplist_sets(source_output_data.proplist, PA_PROP_MEDIA_ROLE, "abstract");
 877
 878     pa_source_output_new_data_set_sample_spec(&source_output_data, &ss);
 879     pa_source_output_new_data_set_channel_map(&source_output_data, &map);
 880     source_output_data.flags = PA_SOURCE_OUTPUT_START_CORKED;
 881
 882     if (!remix)
 883         source_output_data.flags |= PA_SOURCE_OUTPUT_NO_REMIX;
 884
 885     if (!format_set)
 886         source_output_data.flags |= PA_SOURCE_OUTPUT_FIX_FORMAT;
 887
 888     if (!rate_set)
 889         source_output_data.flags |= PA_SOURCE_OUTPUT_FIX_RATE;
 890
 891     if (!channels_set)
 892         source_output_data.flags |= PA_SOURCE_OUTPUT_FIX_CHANNELS;
 893
 894     source_dont_move = FALSE;
 895     if (pa_modargs_get_value_boolean(ma, "source_dont_move", &source_dont_move) < 0) {
 896         pa_log("source_dont_move= expects a boolean argument.");
 897         goto fail;
 898     }
 899
 900     if (source_dont_move)
 901         source_output_data.flags |= PA_SOURCE_OUTPUT_DONT_MOVE;
 902
 903     pa_source_output_new(&u->source_output, m->core, &source_output_data);
 904     pa_source_output_new_data_done(&source_output_data);
 905
 906     if (!u->source_output)
 907         goto fail;
 908
 909     u->source_output->parent.process_msg = source_output_process_msg_cb;
 910     u->source_output->push = source_output_push_cb;
 911     u->source_output->process_rewind = source_output_process_rewind_cb;
 912     u->source_output->kill = source_output_kill_cb;
 913     u->source_output->attach = source_output_attach_cb;
 914     u->source_output->detach = source_output_detach_cb;
 915     u->source_output->state_change = source_output_state_change_cb;
 916     u->source_output->may_move_to = source_output_may_move_to_cb;
 917     u->source_output->moving = source_output_moving_cb;
 918     u->source_output->suspend = source_output_suspend_cb;
 919     u->source_output->userdata = u;
 920
 921     pa_source_output_set_requested_latency(u->source_output, u->latency/3);
 922
 923     pa_sink_input_get_silence(u->sink_input, &silence);
 924     u->memblockq = pa_memblockq_new(
 925             "module-loopback memblockq",
 926             0,                      /* idx */
 927             MEMBLOCKQ_MAXLENGTH,    /* maxlength */
 928             MEMBLOCKQ_MAXLENGTH,    /* tlength */
 929             &ss,                    /* sample_spec */
 930             0,                      /* prebuf */
 931             0,                      /* minreq */
 932             0,                      /* maxrewind */
 933             &silence);              /* silence frame */
 934     pa_memblock_unref(silence.memblock);
 935
 936     u->asyncmsgq = pa_asyncmsgq_new(0);
 937
 938     if (!pa_proplist_contains(u->source_output->proplist, PA_PROP_MEDIA_NAME))
 939         pa_proplist_setf(u->source_output->proplist, PA_PROP_MEDIA_NAME, "Loopback to %s",
 940                          pa_strnull(pa_proplist_gets(u->sink_input->sink->proplist, PA_PROP_DEVICE_DESCRIPTION)));
 941
 942     if (!pa_proplist_contains(u->source_output->proplist, PA_PROP_MEDIA_ICON_NAME)
 943             && (n = pa_proplist_gets(u->sink_input->sink->proplist, PA_PROP_DEVICE_ICON_NAME)))
 944         pa_proplist_sets(u->source_output->proplist, PA_PROP_MEDIA_ICON_NAME, n);
 945
 946     if (!pa_proplist_contains(u->sink_input->proplist, PA_PROP_MEDIA_NAME))
 947         pa_proplist_setf(u->sink_input->proplist, PA_PROP_MEDIA_NAME, "Loopback from %s",
 948                          pa_strnull(pa_proplist_gets(u->source_output->source->proplist, PA_PROP_DEVICE_DESCRIPTION)));
 949
 950     if (source && !pa_proplist_contains(u->sink_input->proplist, PA_PROP_MEDIA_ICON_NAME)
 951             && (n = pa_proplist_gets(u->source_output->source->proplist, PA_PROP_DEVICE_ICON_NAME)))
 952         pa_proplist_sets(u->sink_input->proplist, PA_PROP_MEDIA_ICON_NAME, n);
 953
 954     pa_sink_input_put(u->sink_input);
 955     pa_source_output_put(u->source_output);
 956
 957     if (pa_source_get_state(u->source_output->source) != PA_SOURCE_SUSPENDED)
 958             pa_sink_input_cork(u->sink_input, FALSE);
 959
 960     if (pa_sink_get_state(u->sink_input->sink) != PA_SINK_SUSPENDED)
 961             pa_source_output_cork(u->source_output, FALSE);
 962
 963     update_adjust_timer(u);
 964
 965     pa_modargs_free(ma);
 966     return 0;
 967
 968 fail:
 969     if (ma)
 970         pa_modargs_free(ma);
 971
 972     pa__done(m);
 973
 974     return -1;
 975 }
 976
 977 void pa__done(pa_module*m) {
 978     struct userdata *u;
 979
 980     pa_assert(m);
 981
 982     if (!(u = m->userdata))
 983         return;
 984
 985     teardown(u);
 986
 987     if (u->memblockq)
 988         pa_memblockq_free(u->memblockq);
 989
 990     if (u->asyncmsgq)
 991         pa_asyncmsgq_unref(u->asyncmsgq);
 992
 993     pa_xfree(u);
 994 }