src/pulsecore/source.c

   1 /***
   2   This file is part of PulseAudio.
   3
   4   Copyright 2004-2006 Lennart Poettering
   5   Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
   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 #include <stdlib.h>
  29 #include <string.h>
  30
  31 #include <pulse/utf8.h>
  32 #include <pulse/xmalloc.h>
  33 #include <pulse/timeval.h>
  34 #include <pulse/util.h>
  35
  36 #include <pulsecore/core-util.h>
  37 #include <pulsecore/source-output.h>
  38 #include <pulsecore/namereg.h>
  39 #include <pulsecore/core-subscribe.h>
  40 #include <pulsecore/log.h>
  41 #include <pulsecore/sample-util.h>
  42
  43 #include "source.h"
  44
  45 #define ABSOLUTE_MIN_LATENCY (500)
  46 #define ABSOLUTE_MAX_LATENCY (10*PA_USEC_PER_SEC)
  47 #define DEFAULT_FIXED_LATENCY (250*PA_USEC_PER_MSEC)
  48
  49 PA_DEFINE_PUBLIC_CLASS(pa_source, pa_msgobject);
  50
  51 static void source_free(pa_object *o);
  52
  53 pa_source_new_data* pa_source_new_data_init(pa_source_new_data *data) {
  54     pa_assert(data);
  55
  56     pa_zero(*data);
  57     data->proplist = pa_proplist_new();
  58
  59     return data;
  60 }
  61
  62 void pa_source_new_data_set_name(pa_source_new_data *data, const char *name) {
  63     pa_assert(data);
  64
  65     pa_xfree(data->name);
  66     data->name = pa_xstrdup(name);
  67 }
  68
  69 void pa_source_new_data_set_sample_spec(pa_source_new_data *data, const pa_sample_spec *spec) {
  70     pa_assert(data);
  71
  72     if ((data->sample_spec_is_set = !!spec))
  73         data->sample_spec = *spec;
  74 }
  75
  76 void pa_source_new_data_set_channel_map(pa_source_new_data *data, const pa_channel_map *map) {
  77     pa_assert(data);
  78
  79     if ((data->channel_map_is_set = !!map))
  80         data->channel_map = *map;
  81 }
  82
  83 void pa_source_new_data_set_volume(pa_source_new_data *data, const pa_cvolume *volume) {
  84     pa_assert(data);
  85
  86     if ((data->volume_is_set = !!volume))
  87         data->volume = *volume;
  88 }
  89
  90 void pa_source_new_data_set_muted(pa_source_new_data *data, pa_bool_t mute) {
  91     pa_assert(data);
  92
  93     data->muted_is_set = TRUE;
  94     data->muted = !!mute;
  95 }
  96
  97 void pa_source_new_data_set_port(pa_source_new_data *data, const char *port) {
  98     pa_assert(data);
  99
 100     pa_xfree(data->active_port);
 101     data->active_port = pa_xstrdup(port);
 102 }
 103
 104 void pa_source_new_data_done(pa_source_new_data *data) {
 105     pa_assert(data);
 106
 107     pa_proplist_free(data->proplist);
 108
 109     if (data->ports) {
 110         pa_device_port *p;
 111
 112         while ((p = pa_hashmap_steal_first(data->ports)))
 113             pa_device_port_free(p);
 114
 115         pa_hashmap_free(data->ports, NULL, NULL);
 116     }
 117
 118     pa_xfree(data->name);
 119     pa_xfree(data->active_port);
 120 }
 121
 122 /* Called from main context */
 123 static void reset_callbacks(pa_source *s) {
 124     pa_assert(s);
 125
 126     s->set_state = NULL;
 127     s->get_volume = NULL;
 128     s->set_volume = NULL;
 129     s->get_mute = NULL;
 130     s->set_mute = NULL;
 131     s->update_requested_latency = NULL;
 132     s->set_port = NULL;
 133 }
 134
 135 /* Called from main context */
 136 pa_source* pa_source_new(
 137         pa_core *core,
 138         pa_source_new_data *data,
 139         pa_source_flags_t flags) {
 140
 141     pa_source *s;
 142     const char *name;
 143     char st[PA_SAMPLE_SPEC_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX];
 144     char *pt;
 145
 146     pa_assert(core);
 147     pa_assert(data);
 148     pa_assert(data->name);
 149     pa_assert_ctl_context();
 150
 151     s = pa_msgobject_new(pa_source);
 152
 153     if (!(name = pa_namereg_register(core, data->name, PA_NAMEREG_SOURCE, s, data->namereg_fail))) {
 154         pa_log_debug("Failed to register name %s.", data->name);
 155         pa_xfree(s);
 156         return NULL;
 157     }
 158
 159     pa_source_new_data_set_name(data, name);
 160
 161     if (pa_hook_fire(&core->hooks[PA_CORE_HOOK_SOURCE_NEW], data) < 0) {
 162         pa_xfree(s);
 163         pa_namereg_unregister(core, name);
 164         return NULL;
 165     }
 166
 167     /* FIXME, need to free s here on failure */
 168
 169     pa_return_null_if_fail(!data->driver || pa_utf8_valid(data->driver));
 170     pa_return_null_if_fail(data->name && pa_utf8_valid(data->name) && data->name[0]);
 171
 172     pa_return_null_if_fail(data->sample_spec_is_set && pa_sample_spec_valid(&data->sample_spec));
 173
 174     if (!data->channel_map_is_set)
 175         pa_return_null_if_fail(pa_channel_map_init_auto(&data->channel_map, data->sample_spec.channels, PA_CHANNEL_MAP_DEFAULT));
 176
 177     pa_return_null_if_fail(pa_channel_map_valid(&data->channel_map));
 178     pa_return_null_if_fail(data->channel_map.channels == data->sample_spec.channels);
 179
 180     if (!data->volume_is_set)
 181         pa_cvolume_reset(&data->volume, data->sample_spec.channels);
 182
 183     pa_return_null_if_fail(pa_cvolume_valid(&data->volume));
 184     pa_return_null_if_fail(pa_cvolume_compatible(&data->volume, &data->sample_spec));
 185
 186     if (!data->muted_is_set)
 187         data->muted = FALSE;
 188
 189     if (data->card)
 190         pa_proplist_update(data->proplist, PA_UPDATE_MERGE, data->card->proplist);
 191
 192     pa_device_init_description(data->proplist);
 193     pa_device_init_icon(data->proplist, FALSE);
 194     pa_device_init_intended_roles(data->proplist);
 195
 196     if (pa_hook_fire(&core->hooks[PA_CORE_HOOK_SOURCE_FIXATE], data) < 0) {
 197         pa_xfree(s);
 198         pa_namereg_unregister(core, name);
 199         return NULL;
 200     }
 201
 202     s->parent.parent.free = source_free;
 203     s->parent.process_msg = pa_source_process_msg;
 204
 205     s->core = core;
 206     s->state = PA_SOURCE_INIT;
 207     s->flags = flags;
 208     s->priority = 0;
 209     s->suspend_cause = 0;
 210     s->name = pa_xstrdup(name);
 211     s->proplist = pa_proplist_copy(data->proplist);
 212     s->driver = pa_xstrdup(pa_path_get_filename(data->driver));
 213     s->module = data->module;
 214     s->card = data->card;
 215
 216     s->priority = pa_device_init_priority(s->proplist);
 217
 218     s->sample_spec = data->sample_spec;
 219     s->channel_map = data->channel_map;
 220
 221     s->outputs = pa_idxset_new(NULL, NULL);
 222     s->n_corked = 0;
 223     s->monitor_of = NULL;
 224
 225     s->volume = data->volume;
 226     pa_cvolume_reset(&s->soft_volume, s->sample_spec.channels);
 227     s->base_volume = PA_VOLUME_NORM;
 228     s->n_volume_steps = PA_VOLUME_NORM+1;
 229     s->muted = data->muted;
 230     s->refresh_volume = s->refresh_muted = FALSE;
 231
 232     reset_callbacks(s);
 233     s->userdata = NULL;
 234
 235     s->asyncmsgq = NULL;
 236
 237     /* As a minor optimization we just steal the list instead of
 238      * copying it here */
 239     s->ports = data->ports;
 240     data->ports = NULL;
 241
 242     s->active_port = NULL;
 243     s->save_port = FALSE;
 244
 245     if (data->active_port && s->ports)
 246         if ((s->active_port = pa_hashmap_get(s->ports, data->active_port)))
 247             s->save_port = data->save_port;
 248
 249     if (!s->active_port && s->ports) {
 250         void *state;
 251         pa_device_port *p;
 252
 253         PA_HASHMAP_FOREACH(p, s->ports, state)
 254             if (!s->active_port || p->priority > s->active_port->priority)
 255                 s->active_port = p;
 256     }
 257
 258     s->save_volume = data->save_volume;
 259     s->save_muted = data->save_muted;
 260
 261     pa_silence_memchunk_get(
 262             &core->silence_cache,
 263             core->mempool,
 264             &s->silence,
 265             &s->sample_spec,
 266             0);
 267
 268     s->thread_info.rtpoll = NULL;
 269     s->thread_info.outputs = pa_hashmap_new(pa_idxset_trivial_hash_func, pa_idxset_trivial_compare_func);
 270     s->thread_info.soft_volume = s->soft_volume;
 271     s->thread_info.soft_muted = s->muted;
 272     s->thread_info.state = s->state;
 273     s->thread_info.max_rewind = 0;
 274     s->thread_info.requested_latency_valid = FALSE;
 275     s->thread_info.requested_latency = 0;
 276     s->thread_info.min_latency = ABSOLUTE_MIN_LATENCY;
 277     s->thread_info.max_latency = ABSOLUTE_MAX_LATENCY;
 278     s->thread_info.fixed_latency = flags & PA_SOURCE_DYNAMIC_LATENCY ? 0 : DEFAULT_FIXED_LATENCY;
 279
 280     pa_assert_se(pa_idxset_put(core->sources, s, &s->index) >= 0);
 281
 282     if (s->card)
 283         pa_assert_se(pa_idxset_put(s->card->sources, s, NULL) >= 0);
 284
 285     pt = pa_proplist_to_string_sep(s->proplist, "\n    ");
 286     pa_log_info("Created source %u \"%s\" with sample spec %s and channel map %s\n    %s",
 287                 s->index,
 288                 s->name,
 289                 pa_sample_spec_snprint(st, sizeof(st), &s->sample_spec),
 290                 pa_channel_map_snprint(cm, sizeof(cm), &s->channel_map),
 291                 pt);
 292     pa_xfree(pt);
 293
 294     return s;
 295 }
 296
 297 /* Called from main context */
 298 static int source_set_state(pa_source *s, pa_source_state_t state) {
 299     int ret;
 300     pa_bool_t suspend_change;
 301     pa_source_state_t original_state;
 302
 303     pa_assert(s);
 304     pa_assert_ctl_context();
 305
 306     if (s->state == state)
 307         return 0;
 308
 309     original_state = s->state;
 310
 311     suspend_change =
 312         (original_state == PA_SOURCE_SUSPENDED && PA_SOURCE_IS_OPENED(state)) ||
 313         (PA_SOURCE_IS_OPENED(original_state) && state == PA_SOURCE_SUSPENDED);
 314
 315     if (s->set_state)
 316         if ((ret = s->set_state(s, state)) < 0)
 317             return ret;
 318
 319     if (s->asyncmsgq)
 320         if ((ret = pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL)) < 0) {
 321
 322             if (s->set_state)
 323                 s->set_state(s, original_state);
 324
 325             return ret;
 326         }
 327
 328     s->state = state;
 329
 330     if (state != PA_SOURCE_UNLINKED) { /* if we enter UNLINKED state pa_source_unlink() will fire the apropriate events */
 331         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_STATE_CHANGED], s);
 332         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE | PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 333     }
 334
 335     if (suspend_change) {
 336         pa_source_output *o;
 337         uint32_t idx;
 338
 339         /* We're suspending or resuming, tell everyone about it */
 340
 341         PA_IDXSET_FOREACH(o, s->outputs, idx)
 342             if (s->state == PA_SOURCE_SUSPENDED &&
 343                 (o->flags & PA_SOURCE_OUTPUT_KILL_ON_SUSPEND))
 344                 pa_source_output_kill(o);
 345             else if (o->suspend)
 346                 o->suspend(o, state == PA_SOURCE_SUSPENDED);
 347     }
 348
 349     return 0;
 350 }
 351
 352 /* Called from main context */
 353 void pa_source_put(pa_source *s) {
 354     pa_source_assert_ref(s);
 355     pa_assert_ctl_context();
 356
 357     pa_assert(s->state == PA_SOURCE_INIT);
 358
 359     /* The following fields must be initialized properly when calling _put() */
 360     pa_assert(s->asyncmsgq);
 361     pa_assert(s->thread_info.min_latency <= s->thread_info.max_latency);
 362
 363     /* Generally, flags should be initialized via pa_source_new(). As
 364      * a special exception we allow volume related flags to be set
 365      * between _new() and _put(). */
 366
 367     if (!(s->flags & PA_SOURCE_HW_VOLUME_CTRL))
 368         s->flags |= PA_SOURCE_DECIBEL_VOLUME;
 369
 370     s->thread_info.soft_volume = s->soft_volume;
 371     s->thread_info.soft_muted = s->muted;
 372
 373     pa_assert((s->flags & PA_SOURCE_HW_VOLUME_CTRL) || (s->base_volume == PA_VOLUME_NORM && s->flags & PA_SOURCE_DECIBEL_VOLUME));
 374     pa_assert(!(s->flags & PA_SOURCE_DECIBEL_VOLUME) || s->n_volume_steps == PA_VOLUME_NORM+1);
 375     pa_assert(!(s->flags & PA_SOURCE_DYNAMIC_LATENCY) == (s->thread_info.fixed_latency != 0));
 376
 377     pa_assert_se(source_set_state(s, PA_SOURCE_IDLE) == 0);
 378
 379     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE | PA_SUBSCRIPTION_EVENT_NEW, s->index);
 380     pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_PUT], s);
 381 }
 382
 383 /* Called from main context */
 384 void pa_source_unlink(pa_source *s) {
 385     pa_bool_t linked;
 386     pa_source_output *o, *j = NULL;
 387
 388     pa_assert(s);
 389     pa_assert_ctl_context();
 390
 391     /* See pa_sink_unlink() for a couple of comments how this function
 392      * works. */
 393
 394     linked = PA_SOURCE_IS_LINKED(s->state);
 395
 396     if (linked)
 397         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_UNLINK], s);
 398
 399     if (s->state != PA_SOURCE_UNLINKED)
 400         pa_namereg_unregister(s->core, s->name);
 401     pa_idxset_remove_by_data(s->core->sources, s, NULL);
 402
 403     if (s->card)
 404         pa_idxset_remove_by_data(s->card->sources, s, NULL);
 405
 406     while ((o = pa_idxset_first(s->outputs, NULL))) {
 407         pa_assert(o != j);
 408         pa_source_output_kill(o);
 409         j = o;
 410     }
 411
 412     if (linked)
 413         source_set_state(s, PA_SOURCE_UNLINKED);
 414     else
 415         s->state = PA_SOURCE_UNLINKED;
 416
 417     reset_callbacks(s);
 418
 419     if (linked) {
 420         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE | PA_SUBSCRIPTION_EVENT_REMOVE, s->index);
 421         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_UNLINK_POST], s);
 422     }
 423 }
 424
 425 /* Called from main context */
 426 static void source_free(pa_object *o) {
 427     pa_source_output *so;
 428     pa_source *s = PA_SOURCE(o);
 429
 430     pa_assert(s);
 431     pa_assert_ctl_context();
 432     pa_assert(pa_source_refcnt(s) == 0);
 433
 434     if (PA_SOURCE_IS_LINKED(s->state))
 435         pa_source_unlink(s);
 436
 437     pa_log_info("Freeing source %u \"%s\"", s->index, s->name);
 438
 439     pa_idxset_free(s->outputs, NULL, NULL);
 440
 441     while ((so = pa_hashmap_steal_first(s->thread_info.outputs)))
 442         pa_source_output_unref(so);
 443
 444     pa_hashmap_free(s->thread_info.outputs, NULL, NULL);
 445
 446     if (s->silence.memblock)
 447         pa_memblock_unref(s->silence.memblock);
 448
 449     pa_xfree(s->name);
 450     pa_xfree(s->driver);
 451
 452     if (s->proplist)
 453         pa_proplist_free(s->proplist);
 454
 455     if (s->ports) {
 456         pa_device_port *p;
 457
 458         while ((p = pa_hashmap_steal_first(s->ports)))
 459             pa_device_port_free(p);
 460
 461         pa_hashmap_free(s->ports, NULL, NULL);
 462     }
 463
 464     pa_xfree(s);
 465 }
 466
 467 /* Called from main context, and not while the IO thread is active, please */
 468 void pa_source_set_asyncmsgq(pa_source *s, pa_asyncmsgq *q) {
 469     pa_source_assert_ref(s);
 470     pa_assert_ctl_context();
 471
 472     s->asyncmsgq = q;
 473 }
 474
 475 /* Called from main context, and not while the IO thread is active, please */
 476 void pa_source_update_flags(pa_source *s, pa_source_flags_t mask, pa_source_flags_t value) {
 477     pa_source_assert_ref(s);
 478     pa_assert_ctl_context();
 479
 480     if (mask == 0)
 481         return;
 482
 483     /* For now, allow only a minimal set of flags to be changed. */
 484     pa_assert((mask & ~(PA_SOURCE_DYNAMIC_LATENCY|PA_SOURCE_LATENCY)) == 0);
 485
 486     s->flags = (s->flags & ~mask) | (value & mask);
 487 }
 488
 489 /* Called from IO context, or before _put() from main context */
 490 void pa_source_set_rtpoll(pa_source *s, pa_rtpoll *p) {
 491     pa_source_assert_ref(s);
 492     pa_source_assert_io_context(s);
 493
 494     s->thread_info.rtpoll = p;
 495 }
 496
 497 /* Called from main context */
 498 int pa_source_update_status(pa_source*s) {
 499     pa_source_assert_ref(s);
 500     pa_assert_ctl_context();
 501     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 502
 503     if (s->state == PA_SOURCE_SUSPENDED)
 504         return 0;
 505
 506     return source_set_state(s, pa_source_used_by(s) ? PA_SOURCE_RUNNING : PA_SOURCE_IDLE);
 507 }
 508
 509 /* Called from main context */
 510 int pa_source_suspend(pa_source *s, pa_bool_t suspend, pa_suspend_cause_t cause) {
 511     pa_source_assert_ref(s);
 512     pa_assert_ctl_context();
 513     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 514     pa_assert(cause != 0);
 515
 516     if (s->monitor_of)
 517         return -PA_ERR_NOTSUPPORTED;
 518
 519     if (suspend)
 520         s->suspend_cause |= cause;
 521     else
 522         s->suspend_cause &= ~cause;
 523
 524     if ((pa_source_get_state(s) == PA_SOURCE_SUSPENDED) == !!s->suspend_cause)
 525         return 0;
 526
 527     pa_log_debug("Suspend cause of source %s is 0x%04x, %s", s->name, s->suspend_cause, s->suspend_cause ? "suspending" : "resuming");
 528
 529     if (suspend)
 530         return source_set_state(s, PA_SOURCE_SUSPENDED);
 531     else
 532         return source_set_state(s, pa_source_used_by(s) ? PA_SOURCE_RUNNING : PA_SOURCE_IDLE);
 533 }
 534
 535 /* Called from main context */
 536 int pa_source_sync_suspend(pa_source *s) {
 537     pa_sink_state_t state;
 538
 539     pa_source_assert_ref(s);
 540     pa_assert_ctl_context();
 541     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 542     pa_assert(s->monitor_of);
 543
 544     state = pa_sink_get_state(s->monitor_of);
 545
 546     if (state == PA_SINK_SUSPENDED)
 547         return source_set_state(s, PA_SOURCE_SUSPENDED);
 548
 549     pa_assert(PA_SINK_IS_OPENED(state));
 550
 551     return source_set_state(s, pa_source_used_by(s) ? PA_SOURCE_RUNNING : PA_SOURCE_IDLE);
 552 }
 553
 554 /* Called from main context */
 555 pa_queue *pa_source_move_all_start(pa_source *s, pa_queue *q) {
 556     pa_source_output *o, *n;
 557     uint32_t idx;
 558
 559     pa_source_assert_ref(s);
 560     pa_assert_ctl_context();
 561     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 562
 563     if (!q)
 564         q = pa_queue_new();
 565
 566     for (o = PA_SOURCE_OUTPUT(pa_idxset_first(s->outputs, &idx)); o; o = n) {
 567         n = PA_SOURCE_OUTPUT(pa_idxset_next(s->outputs, &idx));
 568
 569         pa_source_output_ref(o);
 570
 571         if (pa_source_output_start_move(o) >= 0)
 572             pa_queue_push(q, o);
 573         else
 574             pa_source_output_unref(o);
 575     }
 576
 577     return q;
 578 }
 579
 580 /* Called from main context */
 581 void pa_source_move_all_finish(pa_source *s, pa_queue *q, pa_bool_t save) {
 582     pa_source_output *o;
 583
 584     pa_source_assert_ref(s);
 585     pa_assert_ctl_context();
 586     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 587     pa_assert(q);
 588
 589     while ((o = PA_SOURCE_OUTPUT(pa_queue_pop(q)))) {
 590         if (pa_source_output_finish_move(o, s, save) < 0)
 591             pa_source_output_fail_move(o);
 592
 593         pa_source_output_unref(o);
 594     }
 595
 596     pa_queue_free(q, NULL, NULL);
 597 }
 598
 599 /* Called from main context */
 600 void pa_source_move_all_fail(pa_queue *q) {
 601     pa_source_output *o;
 602
 603     pa_assert_ctl_context();
 604     pa_assert(q);
 605
 606     while ((o = PA_SOURCE_OUTPUT(pa_queue_pop(q)))) {
 607         pa_source_output_fail_move(o);
 608         pa_source_output_unref(o);
 609     }
 610
 611     pa_queue_free(q, NULL, NULL);
 612 }
 613
 614 /* Called from IO thread context */
 615 void pa_source_process_rewind(pa_source *s, size_t nbytes) {
 616     pa_source_output *o;
 617     void *state = NULL;
 618
 619     pa_source_assert_ref(s);
 620     pa_source_assert_io_context(s);
 621     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
 622
 623     if (nbytes <= 0)
 624         return;
 625
 626     if (s->thread_info.state == PA_SOURCE_SUSPENDED)
 627         return;
 628
 629     pa_log_debug("Processing rewind...");
 630
 631     PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state) {
 632         pa_source_output_assert_ref(o);
 633         pa_source_output_process_rewind(o, nbytes);
 634     }
 635 }
 636
 637 /* Called from IO thread context */
 638 void pa_source_post(pa_source*s, const pa_memchunk *chunk) {
 639     pa_source_output *o;
 640     void *state = NULL;
 641
 642     pa_source_assert_ref(s);
 643     pa_source_assert_io_context(s);
 644     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
 645     pa_assert(chunk);
 646
 647     if (s->thread_info.state == PA_SOURCE_SUSPENDED)
 648         return;
 649
 650     if (s->thread_info.soft_muted || !pa_cvolume_is_norm(&s->thread_info.soft_volume)) {
 651         pa_memchunk vchunk = *chunk;
 652
 653         pa_memblock_ref(vchunk.memblock);
 654         pa_memchunk_make_writable(&vchunk, 0);
 655
 656         if (s->thread_info.soft_muted || pa_cvolume_is_muted(&s->thread_info.soft_volume))
 657             pa_silence_memchunk(&vchunk, &s->sample_spec);
 658         else
 659             pa_volume_memchunk(&vchunk, &s->sample_spec, &s->thread_info.soft_volume);
 660
 661         while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL))) {
 662             pa_source_output_assert_ref(o);
 663
 664             if (!o->thread_info.direct_on_input)
 665                 pa_source_output_push(o, &vchunk);
 666         }
 667
 668         pa_memblock_unref(vchunk.memblock);
 669     } else {
 670
 671         while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL))) {
 672             pa_source_output_assert_ref(o);
 673
 674             if (!o->thread_info.direct_on_input)
 675                 pa_source_output_push(o, chunk);
 676         }
 677     }
 678 }
 679
 680 /* Called from IO thread context */
 681 void pa_source_post_direct(pa_source*s, pa_source_output *o, const pa_memchunk *chunk) {
 682     pa_source_assert_ref(s);
 683     pa_source_assert_io_context(s);
 684     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
 685     pa_source_output_assert_ref(o);
 686     pa_assert(o->thread_info.direct_on_input);
 687     pa_assert(chunk);
 688
 689     if (s->thread_info.state == PA_SOURCE_SUSPENDED)
 690         return;
 691
 692     if (s->thread_info.soft_muted || !pa_cvolume_is_norm(&s->thread_info.soft_volume)) {
 693         pa_memchunk vchunk = *chunk;
 694
 695         pa_memblock_ref(vchunk.memblock);
 696         pa_memchunk_make_writable(&vchunk, 0);
 697
 698         if (s->thread_info.soft_muted || pa_cvolume_is_muted(&s->thread_info.soft_volume))
 699             pa_silence_memchunk(&vchunk, &s->sample_spec);
 700         else
 701             pa_volume_memchunk(&vchunk, &s->sample_spec, &s->thread_info.soft_volume);
 702
 703         pa_source_output_push(o, &vchunk);
 704
 705         pa_memblock_unref(vchunk.memblock);
 706     } else
 707         pa_source_output_push(o, chunk);
 708 }
 709
 710 /* Called from main thread */
 711 pa_usec_t pa_source_get_latency(pa_source *s) {
 712     pa_usec_t usec;
 713
 714     pa_source_assert_ref(s);
 715     pa_assert_ctl_context();
 716     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 717
 718     if (s->state == PA_SOURCE_SUSPENDED)
 719         return 0;
 720
 721     if (!(s->flags & PA_SOURCE_LATENCY))
 722         return 0;
 723
 724     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY, &usec, 0, NULL) == 0);
 725
 726     return usec;
 727 }
 728
 729 /* Called from IO thread */
 730 pa_usec_t pa_source_get_latency_within_thread(pa_source *s) {
 731     pa_usec_t usec = 0;
 732     pa_msgobject *o;
 733
 734     pa_source_assert_ref(s);
 735     pa_source_assert_io_context(s);
 736     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
 737
 738     /* The returned value is supposed to be in the time domain of the sound card! */
 739
 740     if (s->thread_info.state == PA_SOURCE_SUSPENDED)
 741         return 0;
 742
 743     if (!(s->flags & PA_SOURCE_LATENCY))
 744         return 0;
 745
 746     o = PA_MSGOBJECT(s);
 747
 748     /* We probably should make this a proper vtable callback instead of going through process_msg() */
 749
 750     if (o->process_msg(o, PA_SOURCE_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
 751         return -1;
 752
 753     return usec;
 754 }
 755
 756 /* Called from main thread */
 757 void pa_source_set_volume(
 758         pa_source *s,
 759         const pa_cvolume *volume,
 760         pa_bool_t save) {
 761
 762     pa_bool_t real_changed;
 763     pa_cvolume old_volume;
 764
 765     pa_source_assert_ref(s);
 766     pa_assert_ctl_context();
 767     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 768     pa_assert(pa_cvolume_valid(volume));
 769     pa_assert(volume->channels == 1 || pa_cvolume_compatible(volume, &s->sample_spec));
 770
 771     old_volume = s->volume;
 772
 773     if (pa_cvolume_compatible(volume, &s->sample_spec))
 774         s->volume = *volume;
 775     else
 776         pa_cvolume_scale(&s->volume, pa_cvolume_max(volume));
 777
 778     real_changed = !pa_cvolume_equal(&old_volume, &s->volume);
 779     s->save_volume = (!real_changed && s->save_volume) || save;
 780
 781     if (s->set_volume) {
 782         pa_cvolume_reset(&s->soft_volume, s->sample_spec.channels);
 783         s->set_volume(s);
 784     } else
 785         s->soft_volume = s->volume;
 786
 787     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_VOLUME, NULL, 0, NULL) == 0);
 788
 789     if (real_changed)
 790         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 791 }
 792
 793 /* Called from main thread. Only to be called by source implementor */
 794 void pa_source_set_soft_volume(pa_source *s, const pa_cvolume *volume) {
 795     pa_source_assert_ref(s);
 796     pa_assert_ctl_context();
 797
 798     if (!volume)
 799         pa_cvolume_reset(&s->soft_volume, s->sample_spec.channels);
 800     else
 801         s->soft_volume = *volume;
 802
 803     if (PA_SOURCE_IS_LINKED(s->state))
 804         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_VOLUME, NULL, 0, NULL) == 0);
 805     else
 806         s->thread_info.soft_volume = s->soft_volume;
 807 }
 808
 809 /* Called from main thread */
 810 const pa_cvolume *pa_source_get_volume(pa_source *s, pa_bool_t force_refresh) {
 811     pa_source_assert_ref(s);
 812     pa_assert_ctl_context();
 813     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 814
 815     if (s->refresh_volume || force_refresh) {
 816         pa_cvolume old_volume;
 817
 818         old_volume = s->volume;
 819
 820         if (s->get_volume)
 821             s->get_volume(s);
 822
 823         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_VOLUME, NULL, 0, NULL) == 0);
 824
 825         if (!pa_cvolume_equal(&old_volume, &s->volume)) {
 826             s->save_volume = TRUE;
 827             pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 828         }
 829     }
 830
 831     return &s->volume;
 832 }
 833
 834 /* Called from main thread */
 835 void pa_source_volume_changed(pa_source *s, const pa_cvolume *new_volume) {
 836     pa_source_assert_ref(s);
 837     pa_assert_ctl_context();
 838     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 839
 840     /* The source implementor may call this if the volume changed to make sure everyone is notified */
 841
 842     if (pa_cvolume_equal(&s->volume, new_volume))
 843         return;
 844
 845     s->volume = *new_volume;
 846     s->save_volume = TRUE;
 847
 848     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 849 }
 850
 851 /* Called from main thread */
 852 void pa_source_set_mute(pa_source *s, pa_bool_t mute, pa_bool_t save) {
 853     pa_bool_t old_muted;
 854
 855     pa_source_assert_ref(s);
 856     pa_assert_ctl_context();
 857     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 858
 859     old_muted = s->muted;
 860     s->muted = mute;
 861     s->save_muted = (old_muted == s->muted && s->save_muted) || save;
 862
 863     if (s->set_mute)
 864         s->set_mute(s);
 865
 866     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MUTE, NULL, 0, NULL) == 0);
 867
 868     if (old_muted != s->muted)
 869         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 870 }
 871
 872 /* Called from main thread */
 873 pa_bool_t pa_source_get_mute(pa_source *s, pa_bool_t force_refresh) {
 874     pa_source_assert_ref(s);
 875     pa_assert_ctl_context();
 876     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 877
 878     if (s->refresh_muted || force_refresh) {
 879         pa_bool_t old_muted = s->muted;
 880
 881         if (s->get_mute)
 882             s->get_mute(s);
 883
 884         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MUTE, NULL, 0, NULL) == 0);
 885
 886         if (old_muted != s->muted) {
 887             s->save_muted = TRUE;
 888
 889             pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 890
 891             /* Make sure the soft mute status stays in sync */
 892             pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MUTE, NULL, 0, NULL) == 0);
 893         }
 894     }
 895
 896     return s->muted;
 897 }
 898
 899 /* Called from main thread */
 900 void pa_source_mute_changed(pa_source *s, pa_bool_t new_muted) {
 901     pa_source_assert_ref(s);
 902     pa_assert_ctl_context();
 903     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 904
 905     /* The source implementor may call this if the mute state changed to make sure everyone is notified */
 906
 907     if (s->muted == new_muted)
 908         return;
 909
 910     s->muted = new_muted;
 911     s->save_muted = TRUE;
 912
 913     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 914 }
 915
 916 /* Called from main thread */
 917 pa_bool_t pa_source_update_proplist(pa_source *s, pa_update_mode_t mode, pa_proplist *p) {
 918     pa_source_assert_ref(s);
 919     pa_assert_ctl_context();
 920
 921     if (p)
 922         pa_proplist_update(s->proplist, mode, p);
 923
 924     if (PA_SOURCE_IS_LINKED(s->state)) {
 925         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_PROPLIST_CHANGED], s);
 926         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 927     }
 928
 929     return TRUE;
 930 }
 931
 932 /* Called from main thread */
 933 /* FIXME -- this should be dropped and be merged into pa_source_update_proplist() */
 934 void pa_source_set_description(pa_source *s, const char *description) {
 935     const char *old;
 936     pa_source_assert_ref(s);
 937     pa_assert_ctl_context();
 938
 939     if (!description && !pa_proplist_contains(s->proplist, PA_PROP_DEVICE_DESCRIPTION))
 940         return;
 941
 942     old = pa_proplist_gets(s->proplist, PA_PROP_DEVICE_DESCRIPTION);
 943
 944     if (old && description && pa_streq(old, description))
 945         return;
 946
 947     if (description)
 948         pa_proplist_sets(s->proplist, PA_PROP_DEVICE_DESCRIPTION, description);
 949     else
 950         pa_proplist_unset(s->proplist, PA_PROP_DEVICE_DESCRIPTION);
 951
 952     if (PA_SOURCE_IS_LINKED(s->state)) {
 953         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 954         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_PROPLIST_CHANGED], s);
 955     }
 956 }
 957
 958 /* Called from main thread */
 959 unsigned pa_source_linked_by(pa_source *s) {
 960     pa_source_assert_ref(s);
 961     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 962     pa_assert_ctl_context();
 963
 964     return pa_idxset_size(s->outputs);
 965 }
 966
 967 /* Called from main thread */
 968 unsigned pa_source_used_by(pa_source *s) {
 969     unsigned ret;
 970
 971     pa_source_assert_ref(s);
 972     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 973     pa_assert_ctl_context();
 974
 975     ret = pa_idxset_size(s->outputs);
 976     pa_assert(ret >= s->n_corked);
 977
 978     return ret - s->n_corked;
 979 }
 980
 981 /* Called from main thread */
 982 unsigned pa_source_check_suspend(pa_source *s) {
 983     unsigned ret;
 984     pa_source_output *o;
 985     uint32_t idx;
 986
 987     pa_source_assert_ref(s);
 988     pa_assert_ctl_context();
 989
 990     if (!PA_SOURCE_IS_LINKED(s->state))
 991         return 0;
 992
 993     ret = 0;
 994
 995     PA_IDXSET_FOREACH(o, s->outputs, idx) {
 996         pa_source_output_state_t st;
 997
 998         st = pa_source_output_get_state(o);
 999
1000         /* We do not assert here. It is perfectly valid for a source output to
1001          * be in the INIT state (i.e. created, marked done but not yet put)
1002          * and we should not care if it's unlinked as it won't contribute
1003          * towarards our busy status.
1004          */
1005         if (!PA_SOURCE_OUTPUT_IS_LINKED(st))
1006             continue;
1007
1008         if (st == PA_SOURCE_OUTPUT_CORKED)
1009             continue;
1010
1011         if (o->flags & PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND)
1012             continue;
1013
1014         ret ++;
1015     }
1016
1017     return ret;
1018 }
1019
1020 /* Called from IO thread, except when it is not */
1021 int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
1022     pa_source *s = PA_SOURCE(object);
1023     pa_source_assert_ref(s);
1024
1025     switch ((pa_source_message_t) code) {
1026
1027         case PA_SOURCE_MESSAGE_ADD_OUTPUT: {
1028             pa_source_output *o = PA_SOURCE_OUTPUT(userdata);
1029
1030             pa_hashmap_put(s->thread_info.outputs, PA_UINT32_TO_PTR(o->index), pa_source_output_ref(o));
1031
1032             if (o->direct_on_input) {
1033                 o->thread_info.direct_on_input = o->direct_on_input;
1034                 pa_hashmap_put(o->thread_info.direct_on_input->thread_info.direct_outputs, PA_UINT32_TO_PTR(o->index), o);
1035             }
1036
1037             pa_assert(!o->thread_info.attached);
1038             o->thread_info.attached = TRUE;
1039
1040             if (o->attach)
1041                 o->attach(o);
1042
1043             pa_source_output_set_state_within_thread(o, o->state);
1044
1045             if (o->thread_info.requested_source_latency != (pa_usec_t) -1)
1046                 pa_source_output_set_requested_latency_within_thread(o, o->thread_info.requested_source_latency);
1047
1048             pa_source_output_update_max_rewind(o, s->thread_info.max_rewind);
1049
1050             /* We don't just invalidate the requested latency here,
1051              * because if we are in a move we might need to fix up the
1052              * requested latency. */
1053             pa_source_output_set_requested_latency_within_thread(o, o->thread_info.requested_source_latency);
1054
1055             return 0;
1056         }
1057
1058         case PA_SOURCE_MESSAGE_REMOVE_OUTPUT: {
1059             pa_source_output *o = PA_SOURCE_OUTPUT(userdata);
1060
1061             pa_source_output_set_state_within_thread(o, o->state);
1062
1063             if (o->detach)
1064                 o->detach(o);
1065
1066             pa_assert(o->thread_info.attached);
1067             o->thread_info.attached = FALSE;
1068
1069             if (o->thread_info.direct_on_input) {
1070                 pa_hashmap_remove(o->thread_info.direct_on_input->thread_info.direct_outputs, PA_UINT32_TO_PTR(o->index));
1071                 o->thread_info.direct_on_input = NULL;
1072             }
1073
1074             if (pa_hashmap_remove(s->thread_info.outputs, PA_UINT32_TO_PTR(o->index)))
1075                 pa_source_output_unref(o);
1076
1077             pa_source_invalidate_requested_latency(s, TRUE);
1078
1079             return 0;
1080         }
1081
1082         case PA_SOURCE_MESSAGE_SET_VOLUME:
1083             s->thread_info.soft_volume = s->soft_volume;
1084             return 0;
1085
1086         case PA_SOURCE_MESSAGE_GET_VOLUME:
1087             return 0;
1088
1089         case PA_SOURCE_MESSAGE_SET_MUTE:
1090             s->thread_info.soft_muted = s->muted;
1091             return 0;
1092
1093         case PA_SOURCE_MESSAGE_GET_MUTE:
1094             return 0;
1095
1096         case PA_SOURCE_MESSAGE_SET_STATE: {
1097
1098             pa_bool_t suspend_change =
1099                 (s->thread_info.state == PA_SOURCE_SUSPENDED && PA_SOURCE_IS_OPENED(PA_PTR_TO_UINT(userdata))) ||
1100                 (PA_SOURCE_IS_OPENED(s->thread_info.state) && PA_PTR_TO_UINT(userdata) == PA_SOURCE_SUSPENDED);
1101
1102             s->thread_info.state = PA_PTR_TO_UINT(userdata);
1103
1104             if (suspend_change) {
1105                 pa_source_output *o;
1106                 void *state = NULL;
1107
1108                 while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1109                     if (o->suspend_within_thread)
1110                         o->suspend_within_thread(o, s->thread_info.state == PA_SOURCE_SUSPENDED);
1111             }
1112
1113
1114             return 0;
1115         }
1116
1117         case PA_SOURCE_MESSAGE_DETACH:
1118
1119             /* Detach all streams */
1120             pa_source_detach_within_thread(s);
1121             return 0;
1122
1123         case PA_SOURCE_MESSAGE_ATTACH:
1124
1125             /* Reattach all streams */
1126             pa_source_attach_within_thread(s);
1127             return 0;
1128
1129         case PA_SOURCE_MESSAGE_GET_REQUESTED_LATENCY: {
1130
1131             pa_usec_t *usec = userdata;
1132             *usec = pa_source_get_requested_latency_within_thread(s);
1133
1134             if (*usec == (pa_usec_t) -1)
1135                 *usec = s->thread_info.max_latency;
1136
1137             return 0;
1138         }
1139
1140         case PA_SOURCE_MESSAGE_SET_LATENCY_RANGE: {
1141             pa_usec_t *r = userdata;
1142
1143             pa_source_set_latency_range_within_thread(s, r[0], r[1]);
1144
1145             return 0;
1146         }
1147
1148         case PA_SOURCE_MESSAGE_GET_LATENCY_RANGE: {
1149             pa_usec_t *r = userdata;
1150
1151             r[0] = s->thread_info.min_latency;
1152             r[1] = s->thread_info.max_latency;
1153
1154             return 0;
1155         }
1156
1157         case PA_SOURCE_MESSAGE_GET_FIXED_LATENCY:
1158
1159             *((pa_usec_t*) userdata) = s->thread_info.fixed_latency;
1160             return 0;
1161
1162         case PA_SOURCE_MESSAGE_SET_FIXED_LATENCY:
1163
1164             pa_source_set_fixed_latency_within_thread(s, (pa_usec_t) offset);
1165             return 0;
1166
1167         case PA_SOURCE_MESSAGE_GET_MAX_REWIND:
1168
1169             *((size_t*) userdata) = s->thread_info.max_rewind;
1170             return 0;
1171
1172         case PA_SOURCE_MESSAGE_SET_MAX_REWIND:
1173
1174             pa_source_set_max_rewind_within_thread(s, (size_t) offset);
1175             return 0;
1176
1177         case PA_SOURCE_MESSAGE_GET_LATENCY:
1178
1179             if (s->monitor_of) {
1180                 *((pa_usec_t*) userdata) = 0;
1181                 return 0;
1182             }
1183
1184             /* Implementors need to overwrite this implementation! */
1185             return -1;
1186
1187         case PA_SOURCE_MESSAGE_MAX:
1188             ;
1189     }
1190
1191     return -1;
1192 }
1193
1194 /* Called from main thread */
1195 int pa_source_suspend_all(pa_core *c, pa_bool_t suspend, pa_suspend_cause_t cause) {
1196     uint32_t idx;
1197     pa_source *source;
1198     int ret = 0;
1199
1200     pa_core_assert_ref(c);
1201     pa_assert_ctl_context();
1202     pa_assert(cause != 0);
1203
1204     for (source = PA_SOURCE(pa_idxset_first(c->sources, &idx)); source; source = PA_SOURCE(pa_idxset_next(c->sources, &idx))) {
1205         int r;
1206
1207         if (source->monitor_of)
1208             continue;
1209
1210         if ((r = pa_source_suspend(source, suspend, cause)) < 0)
1211             ret = r;
1212     }
1213
1214     return ret;
1215 }
1216
1217 /* Called from main thread */
1218 void pa_source_detach(pa_source *s) {
1219     pa_source_assert_ref(s);
1220     pa_assert_ctl_context();
1221     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1222
1223     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_DETACH, NULL, 0, NULL) == 0);
1224 }
1225
1226 /* Called from main thread */
1227 void pa_source_attach(pa_source *s) {
1228     pa_source_assert_ref(s);
1229     pa_assert_ctl_context();
1230     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1231
1232     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_ATTACH, NULL, 0, NULL) == 0);
1233 }
1234
1235 /* Called from IO thread */
1236 void pa_source_detach_within_thread(pa_source *s) {
1237     pa_source_output *o;
1238     void *state = NULL;
1239
1240     pa_source_assert_ref(s);
1241     pa_source_assert_io_context(s);
1242     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
1243
1244     PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1245         if (o->detach)
1246             o->detach(o);
1247 }
1248
1249 /* Called from IO thread */
1250 void pa_source_attach_within_thread(pa_source *s) {
1251     pa_source_output *o;
1252     void *state = NULL;
1253
1254     pa_source_assert_ref(s);
1255     pa_source_assert_io_context(s);
1256     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
1257
1258     PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1259         if (o->attach)
1260             o->attach(o);
1261 }
1262
1263 /* Called from IO thread */
1264 pa_usec_t pa_source_get_requested_latency_within_thread(pa_source *s) {
1265     pa_usec_t result = (pa_usec_t) -1;
1266     pa_source_output *o;
1267     void *state = NULL;
1268
1269     pa_source_assert_ref(s);
1270     pa_source_assert_io_context(s);
1271
1272     if (!(s->flags & PA_SOURCE_DYNAMIC_LATENCY))
1273         return PA_CLAMP(s->thread_info.fixed_latency, s->thread_info.min_latency, s->thread_info.max_latency);
1274
1275     if (s->thread_info.requested_latency_valid)
1276         return s->thread_info.requested_latency;
1277
1278     PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1279         if (o->thread_info.requested_source_latency != (pa_usec_t) -1 &&
1280             (result == (pa_usec_t) -1 || result > o->thread_info.requested_source_latency))
1281             result = o->thread_info.requested_source_latency;
1282
1283     if (result != (pa_usec_t) -1)
1284         result = PA_CLAMP(result, s->thread_info.min_latency, s->thread_info.max_latency);
1285
1286     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1287         /* Only cache this if we are fully set up */
1288         s->thread_info.requested_latency = result;
1289         s->thread_info.requested_latency_valid = TRUE;
1290     }
1291
1292     return result;
1293 }
1294
1295 /* Called from main thread */
1296 pa_usec_t pa_source_get_requested_latency(pa_source *s) {
1297     pa_usec_t usec = 0;
1298
1299     pa_source_assert_ref(s);
1300     pa_assert_ctl_context();
1301     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1302
1303     if (s->state == PA_SOURCE_SUSPENDED)
1304         return 0;
1305
1306     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
1307
1308     return usec;
1309 }
1310
1311 /* Called from IO thread */
1312 void pa_source_set_max_rewind_within_thread(pa_source *s, size_t max_rewind) {
1313     pa_source_output *o;
1314     void *state = NULL;
1315
1316     pa_source_assert_ref(s);
1317     pa_source_assert_io_context(s);
1318
1319     if (max_rewind == s->thread_info.max_rewind)
1320         return;
1321
1322     s->thread_info.max_rewind = max_rewind;
1323
1324     if (PA_SOURCE_IS_LINKED(s->thread_info.state))
1325         PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1326             pa_source_output_update_max_rewind(o, s->thread_info.max_rewind);
1327 }
1328
1329 /* Called from main thread */
1330 void pa_source_set_max_rewind(pa_source *s, size_t max_rewind) {
1331     pa_source_assert_ref(s);
1332     pa_assert_ctl_context();
1333
1334     if (PA_SOURCE_IS_LINKED(s->state))
1335         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MAX_REWIND, NULL, max_rewind, NULL) == 0);
1336     else
1337         pa_source_set_max_rewind_within_thread(s, max_rewind);
1338 }
1339
1340 /* Called from IO thread */
1341 void pa_source_invalidate_requested_latency(pa_source *s, pa_bool_t dynamic) {
1342     pa_source_output *o;
1343     void *state = NULL;
1344
1345     pa_source_assert_ref(s);
1346     pa_source_assert_io_context(s);
1347
1348     if ((s->flags & PA_SOURCE_DYNAMIC_LATENCY))
1349         s->thread_info.requested_latency_valid = FALSE;
1350     else if (dynamic)
1351         return;
1352
1353     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1354
1355         if (s->update_requested_latency)
1356             s->update_requested_latency(s);
1357
1358         while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1359             if (o->update_source_requested_latency)
1360                 o->update_source_requested_latency(o);
1361     }
1362
1363     if (s->monitor_of)
1364         pa_sink_invalidate_requested_latency(s->monitor_of, dynamic);
1365 }
1366
1367 /* Called from main thread */
1368 void pa_source_set_latency_range(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency) {
1369     pa_source_assert_ref(s);
1370     pa_assert_ctl_context();
1371
1372     /* min_latency == 0:           no limit
1373      * min_latency anything else:  specified limit
1374      *
1375      * Similar for max_latency */
1376
1377     if (min_latency < ABSOLUTE_MIN_LATENCY)
1378         min_latency = ABSOLUTE_MIN_LATENCY;
1379
1380     if (max_latency <= 0 ||
1381         max_latency > ABSOLUTE_MAX_LATENCY)
1382         max_latency = ABSOLUTE_MAX_LATENCY;
1383
1384     pa_assert(min_latency <= max_latency);
1385
1386     /* Hmm, let's see if someone forgot to set PA_SOURCE_DYNAMIC_LATENCY here... */
1387     pa_assert((min_latency == ABSOLUTE_MIN_LATENCY &&
1388                max_latency == ABSOLUTE_MAX_LATENCY) ||
1389               (s->flags & PA_SOURCE_DYNAMIC_LATENCY));
1390
1391     if (PA_SOURCE_IS_LINKED(s->state)) {
1392         pa_usec_t r[2];
1393
1394         r[0] = min_latency;
1395         r[1] = max_latency;
1396
1397         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_LATENCY_RANGE, r, 0, NULL) == 0);
1398     } else
1399         pa_source_set_latency_range_within_thread(s, min_latency, max_latency);
1400 }
1401
1402 /* Called from main thread */
1403 void pa_source_get_latency_range(pa_source *s, pa_usec_t *min_latency, pa_usec_t *max_latency) {
1404    pa_source_assert_ref(s);
1405    pa_assert_ctl_context();
1406    pa_assert(min_latency);
1407    pa_assert(max_latency);
1408
1409    if (PA_SOURCE_IS_LINKED(s->state)) {
1410        pa_usec_t r[2] = { 0, 0 };
1411
1412        pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY_RANGE, r, 0, NULL) == 0);
1413
1414        *min_latency = r[0];
1415        *max_latency = r[1];
1416    } else {
1417        *min_latency = s->thread_info.min_latency;
1418        *max_latency = s->thread_info.max_latency;
1419    }
1420 }
1421
1422 /* Called from IO thread, and from main thread before pa_source_put() is called */
1423 void pa_source_set_latency_range_within_thread(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency) {
1424     pa_source_assert_ref(s);
1425     pa_source_assert_io_context(s);
1426
1427     pa_assert(min_latency >= ABSOLUTE_MIN_LATENCY);
1428     pa_assert(max_latency <= ABSOLUTE_MAX_LATENCY);
1429     pa_assert(min_latency <= max_latency);
1430
1431     /* Hmm, let's see if someone forgot to set PA_SOURCE_DYNAMIC_LATENCY here... */
1432     pa_assert((min_latency == ABSOLUTE_MIN_LATENCY &&
1433                max_latency == ABSOLUTE_MAX_LATENCY) ||
1434               (s->flags & PA_SOURCE_DYNAMIC_LATENCY) ||
1435               s->monitor_of);
1436
1437     if (s->thread_info.min_latency == min_latency &&
1438         s->thread_info.max_latency == max_latency)
1439         return;
1440
1441     s->thread_info.min_latency = min_latency;
1442     s->thread_info.max_latency = max_latency;
1443
1444     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1445         pa_source_output *o;
1446         void *state = NULL;
1447
1448         PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1449             if (o->update_source_latency_range)
1450                 o->update_source_latency_range(o);
1451     }
1452
1453     pa_source_invalidate_requested_latency(s, FALSE);
1454 }
1455
1456 /* Called from main thread, before the source is put */
1457 void pa_source_set_fixed_latency(pa_source *s, pa_usec_t latency) {
1458     pa_source_assert_ref(s);
1459     pa_assert_ctl_context();
1460
1461     if (s->flags & PA_SOURCE_DYNAMIC_LATENCY) {
1462         pa_assert(latency == 0);
1463         return;
1464     }
1465
1466     if (latency < ABSOLUTE_MIN_LATENCY)
1467         latency = ABSOLUTE_MIN_LATENCY;
1468
1469     if (latency > ABSOLUTE_MAX_LATENCY)
1470         latency = ABSOLUTE_MAX_LATENCY;
1471
1472     if (PA_SOURCE_IS_LINKED(s->state))
1473         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_FIXED_LATENCY, NULL, (int64_t) latency, NULL) == 0);
1474     else
1475         s->thread_info.fixed_latency = latency;
1476 }
1477
1478 /* Called from main thread */
1479 pa_usec_t pa_source_get_fixed_latency(pa_source *s) {
1480     pa_usec_t latency;
1481
1482     pa_source_assert_ref(s);
1483     pa_assert_ctl_context();
1484
1485     if (s->flags & PA_SOURCE_DYNAMIC_LATENCY)
1486         return 0;
1487
1488     if (PA_SOURCE_IS_LINKED(s->state))
1489         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_FIXED_LATENCY, &latency, 0, NULL) == 0);
1490     else
1491         latency = s->thread_info.fixed_latency;
1492
1493     return latency;
1494 }
1495
1496 /* Called from IO thread */
1497 void pa_source_set_fixed_latency_within_thread(pa_source *s, pa_usec_t latency) {
1498     pa_source_assert_ref(s);
1499     pa_source_assert_io_context(s);
1500
1501     if (s->flags & PA_SOURCE_DYNAMIC_LATENCY) {
1502         pa_assert(latency == 0);
1503         return;
1504     }
1505
1506     pa_assert(latency >= ABSOLUTE_MIN_LATENCY);
1507     pa_assert(latency <= ABSOLUTE_MAX_LATENCY);
1508
1509     if (s->thread_info.fixed_latency == latency)
1510         return;
1511
1512     s->thread_info.fixed_latency = latency;
1513
1514     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1515         pa_source_output *o;
1516         void *state = NULL;
1517
1518         PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1519             if (o->update_source_fixed_latency)
1520                 o->update_source_fixed_latency(o);
1521     }
1522
1523     pa_source_invalidate_requested_latency(s, FALSE);
1524 }
1525
1526 /* Called from main thread */
1527 size_t pa_source_get_max_rewind(pa_source *s) {
1528     size_t r;
1529     pa_assert_ctl_context();
1530     pa_source_assert_ref(s);
1531
1532     if (!PA_SOURCE_IS_LINKED(s->state))
1533         return s->thread_info.max_rewind;
1534
1535     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MAX_REWIND, &r, 0, NULL) == 0);
1536
1537     return r;
1538 }
1539
1540 /* Called from main context */
1541 int pa_source_set_port(pa_source *s, const char *name, pa_bool_t save) {
1542     pa_device_port *port;
1543
1544     pa_assert(s);
1545     pa_assert_ctl_context();
1546
1547     if (!s->set_port) {
1548         pa_log_debug("set_port() operation not implemented for source %u \"%s\"", s->index, s->name);
1549         return -PA_ERR_NOTIMPLEMENTED;
1550     }
1551
1552     if (!s->ports)
1553         return -PA_ERR_NOENTITY;
1554
1555     if (!(port = pa_hashmap_get(s->ports, name)))
1556         return -PA_ERR_NOENTITY;
1557
1558     if (s->active_port == port) {
1559         s->save_port = s->save_port || save;
1560         return 0;
1561     }
1562
1563     if ((s->set_port(s, port)) < 0)
1564         return -PA_ERR_NOENTITY;
1565
1566     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
1567
1568     pa_log_info("Changed port of source %u \"%s\" to %s", s->index, s->name, port->name);
1569
1570     s->active_port = port;
1571     s->save_port = save;
1572
1573     return 0;
1574 }