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         pa_assert(PA_SOURCE_OUTPUT_IS_LINKED(st));
1000
1001         if (st == PA_SOURCE_OUTPUT_CORKED)
1002             continue;
1003
1004         if (o->flags & PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND)
1005             continue;
1006
1007         ret ++;
1008     }
1009
1010     return ret;
1011 }
1012
1013 /* Called from IO thread, except when it is not */
1014 int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
1015     pa_source *s = PA_SOURCE(object);
1016     pa_source_assert_ref(s);
1017
1018     switch ((pa_source_message_t) code) {
1019
1020         case PA_SOURCE_MESSAGE_ADD_OUTPUT: {
1021             pa_source_output *o = PA_SOURCE_OUTPUT(userdata);
1022
1023             pa_hashmap_put(s->thread_info.outputs, PA_UINT32_TO_PTR(o->index), pa_source_output_ref(o));
1024
1025             if (o->direct_on_input) {
1026                 o->thread_info.direct_on_input = o->direct_on_input;
1027                 pa_hashmap_put(o->thread_info.direct_on_input->thread_info.direct_outputs, PA_UINT32_TO_PTR(o->index), o);
1028             }
1029
1030             pa_assert(!o->thread_info.attached);
1031             o->thread_info.attached = TRUE;
1032
1033             if (o->attach)
1034                 o->attach(o);
1035
1036             pa_source_output_set_state_within_thread(o, o->state);
1037
1038             if (o->thread_info.requested_source_latency != (pa_usec_t) -1)
1039                 pa_source_output_set_requested_latency_within_thread(o, o->thread_info.requested_source_latency);
1040
1041             pa_source_output_update_max_rewind(o, s->thread_info.max_rewind);
1042
1043             /* We don't just invalidate the requested latency here,
1044              * because if we are in a move we might need to fix up the
1045              * requested latency. */
1046             pa_source_output_set_requested_latency_within_thread(o, o->thread_info.requested_source_latency);
1047
1048             return 0;
1049         }
1050
1051         case PA_SOURCE_MESSAGE_REMOVE_OUTPUT: {
1052             pa_source_output *o = PA_SOURCE_OUTPUT(userdata);
1053
1054             pa_source_output_set_state_within_thread(o, o->state);
1055
1056             if (o->detach)
1057                 o->detach(o);
1058
1059             pa_assert(o->thread_info.attached);
1060             o->thread_info.attached = FALSE;
1061
1062             if (o->thread_info.direct_on_input) {
1063                 pa_hashmap_remove(o->thread_info.direct_on_input->thread_info.direct_outputs, PA_UINT32_TO_PTR(o->index));
1064                 o->thread_info.direct_on_input = NULL;
1065             }
1066
1067             if (pa_hashmap_remove(s->thread_info.outputs, PA_UINT32_TO_PTR(o->index)))
1068                 pa_source_output_unref(o);
1069
1070             pa_source_invalidate_requested_latency(s, TRUE);
1071
1072             return 0;
1073         }
1074
1075         case PA_SOURCE_MESSAGE_SET_VOLUME:
1076             s->thread_info.soft_volume = s->soft_volume;
1077             return 0;
1078
1079         case PA_SOURCE_MESSAGE_GET_VOLUME:
1080             return 0;
1081
1082         case PA_SOURCE_MESSAGE_SET_MUTE:
1083             s->thread_info.soft_muted = s->muted;
1084             return 0;
1085
1086         case PA_SOURCE_MESSAGE_GET_MUTE:
1087             return 0;
1088
1089         case PA_SOURCE_MESSAGE_SET_STATE: {
1090
1091             pa_bool_t suspend_change =
1092                 (s->thread_info.state == PA_SOURCE_SUSPENDED && PA_SOURCE_IS_OPENED(PA_PTR_TO_UINT(userdata))) ||
1093                 (PA_SOURCE_IS_OPENED(s->thread_info.state) && PA_PTR_TO_UINT(userdata) == PA_SOURCE_SUSPENDED);
1094
1095             s->thread_info.state = PA_PTR_TO_UINT(userdata);
1096
1097             if (suspend_change) {
1098                 pa_source_output *o;
1099                 void *state = NULL;
1100
1101                 while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1102                     if (o->suspend_within_thread)
1103                         o->suspend_within_thread(o, s->thread_info.state == PA_SOURCE_SUSPENDED);
1104             }
1105
1106
1107             return 0;
1108         }
1109
1110         case PA_SOURCE_MESSAGE_DETACH:
1111
1112             /* Detach all streams */
1113             pa_source_detach_within_thread(s);
1114             return 0;
1115
1116         case PA_SOURCE_MESSAGE_ATTACH:
1117
1118             /* Reattach all streams */
1119             pa_source_attach_within_thread(s);
1120             return 0;
1121
1122         case PA_SOURCE_MESSAGE_GET_REQUESTED_LATENCY: {
1123
1124             pa_usec_t *usec = userdata;
1125             *usec = pa_source_get_requested_latency_within_thread(s);
1126
1127             if (*usec == (pa_usec_t) -1)
1128                 *usec = s->thread_info.max_latency;
1129
1130             return 0;
1131         }
1132
1133         case PA_SOURCE_MESSAGE_SET_LATENCY_RANGE: {
1134             pa_usec_t *r = userdata;
1135
1136             pa_source_set_latency_range_within_thread(s, r[0], r[1]);
1137
1138             return 0;
1139         }
1140
1141         case PA_SOURCE_MESSAGE_GET_LATENCY_RANGE: {
1142             pa_usec_t *r = userdata;
1143
1144             r[0] = s->thread_info.min_latency;
1145             r[1] = s->thread_info.max_latency;
1146
1147             return 0;
1148         }
1149
1150         case PA_SOURCE_MESSAGE_GET_FIXED_LATENCY:
1151
1152             *((pa_usec_t*) userdata) = s->thread_info.fixed_latency;
1153             return 0;
1154
1155         case PA_SOURCE_MESSAGE_SET_FIXED_LATENCY:
1156
1157             pa_source_set_fixed_latency_within_thread(s, (pa_usec_t) offset);
1158             return 0;
1159
1160         case PA_SOURCE_MESSAGE_GET_MAX_REWIND:
1161
1162             *((size_t*) userdata) = s->thread_info.max_rewind;
1163             return 0;
1164
1165         case PA_SOURCE_MESSAGE_SET_MAX_REWIND:
1166
1167             pa_source_set_max_rewind_within_thread(s, (size_t) offset);
1168             return 0;
1169
1170         case PA_SOURCE_MESSAGE_GET_LATENCY:
1171
1172             if (s->monitor_of) {
1173                 *((pa_usec_t*) userdata) = 0;
1174                 return 0;
1175             }
1176
1177             /* Implementors need to overwrite this implementation! */
1178             return -1;
1179
1180         case PA_SOURCE_MESSAGE_MAX:
1181             ;
1182     }
1183
1184     return -1;
1185 }
1186
1187 /* Called from main thread */
1188 int pa_source_suspend_all(pa_core *c, pa_bool_t suspend, pa_suspend_cause_t cause) {
1189     uint32_t idx;
1190     pa_source *source;
1191     int ret = 0;
1192
1193     pa_core_assert_ref(c);
1194     pa_assert_ctl_context();
1195     pa_assert(cause != 0);
1196
1197     for (source = PA_SOURCE(pa_idxset_first(c->sources, &idx)); source; source = PA_SOURCE(pa_idxset_next(c->sources, &idx))) {
1198         int r;
1199
1200         if (source->monitor_of)
1201             continue;
1202
1203         if ((r = pa_source_suspend(source, suspend, cause)) < 0)
1204             ret = r;
1205     }
1206
1207     return ret;
1208 }
1209
1210 /* Called from main thread */
1211 void pa_source_detach(pa_source *s) {
1212     pa_source_assert_ref(s);
1213     pa_assert_ctl_context();
1214     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1215
1216     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_DETACH, NULL, 0, NULL) == 0);
1217 }
1218
1219 /* Called from main thread */
1220 void pa_source_attach(pa_source *s) {
1221     pa_source_assert_ref(s);
1222     pa_assert_ctl_context();
1223     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1224
1225     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_ATTACH, NULL, 0, NULL) == 0);
1226 }
1227
1228 /* Called from IO thread */
1229 void pa_source_detach_within_thread(pa_source *s) {
1230     pa_source_output *o;
1231     void *state = NULL;
1232
1233     pa_source_assert_ref(s);
1234     pa_source_assert_io_context(s);
1235     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
1236
1237     PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1238         if (o->detach)
1239             o->detach(o);
1240 }
1241
1242 /* Called from IO thread */
1243 void pa_source_attach_within_thread(pa_source *s) {
1244     pa_source_output *o;
1245     void *state = NULL;
1246
1247     pa_source_assert_ref(s);
1248     pa_source_assert_io_context(s);
1249     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
1250
1251     PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1252         if (o->attach)
1253             o->attach(o);
1254 }
1255
1256 /* Called from IO thread */
1257 pa_usec_t pa_source_get_requested_latency_within_thread(pa_source *s) {
1258     pa_usec_t result = (pa_usec_t) -1;
1259     pa_source_output *o;
1260     void *state = NULL;
1261
1262     pa_source_assert_ref(s);
1263     pa_source_assert_io_context(s);
1264
1265     if (!(s->flags & PA_SOURCE_DYNAMIC_LATENCY))
1266         return PA_CLAMP(s->thread_info.fixed_latency, s->thread_info.min_latency, s->thread_info.max_latency);
1267
1268     if (s->thread_info.requested_latency_valid)
1269         return s->thread_info.requested_latency;
1270
1271     PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1272         if (o->thread_info.requested_source_latency != (pa_usec_t) -1 &&
1273             (result == (pa_usec_t) -1 || result > o->thread_info.requested_source_latency))
1274             result = o->thread_info.requested_source_latency;
1275
1276     if (result != (pa_usec_t) -1)
1277         result = PA_CLAMP(result, s->thread_info.min_latency, s->thread_info.max_latency);
1278
1279     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1280         /* Only cache this if we are fully set up */
1281         s->thread_info.requested_latency = result;
1282         s->thread_info.requested_latency_valid = TRUE;
1283     }
1284
1285     return result;
1286 }
1287
1288 /* Called from main thread */
1289 pa_usec_t pa_source_get_requested_latency(pa_source *s) {
1290     pa_usec_t usec = 0;
1291
1292     pa_source_assert_ref(s);
1293     pa_assert_ctl_context();
1294     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1295
1296     if (s->state == PA_SOURCE_SUSPENDED)
1297         return 0;
1298
1299     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
1300
1301     return usec;
1302 }
1303
1304 /* Called from IO thread */
1305 void pa_source_set_max_rewind_within_thread(pa_source *s, size_t max_rewind) {
1306     pa_source_output *o;
1307     void *state = NULL;
1308
1309     pa_source_assert_ref(s);
1310     pa_source_assert_io_context(s);
1311
1312     if (max_rewind == s->thread_info.max_rewind)
1313         return;
1314
1315     s->thread_info.max_rewind = max_rewind;
1316
1317     if (PA_SOURCE_IS_LINKED(s->thread_info.state))
1318         PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1319             pa_source_output_update_max_rewind(o, s->thread_info.max_rewind);
1320 }
1321
1322 /* Called from main thread */
1323 void pa_source_set_max_rewind(pa_source *s, size_t max_rewind) {
1324     pa_source_assert_ref(s);
1325     pa_assert_ctl_context();
1326
1327     if (PA_SOURCE_IS_LINKED(s->state))
1328         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MAX_REWIND, NULL, max_rewind, NULL) == 0);
1329     else
1330         pa_source_set_max_rewind_within_thread(s, max_rewind);
1331 }
1332
1333 /* Called from IO thread */
1334 void pa_source_invalidate_requested_latency(pa_source *s, pa_bool_t dynamic) {
1335     pa_source_output *o;
1336     void *state = NULL;
1337
1338     pa_source_assert_ref(s);
1339     pa_source_assert_io_context(s);
1340
1341     if ((s->flags & PA_SOURCE_DYNAMIC_LATENCY))
1342         s->thread_info.requested_latency_valid = FALSE;
1343     else if (dynamic)
1344         return;
1345
1346     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1347
1348         if (s->update_requested_latency)
1349             s->update_requested_latency(s);
1350
1351         while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1352             if (o->update_source_requested_latency)
1353                 o->update_source_requested_latency(o);
1354     }
1355
1356     if (s->monitor_of)
1357         pa_sink_invalidate_requested_latency(s->monitor_of, dynamic);
1358 }
1359
1360 /* Called from main thread */
1361 void pa_source_set_latency_range(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency) {
1362     pa_source_assert_ref(s);
1363     pa_assert_ctl_context();
1364
1365     /* min_latency == 0:           no limit
1366      * min_latency anything else:  specified limit
1367      *
1368      * Similar for max_latency */
1369
1370     if (min_latency < ABSOLUTE_MIN_LATENCY)
1371         min_latency = ABSOLUTE_MIN_LATENCY;
1372
1373     if (max_latency <= 0 ||
1374         max_latency > ABSOLUTE_MAX_LATENCY)
1375         max_latency = ABSOLUTE_MAX_LATENCY;
1376
1377     pa_assert(min_latency <= max_latency);
1378
1379     /* Hmm, let's see if someone forgot to set PA_SOURCE_DYNAMIC_LATENCY here... */
1380     pa_assert((min_latency == ABSOLUTE_MIN_LATENCY &&
1381                max_latency == ABSOLUTE_MAX_LATENCY) ||
1382               (s->flags & PA_SOURCE_DYNAMIC_LATENCY));
1383
1384     if (PA_SOURCE_IS_LINKED(s->state)) {
1385         pa_usec_t r[2];
1386
1387         r[0] = min_latency;
1388         r[1] = max_latency;
1389
1390         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_LATENCY_RANGE, r, 0, NULL) == 0);
1391     } else
1392         pa_source_set_latency_range_within_thread(s, min_latency, max_latency);
1393 }
1394
1395 /* Called from main thread */
1396 void pa_source_get_latency_range(pa_source *s, pa_usec_t *min_latency, pa_usec_t *max_latency) {
1397    pa_source_assert_ref(s);
1398    pa_assert_ctl_context();
1399    pa_assert(min_latency);
1400    pa_assert(max_latency);
1401
1402    if (PA_SOURCE_IS_LINKED(s->state)) {
1403        pa_usec_t r[2] = { 0, 0 };
1404
1405        pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY_RANGE, r, 0, NULL) == 0);
1406
1407        *min_latency = r[0];
1408        *max_latency = r[1];
1409    } else {
1410        *min_latency = s->thread_info.min_latency;
1411        *max_latency = s->thread_info.max_latency;
1412    }
1413 }
1414
1415 /* Called from IO thread, and from main thread before pa_source_put() is called */
1416 void pa_source_set_latency_range_within_thread(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency) {
1417     pa_source_assert_ref(s);
1418     pa_source_assert_io_context(s);
1419
1420     pa_assert(min_latency >= ABSOLUTE_MIN_LATENCY);
1421     pa_assert(max_latency <= ABSOLUTE_MAX_LATENCY);
1422     pa_assert(min_latency <= max_latency);
1423
1424     /* Hmm, let's see if someone forgot to set PA_SOURCE_DYNAMIC_LATENCY here... */
1425     pa_assert((min_latency == ABSOLUTE_MIN_LATENCY &&
1426                max_latency == ABSOLUTE_MAX_LATENCY) ||
1427               (s->flags & PA_SOURCE_DYNAMIC_LATENCY) ||
1428               s->monitor_of);
1429
1430     if (s->thread_info.min_latency == min_latency &&
1431         s->thread_info.max_latency == max_latency)
1432         return;
1433
1434     s->thread_info.min_latency = min_latency;
1435     s->thread_info.max_latency = max_latency;
1436
1437     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1438         pa_source_output *o;
1439         void *state = NULL;
1440
1441         PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1442             if (o->update_source_latency_range)
1443                 o->update_source_latency_range(o);
1444     }
1445
1446     pa_source_invalidate_requested_latency(s, FALSE);
1447 }
1448
1449 /* Called from main thread, before the source is put */
1450 void pa_source_set_fixed_latency(pa_source *s, pa_usec_t latency) {
1451     pa_source_assert_ref(s);
1452     pa_assert_ctl_context();
1453
1454     if (s->flags & PA_SOURCE_DYNAMIC_LATENCY) {
1455         pa_assert(latency == 0);
1456         return;
1457     }
1458
1459     if (latency < ABSOLUTE_MIN_LATENCY)
1460         latency = ABSOLUTE_MIN_LATENCY;
1461
1462     if (latency > ABSOLUTE_MAX_LATENCY)
1463         latency = ABSOLUTE_MAX_LATENCY;
1464
1465     if (PA_SOURCE_IS_LINKED(s->state))
1466         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_FIXED_LATENCY, NULL, (int64_t) latency, NULL) == 0);
1467     else
1468         s->thread_info.fixed_latency = latency;
1469 }
1470
1471 /* Called from main thread */
1472 pa_usec_t pa_source_get_fixed_latency(pa_source *s) {
1473     pa_usec_t latency;
1474
1475     pa_source_assert_ref(s);
1476     pa_assert_ctl_context();
1477
1478     if (s->flags & PA_SOURCE_DYNAMIC_LATENCY)
1479         return 0;
1480
1481     if (PA_SOURCE_IS_LINKED(s->state))
1482         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_FIXED_LATENCY, &latency, 0, NULL) == 0);
1483     else
1484         latency = s->thread_info.fixed_latency;
1485
1486     return latency;
1487 }
1488
1489 /* Called from IO thread */
1490 void pa_source_set_fixed_latency_within_thread(pa_source *s, pa_usec_t latency) {
1491     pa_source_assert_ref(s);
1492     pa_source_assert_io_context(s);
1493
1494     if (s->flags & PA_SOURCE_DYNAMIC_LATENCY) {
1495         pa_assert(latency == 0);
1496         return;
1497     }
1498
1499     pa_assert(latency >= ABSOLUTE_MIN_LATENCY);
1500     pa_assert(latency <= ABSOLUTE_MAX_LATENCY);
1501
1502     if (s->thread_info.fixed_latency == latency)
1503         return;
1504
1505     s->thread_info.fixed_latency = latency;
1506
1507     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1508         pa_source_output *o;
1509         void *state = NULL;
1510
1511         PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1512             if (o->update_source_fixed_latency)
1513                 o->update_source_fixed_latency(o);
1514     }
1515
1516     pa_source_invalidate_requested_latency(s, FALSE);
1517 }
1518
1519 /* Called from main thread */
1520 size_t pa_source_get_max_rewind(pa_source *s) {
1521     size_t r;
1522     pa_assert_ctl_context();
1523     pa_source_assert_ref(s);
1524
1525     if (!PA_SOURCE_IS_LINKED(s->state))
1526         return s->thread_info.max_rewind;
1527
1528     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MAX_REWIND, &r, 0, NULL) == 0);
1529
1530     return r;
1531 }
1532
1533 /* Called from main context */
1534 int pa_source_set_port(pa_source *s, const char *name, pa_bool_t save) {
1535     pa_device_port *port;
1536
1537     pa_assert(s);
1538     pa_assert_ctl_context();
1539
1540     if (!s->set_port) {
1541         pa_log_debug("set_port() operation not implemented for source %u \"%s\"", s->index, s->name);
1542         return -PA_ERR_NOTIMPLEMENTED;
1543     }
1544
1545     if (!s->ports)
1546         return -PA_ERR_NOENTITY;
1547
1548     if (!(port = pa_hashmap_get(s->ports, name)))
1549         return -PA_ERR_NOENTITY;
1550
1551     if (s->active_port == port) {
1552         s->save_port = s->save_port || save;
1553         return 0;
1554     }
1555
1556     if ((s->set_port(s, port)) < 0)
1557         return -PA_ERR_NOENTITY;
1558
1559     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
1560
1561     pa_log_info("Changed port of source %u \"%s\" to %s", s->index, s->name, port->name);
1562
1563     s->active_port = port;
1564     s->save_port = save;
1565
1566     return 0;
1567 }