code.delx.au - pulseaudio/blob - 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
 764     pa_source_assert_ref(s);
 765     pa_assert_ctl_context();
 766     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 767     pa_assert(pa_cvolume_valid(volume));
 768     pa_assert(pa_cvolume_compatible(volume, &s->sample_spec));
 769
 770     real_changed = !pa_cvolume_equal(volume, &s->volume);
 771     s->volume = *volume;
 772     s->save_volume = (!real_changed && s->save_volume) || save;
 773
 774     if (s->set_volume) {
 775         pa_cvolume_reset(&s->soft_volume, s->sample_spec.channels);
 776         s->set_volume(s);
 777     } else
 778         s->soft_volume = s->volume;
 779
 780     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_VOLUME, NULL, 0, NULL) == 0);
 781
 782     if (real_changed)
 783         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 784 }
 785
 786 /* Called from main thread. Only to be called by source implementor */
 787 void pa_source_set_soft_volume(pa_source *s, const pa_cvolume *volume) {
 788     pa_source_assert_ref(s);
 789     pa_assert_ctl_context();
 790
 791     if (!volume)
 792         pa_cvolume_reset(&s->soft_volume, s->sample_spec.channels);
 793     else
 794         s->soft_volume = *volume;
 795
 796     if (PA_SOURCE_IS_LINKED(s->state))
 797         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_VOLUME, NULL, 0, NULL) == 0);
 798     else
 799         s->thread_info.soft_volume = s->soft_volume;
 800 }
 801
 802 /* Called from main thread */
 803 const pa_cvolume *pa_source_get_volume(pa_source *s, pa_bool_t force_refresh) {
 804     pa_source_assert_ref(s);
 805     pa_assert_ctl_context();
 806     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 807
 808     if (s->refresh_volume || force_refresh) {
 809         pa_cvolume old_volume;
 810
 811         old_volume = s->volume;
 812
 813         if (s->get_volume)
 814             s->get_volume(s);
 815
 816         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_VOLUME, NULL, 0, NULL) == 0);
 817
 818         if (!pa_cvolume_equal(&old_volume, &s->volume)) {
 819             s->save_volume = TRUE;
 820             pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 821         }
 822     }
 823
 824     return &s->volume;
 825 }
 826
 827 /* Called from main thread */
 828 void pa_source_volume_changed(pa_source *s, const pa_cvolume *new_volume) {
 829     pa_source_assert_ref(s);
 830     pa_assert_ctl_context();
 831     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 832
 833     /* The source implementor may call this if the volume changed to make sure everyone is notified */
 834
 835     if (pa_cvolume_equal(&s->volume, new_volume))
 836         return;
 837
 838     s->volume = *new_volume;
 839     s->save_volume = TRUE;
 840
 841     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 842 }
 843
 844 /* Called from main thread */
 845 void pa_source_set_mute(pa_source *s, pa_bool_t mute, pa_bool_t save) {
 846     pa_bool_t old_muted;
 847
 848     pa_source_assert_ref(s);
 849     pa_assert_ctl_context();
 850     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 851
 852     old_muted = s->muted;
 853     s->muted = mute;
 854     s->save_muted = (old_muted == s->muted && s->save_muted) || save;
 855
 856     if (s->set_mute)
 857         s->set_mute(s);
 858
 859     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MUTE, NULL, 0, NULL) == 0);
 860
 861     if (old_muted != s->muted)
 862         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 863 }
 864
 865 /* Called from main thread */
 866 pa_bool_t pa_source_get_mute(pa_source *s, pa_bool_t force_refresh) {
 867     pa_source_assert_ref(s);
 868     pa_assert_ctl_context();
 869     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 870
 871     if (s->refresh_muted || force_refresh) {
 872         pa_bool_t old_muted = s->muted;
 873
 874         if (s->get_mute)
 875             s->get_mute(s);
 876
 877         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MUTE, NULL, 0, NULL) == 0);
 878
 879         if (old_muted != s->muted) {
 880             s->save_muted = TRUE;
 881
 882             pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 883
 884             /* Make sure the soft mute status stays in sync */
 885             pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MUTE, NULL, 0, NULL) == 0);
 886         }
 887     }
 888
 889     return s->muted;
 890 }
 891
 892 /* Called from main thread */
 893 void pa_source_mute_changed(pa_source *s, pa_bool_t new_muted) {
 894     pa_source_assert_ref(s);
 895     pa_assert_ctl_context();
 896     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 897
 898     /* The source implementor may call this if the mute state changed to make sure everyone is notified */
 899
 900     if (s->muted == new_muted)
 901         return;
 902
 903     s->muted = new_muted;
 904     s->save_muted = TRUE;
 905
 906     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 907 }
 908
 909 /* Called from main thread */
 910 pa_bool_t pa_source_update_proplist(pa_source *s, pa_update_mode_t mode, pa_proplist *p) {
 911     pa_source_assert_ref(s);
 912     pa_assert_ctl_context();
 913
 914     if (p)
 915         pa_proplist_update(s->proplist, mode, p);
 916
 917     if (PA_SOURCE_IS_LINKED(s->state)) {
 918         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_PROPLIST_CHANGED], s);
 919         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 920     }
 921
 922     return TRUE;
 923 }
 924
 925 /* Called from main thread */
 926 /* FIXME -- this should be dropped and be merged into pa_source_update_proplist() */
 927 void pa_source_set_description(pa_source *s, const char *description) {
 928     const char *old;
 929     pa_source_assert_ref(s);
 930     pa_assert_ctl_context();
 931
 932     if (!description && !pa_proplist_contains(s->proplist, PA_PROP_DEVICE_DESCRIPTION))
 933         return;
 934
 935     old = pa_proplist_gets(s->proplist, PA_PROP_DEVICE_DESCRIPTION);
 936
 937     if (old && description && pa_streq(old, description))
 938         return;
 939
 940     if (description)
 941         pa_proplist_sets(s->proplist, PA_PROP_DEVICE_DESCRIPTION, description);
 942     else
 943         pa_proplist_unset(s->proplist, PA_PROP_DEVICE_DESCRIPTION);
 944
 945     if (PA_SOURCE_IS_LINKED(s->state)) {
 946         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 947         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_PROPLIST_CHANGED], s);
 948     }
 949 }
 950
 951 /* Called from main thread */
 952 unsigned pa_source_linked_by(pa_source *s) {
 953     pa_source_assert_ref(s);
 954     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 955     pa_assert_ctl_context();
 956
 957     return pa_idxset_size(s->outputs);
 958 }
 959
 960 /* Called from main thread */
 961 unsigned pa_source_used_by(pa_source *s) {
 962     unsigned ret;
 963
 964     pa_source_assert_ref(s);
 965     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 966     pa_assert_ctl_context();
 967
 968     ret = pa_idxset_size(s->outputs);
 969     pa_assert(ret >= s->n_corked);
 970
 971     return ret - s->n_corked;
 972 }
 973
 974 /* Called from main thread */
 975 unsigned pa_source_check_suspend(pa_source *s) {
 976     unsigned ret;
 977     pa_source_output *o;
 978     uint32_t idx;
 979
 980     pa_source_assert_ref(s);
 981     pa_assert_ctl_context();
 982
 983     if (!PA_SOURCE_IS_LINKED(s->state))
 984         return 0;
 985
 986     ret = 0;
 987
 988     PA_IDXSET_FOREACH(o, s->outputs, idx) {
 989         pa_source_output_state_t st;
 990
 991         st = pa_source_output_get_state(o);
 992         pa_assert(PA_SOURCE_OUTPUT_IS_LINKED(st));
 993
 994         if (st == PA_SOURCE_OUTPUT_CORKED)
 995             continue;
 996
 997         if (o->flags & PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND)
 998             continue;
 999
1000         ret ++;
1001     }
1002
1003     return ret;
1004 }
1005
1006 /* Called from IO thread, except when it is not */
1007 int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
1008     pa_source *s = PA_SOURCE(object);
1009     pa_source_assert_ref(s);
1010
1011     switch ((pa_source_message_t) code) {
1012
1013         case PA_SOURCE_MESSAGE_ADD_OUTPUT: {
1014             pa_source_output *o = PA_SOURCE_OUTPUT(userdata);
1015
1016             pa_hashmap_put(s->thread_info.outputs, PA_UINT32_TO_PTR(o->index), pa_source_output_ref(o));
1017
1018             if (o->direct_on_input) {
1019                 o->thread_info.direct_on_input = o->direct_on_input;
1020                 pa_hashmap_put(o->thread_info.direct_on_input->thread_info.direct_outputs, PA_UINT32_TO_PTR(o->index), o);
1021             }
1022
1023             pa_assert(!o->thread_info.attached);
1024             o->thread_info.attached = TRUE;
1025
1026             if (o->attach)
1027                 o->attach(o);
1028
1029             pa_source_output_set_state_within_thread(o, o->state);
1030
1031             if (o->thread_info.requested_source_latency != (pa_usec_t) -1)
1032                 pa_source_output_set_requested_latency_within_thread(o, o->thread_info.requested_source_latency);
1033
1034             pa_source_output_update_max_rewind(o, s->thread_info.max_rewind);
1035
1036             /* We don't just invalidate the requested latency here,
1037              * because if we are in a move we might need to fix up the
1038              * requested latency. */
1039             pa_source_output_set_requested_latency_within_thread(o, o->thread_info.requested_source_latency);
1040
1041             return 0;
1042         }
1043
1044         case PA_SOURCE_MESSAGE_REMOVE_OUTPUT: {
1045             pa_source_output *o = PA_SOURCE_OUTPUT(userdata);
1046
1047             pa_source_output_set_state_within_thread(o, o->state);
1048
1049             if (o->detach)
1050                 o->detach(o);
1051
1052             pa_assert(o->thread_info.attached);
1053             o->thread_info.attached = FALSE;
1054
1055             if (o->thread_info.direct_on_input) {
1056                 pa_hashmap_remove(o->thread_info.direct_on_input->thread_info.direct_outputs, PA_UINT32_TO_PTR(o->index));
1057                 o->thread_info.direct_on_input = NULL;
1058             }
1059
1060             if (pa_hashmap_remove(s->thread_info.outputs, PA_UINT32_TO_PTR(o->index)))
1061                 pa_source_output_unref(o);
1062
1063             pa_source_invalidate_requested_latency(s, TRUE);
1064
1065             return 0;
1066         }
1067
1068         case PA_SOURCE_MESSAGE_SET_VOLUME:
1069             s->thread_info.soft_volume = s->soft_volume;
1070             return 0;
1071
1072         case PA_SOURCE_MESSAGE_GET_VOLUME:
1073             return 0;
1074
1075         case PA_SOURCE_MESSAGE_SET_MUTE:
1076             s->thread_info.soft_muted = s->muted;
1077             return 0;
1078
1079         case PA_SOURCE_MESSAGE_GET_MUTE:
1080             return 0;
1081
1082         case PA_SOURCE_MESSAGE_SET_STATE: {
1083
1084             pa_bool_t suspend_change =
1085                 (s->thread_info.state == PA_SOURCE_SUSPENDED && PA_SOURCE_IS_OPENED(PA_PTR_TO_UINT(userdata))) ||
1086                 (PA_SOURCE_IS_OPENED(s->thread_info.state) && PA_PTR_TO_UINT(userdata) == PA_SOURCE_SUSPENDED);
1087
1088             s->thread_info.state = PA_PTR_TO_UINT(userdata);
1089
1090             if (suspend_change) {
1091                 pa_source_output *o;
1092                 void *state = NULL;
1093
1094                 while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1095                     if (o->suspend_within_thread)
1096                         o->suspend_within_thread(o, s->thread_info.state == PA_SOURCE_SUSPENDED);
1097             }
1098
1099
1100             return 0;
1101         }
1102
1103         case PA_SOURCE_MESSAGE_DETACH:
1104
1105             /* Detach all streams */
1106             pa_source_detach_within_thread(s);
1107             return 0;
1108
1109         case PA_SOURCE_MESSAGE_ATTACH:
1110
1111             /* Reattach all streams */
1112             pa_source_attach_within_thread(s);
1113             return 0;
1114
1115         case PA_SOURCE_MESSAGE_GET_REQUESTED_LATENCY: {
1116
1117             pa_usec_t *usec = userdata;
1118             *usec = pa_source_get_requested_latency_within_thread(s);
1119
1120             if (*usec == (pa_usec_t) -1)
1121                 *usec = s->thread_info.max_latency;
1122
1123             return 0;
1124         }
1125
1126         case PA_SOURCE_MESSAGE_SET_LATENCY_RANGE: {
1127             pa_usec_t *r = userdata;
1128
1129             pa_source_set_latency_range_within_thread(s, r[0], r[1]);
1130
1131             return 0;
1132         }
1133
1134         case PA_SOURCE_MESSAGE_GET_LATENCY_RANGE: {
1135             pa_usec_t *r = userdata;
1136
1137             r[0] = s->thread_info.min_latency;
1138             r[1] = s->thread_info.max_latency;
1139
1140             return 0;
1141         }
1142
1143         case PA_SOURCE_MESSAGE_GET_FIXED_LATENCY:
1144
1145             *((pa_usec_t*) userdata) = s->thread_info.fixed_latency;
1146             return 0;
1147
1148         case PA_SOURCE_MESSAGE_SET_FIXED_LATENCY:
1149
1150             pa_source_set_fixed_latency_within_thread(s, (pa_usec_t) offset);
1151             return 0;
1152
1153         case PA_SOURCE_MESSAGE_GET_MAX_REWIND:
1154
1155             *((size_t*) userdata) = s->thread_info.max_rewind;
1156             return 0;
1157
1158         case PA_SOURCE_MESSAGE_SET_MAX_REWIND:
1159
1160             pa_source_set_max_rewind_within_thread(s, (size_t) offset);
1161             return 0;
1162
1163         case PA_SOURCE_MESSAGE_GET_LATENCY:
1164
1165             if (s->monitor_of) {
1166                 *((pa_usec_t*) userdata) = 0;
1167                 return 0;
1168             }
1169
1170             /* Implementors need to overwrite this implementation! */
1171             return -1;
1172
1173         case PA_SOURCE_MESSAGE_MAX:
1174             ;
1175     }
1176
1177     return -1;
1178 }
1179
1180 /* Called from main thread */
1181 int pa_source_suspend_all(pa_core *c, pa_bool_t suspend, pa_suspend_cause_t cause) {
1182     uint32_t idx;
1183     pa_source *source;
1184     int ret = 0;
1185
1186     pa_core_assert_ref(c);
1187     pa_assert_ctl_context();
1188     pa_assert(cause != 0);
1189
1190     for (source = PA_SOURCE(pa_idxset_first(c->sources, &idx)); source; source = PA_SOURCE(pa_idxset_next(c->sources, &idx))) {
1191         int r;
1192
1193         if (source->monitor_of)
1194             continue;
1195
1196         if ((r = pa_source_suspend(source, suspend, cause)) < 0)
1197             ret = r;
1198     }
1199
1200     return ret;
1201 }
1202
1203 /* Called from main thread */
1204 void pa_source_detach(pa_source *s) {
1205     pa_source_assert_ref(s);
1206     pa_assert_ctl_context();
1207     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1208
1209     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_DETACH, NULL, 0, NULL) == 0);
1210 }
1211
1212 /* Called from main thread */
1213 void pa_source_attach(pa_source *s) {
1214     pa_source_assert_ref(s);
1215     pa_assert_ctl_context();
1216     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1217
1218     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_ATTACH, NULL, 0, NULL) == 0);
1219 }
1220
1221 /* Called from IO thread */
1222 void pa_source_detach_within_thread(pa_source *s) {
1223     pa_source_output *o;
1224     void *state = NULL;
1225
1226     pa_source_assert_ref(s);
1227     pa_source_assert_io_context(s);
1228     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
1229
1230     PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1231         if (o->detach)
1232             o->detach(o);
1233 }
1234
1235 /* Called from IO thread */
1236 void pa_source_attach_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->attach)
1246             o->attach(o);
1247 }
1248
1249 /* Called from IO thread */
1250 pa_usec_t pa_source_get_requested_latency_within_thread(pa_source *s) {
1251     pa_usec_t result = (pa_usec_t) -1;
1252     pa_source_output *o;
1253     void *state = NULL;
1254
1255     pa_source_assert_ref(s);
1256     pa_source_assert_io_context(s);
1257
1258     if (!(s->flags & PA_SOURCE_DYNAMIC_LATENCY))
1259         return PA_CLAMP(s->thread_info.fixed_latency, s->thread_info.min_latency, s->thread_info.max_latency);
1260
1261     if (s->thread_info.requested_latency_valid)
1262         return s->thread_info.requested_latency;
1263
1264     PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1265         if (o->thread_info.requested_source_latency != (pa_usec_t) -1 &&
1266             (result == (pa_usec_t) -1 || result > o->thread_info.requested_source_latency))
1267             result = o->thread_info.requested_source_latency;
1268
1269     if (result != (pa_usec_t) -1)
1270         result = PA_CLAMP(result, s->thread_info.min_latency, s->thread_info.max_latency);
1271
1272     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1273         /* Only cache this if we are fully set up */
1274         s->thread_info.requested_latency = result;
1275         s->thread_info.requested_latency_valid = TRUE;
1276     }
1277
1278     return result;
1279 }
1280
1281 /* Called from main thread */
1282 pa_usec_t pa_source_get_requested_latency(pa_source *s) {
1283     pa_usec_t usec = 0;
1284
1285     pa_source_assert_ref(s);
1286     pa_assert_ctl_context();
1287     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1288
1289     if (s->state == PA_SOURCE_SUSPENDED)
1290         return 0;
1291
1292     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
1293
1294     return usec;
1295 }
1296
1297 /* Called from IO thread */
1298 void pa_source_set_max_rewind_within_thread(pa_source *s, size_t max_rewind) {
1299     pa_source_output *o;
1300     void *state = NULL;
1301
1302     pa_source_assert_ref(s);
1303     pa_source_assert_io_context(s);
1304
1305     if (max_rewind == s->thread_info.max_rewind)
1306         return;
1307
1308     s->thread_info.max_rewind = max_rewind;
1309
1310     if (PA_SOURCE_IS_LINKED(s->thread_info.state))
1311         PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1312             pa_source_output_update_max_rewind(o, s->thread_info.max_rewind);
1313 }
1314
1315 /* Called from main thread */
1316 void pa_source_set_max_rewind(pa_source *s, size_t max_rewind) {
1317     pa_source_assert_ref(s);
1318     pa_assert_ctl_context();
1319
1320     if (PA_SOURCE_IS_LINKED(s->state))
1321         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MAX_REWIND, NULL, max_rewind, NULL) == 0);
1322     else
1323         pa_source_set_max_rewind_within_thread(s, max_rewind);
1324 }
1325
1326 /* Called from IO thread */
1327 void pa_source_invalidate_requested_latency(pa_source *s, pa_bool_t dynamic) {
1328     pa_source_output *o;
1329     void *state = NULL;
1330
1331     pa_source_assert_ref(s);
1332     pa_source_assert_io_context(s);
1333
1334     if ((s->flags & PA_SOURCE_DYNAMIC_LATENCY))
1335         s->thread_info.requested_latency_valid = FALSE;
1336     else if (dynamic)
1337         return;
1338
1339     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1340
1341         if (s->update_requested_latency)
1342             s->update_requested_latency(s);
1343
1344         while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1345             if (o->update_source_requested_latency)
1346                 o->update_source_requested_latency(o);
1347     }
1348
1349     if (s->monitor_of)
1350         pa_sink_invalidate_requested_latency(s->monitor_of, dynamic);
1351 }
1352
1353 /* Called from main thread */
1354 void pa_source_set_latency_range(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency) {
1355     pa_source_assert_ref(s);
1356     pa_assert_ctl_context();
1357
1358     /* min_latency == 0:           no limit
1359      * min_latency anything else:  specified limit
1360      *
1361      * Similar for max_latency */
1362
1363     if (min_latency < ABSOLUTE_MIN_LATENCY)
1364         min_latency = ABSOLUTE_MIN_LATENCY;
1365
1366     if (max_latency <= 0 ||
1367         max_latency > ABSOLUTE_MAX_LATENCY)
1368         max_latency = ABSOLUTE_MAX_LATENCY;
1369
1370     pa_assert(min_latency <= max_latency);
1371
1372     /* Hmm, let's see if someone forgot to set PA_SOURCE_DYNAMIC_LATENCY here... */
1373     pa_assert((min_latency == ABSOLUTE_MIN_LATENCY &&
1374                max_latency == ABSOLUTE_MAX_LATENCY) ||
1375               (s->flags & PA_SOURCE_DYNAMIC_LATENCY));
1376
1377     if (PA_SOURCE_IS_LINKED(s->state)) {
1378         pa_usec_t r[2];
1379
1380         r[0] = min_latency;
1381         r[1] = max_latency;
1382
1383         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_LATENCY_RANGE, r, 0, NULL) == 0);
1384     } else
1385         pa_source_set_latency_range_within_thread(s, min_latency, max_latency);
1386 }
1387
1388 /* Called from main thread */
1389 void pa_source_get_latency_range(pa_source *s, pa_usec_t *min_latency, pa_usec_t *max_latency) {
1390    pa_source_assert_ref(s);
1391    pa_assert_ctl_context();
1392    pa_assert(min_latency);
1393    pa_assert(max_latency);
1394
1395    if (PA_SOURCE_IS_LINKED(s->state)) {
1396        pa_usec_t r[2] = { 0, 0 };
1397
1398        pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY_RANGE, r, 0, NULL) == 0);
1399
1400        *min_latency = r[0];
1401        *max_latency = r[1];
1402    } else {
1403        *min_latency = s->thread_info.min_latency;
1404        *max_latency = s->thread_info.max_latency;
1405    }
1406 }
1407
1408 /* Called from IO thread, and from main thread before pa_source_put() is called */
1409 void pa_source_set_latency_range_within_thread(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency) {
1410     pa_source_assert_ref(s);
1411     pa_source_assert_io_context(s);
1412
1413     pa_assert(min_latency >= ABSOLUTE_MIN_LATENCY);
1414     pa_assert(max_latency <= ABSOLUTE_MAX_LATENCY);
1415     pa_assert(min_latency <= max_latency);
1416
1417     /* Hmm, let's see if someone forgot to set PA_SOURCE_DYNAMIC_LATENCY here... */
1418     pa_assert((min_latency == ABSOLUTE_MIN_LATENCY &&
1419                max_latency == ABSOLUTE_MAX_LATENCY) ||
1420               (s->flags & PA_SOURCE_DYNAMIC_LATENCY) ||
1421               s->monitor_of);
1422
1423     if (s->thread_info.min_latency == min_latency &&
1424         s->thread_info.max_latency == max_latency)
1425         return;
1426
1427     s->thread_info.min_latency = min_latency;
1428     s->thread_info.max_latency = max_latency;
1429
1430     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1431         pa_source_output *o;
1432         void *state = NULL;
1433
1434         PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1435             if (o->update_source_latency_range)
1436                 o->update_source_latency_range(o);
1437     }
1438
1439     pa_source_invalidate_requested_latency(s, FALSE);
1440 }
1441
1442 /* Called from main thread, before the source is put */
1443 void pa_source_set_fixed_latency(pa_source *s, pa_usec_t latency) {
1444     pa_source_assert_ref(s);
1445     pa_assert_ctl_context();
1446
1447     if (s->flags & PA_SOURCE_DYNAMIC_LATENCY) {
1448         pa_assert(latency == 0);
1449         return;
1450     }
1451
1452     if (latency < ABSOLUTE_MIN_LATENCY)
1453         latency = ABSOLUTE_MIN_LATENCY;
1454
1455     if (latency > ABSOLUTE_MAX_LATENCY)
1456         latency = ABSOLUTE_MAX_LATENCY;
1457
1458     if (PA_SOURCE_IS_LINKED(s->state))
1459         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_FIXED_LATENCY, NULL, (int64_t) latency, NULL) == 0);
1460     else
1461         s->thread_info.fixed_latency = latency;
1462 }
1463
1464 /* Called from main thread */
1465 pa_usec_t pa_source_get_fixed_latency(pa_source *s) {
1466     pa_usec_t latency;
1467
1468     pa_source_assert_ref(s);
1469     pa_assert_ctl_context();
1470
1471     if (s->flags & PA_SOURCE_DYNAMIC_LATENCY)
1472         return 0;
1473
1474     if (PA_SOURCE_IS_LINKED(s->state))
1475         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_FIXED_LATENCY, &latency, 0, NULL) == 0);
1476     else
1477         latency = s->thread_info.fixed_latency;
1478
1479     return latency;
1480 }
1481
1482 /* Called from IO thread */
1483 void pa_source_set_fixed_latency_within_thread(pa_source *s, pa_usec_t latency) {
1484     pa_source_assert_ref(s);
1485     pa_source_assert_io_context(s);
1486
1487     if (s->flags & PA_SOURCE_DYNAMIC_LATENCY) {
1488         pa_assert(latency == 0);
1489         return;
1490     }
1491
1492     pa_assert(latency >= ABSOLUTE_MIN_LATENCY);
1493     pa_assert(latency <= ABSOLUTE_MAX_LATENCY);
1494
1495     if (s->thread_info.fixed_latency == latency)
1496         return;
1497
1498     s->thread_info.fixed_latency = latency;
1499
1500     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1501         pa_source_output *o;
1502         void *state = NULL;
1503
1504         PA_HASHMAP_FOREACH(o, s->thread_info.outputs, state)
1505             if (o->update_source_fixed_latency)
1506                 o->update_source_fixed_latency(o);
1507     }
1508
1509     pa_source_invalidate_requested_latency(s, FALSE);
1510 }
1511
1512 /* Called from main thread */
1513 size_t pa_source_get_max_rewind(pa_source *s) {
1514     size_t r;
1515     pa_assert_ctl_context();
1516     pa_source_assert_ref(s);
1517
1518     if (!PA_SOURCE_IS_LINKED(s->state))
1519         return s->thread_info.max_rewind;
1520
1521     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MAX_REWIND, &r, 0, NULL) == 0);
1522
1523     return r;
1524 }
1525
1526 /* Called from main context */
1527 int pa_source_set_port(pa_source *s, const char *name, pa_bool_t save) {
1528     pa_device_port *port;
1529
1530     pa_assert(s);
1531     pa_assert_ctl_context();
1532
1533     if (!s->set_port) {
1534         pa_log_debug("set_port() operation not implemented for source %u \"%s\"", s->index, s->name);
1535         return -PA_ERR_NOTIMPLEMENTED;
1536     }
1537
1538     if (!s->ports)
1539         return -PA_ERR_NOENTITY;
1540
1541     if (!(port = pa_hashmap_get(s->ports, name)))
1542         return -PA_ERR_NOENTITY;
1543
1544     if (s->active_port == port) {
1545         s->save_port = s->save_port || save;
1546         return 0;
1547     }
1548
1549     if ((s->set_port(s, port)) < 0)
1550         return -PA_ERR_NOENTITY;
1551
1552     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
1553
1554     pa_log_info("Changed port of source %u \"%s\" to %s", s->index, s->name, port->name);
1555
1556     s->active_port = port;
1557     s->save_port = save;
1558
1559     return 0;
1560 }