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/source-output.h>
  37 #include <pulsecore/namereg.h>
  38 #include <pulsecore/core-subscribe.h>
  39 #include <pulsecore/log.h>
  40 #include <pulsecore/sample-util.h>
  41
  42 #include "source.h"
  43
  44 #define ABSOLUTE_MIN_LATENCY (500)
  45 #define ABSOLUTE_MAX_LATENCY (10*PA_USEC_PER_SEC)
  46
  47 static PA_DEFINE_CHECK_TYPE(pa_source, pa_msgobject);
  48
  49 static void source_free(pa_object *o);
  50
  51 pa_source_new_data* pa_source_new_data_init(pa_source_new_data *data) {
  52     pa_assert(data);
  53
  54     memset(data, 0, sizeof(*data));
  55     data->proplist = pa_proplist_new();
  56
  57     return data;
  58 }
  59
  60 void pa_source_new_data_set_name(pa_source_new_data *data, const char *name) {
  61     pa_assert(data);
  62
  63     pa_xfree(data->name);
  64     data->name = pa_xstrdup(name);
  65 }
  66
  67 void pa_source_new_data_set_sample_spec(pa_source_new_data *data, const pa_sample_spec *spec) {
  68     pa_assert(data);
  69
  70     if ((data->sample_spec_is_set = !!spec))
  71         data->sample_spec = *spec;
  72 }
  73
  74 void pa_source_new_data_set_channel_map(pa_source_new_data *data, const pa_channel_map *map) {
  75     pa_assert(data);
  76
  77     if ((data->channel_map_is_set = !!map))
  78         data->channel_map = *map;
  79 }
  80
  81 void pa_source_new_data_set_volume(pa_source_new_data *data, const pa_cvolume *volume) {
  82     pa_assert(data);
  83
  84     if ((data->volume_is_set = !!volume))
  85         data->volume = *volume;
  86 }
  87
  88 void pa_source_new_data_set_muted(pa_source_new_data *data, pa_bool_t mute) {
  89     pa_assert(data);
  90
  91     data->muted_is_set = TRUE;
  92     data->muted = !!mute;
  93 }
  94
  95 void pa_source_new_data_done(pa_source_new_data *data) {
  96     pa_assert(data);
  97
  98     pa_xfree(data->name);
  99     pa_proplist_free(data->proplist);
 100 }
 101
 102 /* Called from main context */
 103 static void reset_callbacks(pa_source *s) {
 104     pa_assert(s);
 105
 106     s->set_state = NULL;
 107     s->get_volume = NULL;
 108     s->set_volume = NULL;
 109     s->get_mute = NULL;
 110     s->set_mute = NULL;
 111     s->update_requested_latency = NULL;
 112 }
 113
 114 /* Called from main context */
 115 pa_source* pa_source_new(
 116         pa_core *core,
 117         pa_source_new_data *data,
 118         pa_source_flags_t flags) {
 119
 120     pa_source *s;
 121     const char *name;
 122     char st[PA_SAMPLE_SPEC_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX];
 123     char *pt;
 124
 125     pa_assert(core);
 126     pa_assert(data);
 127     pa_assert(data->name);
 128
 129     s = pa_msgobject_new(pa_source);
 130
 131     if (!(name = pa_namereg_register(core, data->name, PA_NAMEREG_SOURCE, s, data->namereg_fail))) {
 132         pa_xfree(s);
 133         return NULL;
 134     }
 135
 136     pa_source_new_data_set_name(data, name);
 137
 138     if (pa_hook_fire(&core->hooks[PA_CORE_HOOK_SOURCE_NEW], data) < 0) {
 139         pa_xfree(s);
 140         pa_namereg_unregister(core, name);
 141         return NULL;
 142     }
 143
 144     pa_return_null_if_fail(!data->driver || pa_utf8_valid(data->driver));
 145     pa_return_null_if_fail(data->name && pa_utf8_valid(data->name) && data->name[0]);
 146
 147     pa_return_null_if_fail(data->sample_spec_is_set && pa_sample_spec_valid(&data->sample_spec));
 148
 149     if (!data->channel_map_is_set)
 150         pa_return_null_if_fail(pa_channel_map_init_auto(&data->channel_map, data->sample_spec.channels, PA_CHANNEL_MAP_DEFAULT));
 151
 152     pa_return_null_if_fail(pa_channel_map_valid(&data->channel_map));
 153     pa_return_null_if_fail(data->channel_map.channels == data->sample_spec.channels);
 154
 155     if (!data->volume_is_set)
 156         pa_cvolume_reset(&data->volume, data->sample_spec.channels);
 157
 158     pa_return_null_if_fail(pa_cvolume_valid(&data->volume));
 159     pa_return_null_if_fail(data->volume.channels == data->sample_spec.channels);
 160
 161     if (!data->muted_is_set)
 162         data->muted = FALSE;
 163
 164     if (data->card)
 165         pa_proplist_update(data->proplist, PA_UPDATE_MERGE, data->card->proplist);
 166
 167     pa_device_init_description(data->proplist);
 168     pa_device_init_icon(data->proplist, FALSE);
 169
 170     if (pa_hook_fire(&core->hooks[PA_CORE_HOOK_SOURCE_FIXATE], data) < 0) {
 171         pa_xfree(s);
 172         pa_namereg_unregister(core, name);
 173         return NULL;
 174     }
 175
 176     s->parent.parent.free = source_free;
 177     s->parent.process_msg = pa_source_process_msg;
 178
 179     s->core = core;
 180     s->state = PA_SOURCE_INIT;
 181     s->flags = flags;
 182     s->name = pa_xstrdup(name);
 183     s->proplist = pa_proplist_copy(data->proplist);
 184     s->driver = pa_xstrdup(pa_path_get_filename(data->driver));
 185     s->module = data->module;
 186     s->card = data->card;
 187
 188     s->sample_spec = data->sample_spec;
 189     s->channel_map = data->channel_map;
 190
 191     s->outputs = pa_idxset_new(NULL, NULL);
 192     s->n_corked = 0;
 193     s->monitor_of = NULL;
 194
 195     s->virtual_volume = data->volume;
 196     pa_cvolume_reset(&s->soft_volume, s->sample_spec.channels);
 197     s->base_volume = PA_VOLUME_NORM;
 198     s->n_volume_steps = PA_VOLUME_NORM+1;
 199     s->muted = data->muted;
 200     s->refresh_volume = s->refresh_muted = FALSE;
 201
 202     reset_callbacks(s);
 203     s->userdata = NULL;
 204
 205     s->asyncmsgq = NULL;
 206     s->rtpoll = NULL;
 207
 208     pa_silence_memchunk_get(
 209             &core->silence_cache,
 210             core->mempool,
 211             &s->silence,
 212             &s->sample_spec,
 213             0);
 214
 215     s->thread_info.outputs = pa_hashmap_new(pa_idxset_trivial_hash_func, pa_idxset_trivial_compare_func);
 216     s->thread_info.soft_volume = s->soft_volume;
 217     s->thread_info.soft_muted = s->muted;
 218     s->thread_info.state = s->state;
 219     s->thread_info.max_rewind = 0;
 220     s->thread_info.requested_latency_valid = FALSE;
 221     s->thread_info.requested_latency = 0;
 222     s->thread_info.min_latency = ABSOLUTE_MIN_LATENCY;
 223     s->thread_info.max_latency = ABSOLUTE_MAX_LATENCY;
 224
 225     pa_assert_se(pa_idxset_put(core->sources, s, &s->index) >= 0);
 226
 227     if (s->card)
 228         pa_assert_se(pa_idxset_put(s->card->sources, s, NULL) >= 0);
 229
 230     pt = pa_proplist_to_string_sep(s->proplist, "\n    ");
 231     pa_log_info("Created source %u \"%s\" with sample spec %s and channel map %s\n    %s",
 232                 s->index,
 233                 s->name,
 234                 pa_sample_spec_snprint(st, sizeof(st), &s->sample_spec),
 235                 pa_channel_map_snprint(cm, sizeof(cm), &s->channel_map),
 236                 pt);
 237     pa_xfree(pt);
 238
 239     return s;
 240 }
 241
 242 /* Called from main context */
 243 static int source_set_state(pa_source *s, pa_source_state_t state) {
 244     int ret;
 245     pa_bool_t suspend_change;
 246     pa_source_state_t original_state;
 247
 248     pa_assert(s);
 249
 250     if (s->state == state)
 251         return 0;
 252
 253     original_state = s->state;
 254
 255     suspend_change =
 256         (original_state == PA_SOURCE_SUSPENDED && PA_SOURCE_IS_OPENED(state)) ||
 257         (PA_SOURCE_IS_OPENED(original_state) && state == PA_SOURCE_SUSPENDED);
 258
 259     if (s->set_state)
 260         if ((ret = s->set_state(s, state)) < 0)
 261             return ret;
 262
 263     if (s->asyncmsgq)
 264         if ((ret = pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL)) < 0) {
 265
 266             if (s->set_state)
 267                 s->set_state(s, original_state);
 268
 269             return ret;
 270         }
 271
 272     s->state = state;
 273
 274     if (state != PA_SOURCE_UNLINKED) { /* if we enter UNLINKED state pa_source_unlink() will fire the apropriate events */
 275         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_STATE_CHANGED], s);
 276         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE | PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 277     }
 278
 279     if (suspend_change) {
 280         pa_source_output *o;
 281         uint32_t idx;
 282
 283         /* We're suspending or resuming, tell everyone about it */
 284
 285         for (o = PA_SOURCE_OUTPUT(pa_idxset_first(s->outputs, &idx)); o; o = PA_SOURCE_OUTPUT(pa_idxset_next(s->outputs, &idx)))
 286             if (s->state == PA_SOURCE_SUSPENDED &&
 287                 (o->flags & PA_SOURCE_OUTPUT_FAIL_ON_SUSPEND))
 288                 pa_source_output_kill(o);
 289             else if (o->suspend)
 290                 o->suspend(o, state == PA_SOURCE_SUSPENDED);
 291     }
 292
 293
 294     return 0;
 295 }
 296
 297 /* Called from main context */
 298 void pa_source_put(pa_source *s) {
 299     pa_source_assert_ref(s);
 300
 301     pa_assert(s->state == PA_SOURCE_INIT);
 302
 303     /* The following fields must be initialized properly when calling _put() */
 304     pa_assert(s->asyncmsgq);
 305     pa_assert(s->rtpoll);
 306     pa_assert(!s->thread_info.min_latency || !s->thread_info.max_latency ||
 307               s->thread_info.min_latency <= s->thread_info.max_latency);
 308
 309     if (!(s->flags & PA_SOURCE_HW_VOLUME_CTRL)) {
 310         s->flags |= PA_SOURCE_DECIBEL_VOLUME;
 311
 312         s->thread_info.soft_volume = s->soft_volume;
 313         s->thread_info.soft_muted = s->muted;
 314     }
 315
 316     if (s->flags & PA_SOURCE_DECIBEL_VOLUME)
 317         s->n_volume_steps = PA_VOLUME_NORM+1;
 318
 319     pa_assert_se(source_set_state(s, PA_SOURCE_IDLE) == 0);
 320
 321     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE | PA_SUBSCRIPTION_EVENT_NEW, s->index);
 322     pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_PUT], s);
 323 }
 324
 325 /* Called from main context */
 326 void pa_source_unlink(pa_source *s) {
 327     pa_bool_t linked;
 328     pa_source_output *o, *j = NULL;
 329
 330     pa_assert(s);
 331
 332     /* See pa_sink_unlink() for a couple of comments how this function
 333      * works. */
 334
 335     linked = PA_SOURCE_IS_LINKED(s->state);
 336
 337     if (linked)
 338         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_UNLINK], s);
 339
 340     if (s->state != PA_SOURCE_UNLINKED)
 341         pa_namereg_unregister(s->core, s->name);
 342     pa_idxset_remove_by_data(s->core->sources, s, NULL);
 343
 344     if (s->card)
 345         pa_idxset_remove_by_data(s->card->sources, s, NULL);
 346
 347     while ((o = pa_idxset_first(s->outputs, NULL))) {
 348         pa_assert(o != j);
 349         pa_source_output_kill(o);
 350         j = o;
 351     }
 352
 353     if (linked)
 354         source_set_state(s, PA_SOURCE_UNLINKED);
 355     else
 356         s->state = PA_SOURCE_UNLINKED;
 357
 358     reset_callbacks(s);
 359
 360     if (linked) {
 361         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE | PA_SUBSCRIPTION_EVENT_REMOVE, s->index);
 362         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_UNLINK_POST], s);
 363     }
 364 }
 365
 366 /* Called from main context */
 367 static void source_free(pa_object *o) {
 368     pa_source_output *so;
 369     pa_source *s = PA_SOURCE(o);
 370
 371     pa_assert(s);
 372     pa_assert(pa_source_refcnt(s) == 0);
 373
 374     if (PA_SOURCE_IS_LINKED(s->state))
 375         pa_source_unlink(s);
 376
 377     pa_log_info("Freeing source %u \"%s\"", s->index, s->name);
 378
 379     pa_idxset_free(s->outputs, NULL, NULL);
 380
 381     while ((so = pa_hashmap_steal_first(s->thread_info.outputs)))
 382         pa_source_output_unref(so);
 383
 384     pa_hashmap_free(s->thread_info.outputs, NULL, NULL);
 385
 386     if (s->silence.memblock)
 387         pa_memblock_unref(s->silence.memblock);
 388
 389     pa_xfree(s->name);
 390     pa_xfree(s->driver);
 391
 392     if (s->proplist)
 393         pa_proplist_free(s->proplist);
 394
 395     pa_xfree(s);
 396 }
 397
 398 /* Called from main context */
 399 void pa_source_set_asyncmsgq(pa_source *s, pa_asyncmsgq *q) {
 400     pa_source_assert_ref(s);
 401
 402     s->asyncmsgq = q;
 403 }
 404
 405 /* Called from main context */
 406 void pa_source_set_rtpoll(pa_source *s, pa_rtpoll *p) {
 407     pa_source_assert_ref(s);
 408
 409     s->rtpoll = p;
 410 }
 411
 412 /* Called from main context */
 413 int pa_source_update_status(pa_source*s) {
 414     pa_source_assert_ref(s);
 415     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 416
 417     if (s->state == PA_SOURCE_SUSPENDED)
 418         return 0;
 419
 420     return source_set_state(s, pa_source_used_by(s) ? PA_SOURCE_RUNNING : PA_SOURCE_IDLE);
 421 }
 422
 423 /* Called from main context */
 424 int pa_source_suspend(pa_source *s, pa_bool_t suspend) {
 425     pa_source_assert_ref(s);
 426     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 427
 428     if (s->monitor_of)
 429         return -PA_ERR_NOTSUPPORTED;
 430
 431     if (suspend)
 432         return source_set_state(s, PA_SOURCE_SUSPENDED);
 433     else
 434         return source_set_state(s, pa_source_used_by(s) ? PA_SOURCE_RUNNING : PA_SOURCE_IDLE);
 435 }
 436
 437 /* Called from main context */
 438 int pa_source_sync_suspend(pa_source *s) {
 439     pa_sink_state_t state;
 440
 441     pa_source_assert_ref(s);
 442     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 443     pa_assert(s->monitor_of);
 444
 445     state = pa_sink_get_state(s->monitor_of);
 446
 447     if (state == PA_SINK_SUSPENDED)
 448         return source_set_state(s, PA_SOURCE_SUSPENDED);
 449
 450     pa_assert(PA_SINK_IS_OPENED(state));
 451
 452     return source_set_state(s, pa_source_used_by(s) ? PA_SOURCE_RUNNING : PA_SOURCE_IDLE);
 453 }
 454
 455 /* Called from main context */
 456 pa_queue *pa_source_move_all_start(pa_source *s) {
 457     pa_queue *q;
 458     pa_source_output *o, *n;
 459     uint32_t idx;
 460
 461     pa_source_assert_ref(s);
 462     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 463
 464     q = pa_queue_new();
 465
 466     for (o = PA_SOURCE_OUTPUT(pa_idxset_first(s->outputs, &idx)); o; o = n) {
 467         n = PA_SOURCE_OUTPUT(pa_idxset_next(s->outputs, &idx));
 468
 469         pa_source_output_ref(o);
 470
 471         if (pa_source_output_start_move(o) >= 0)
 472             pa_queue_push(q, o);
 473         else
 474             pa_source_output_unref(o);
 475     }
 476
 477     return q;
 478 }
 479
 480 /* Called from main context */
 481 void pa_source_move_all_finish(pa_source *s, pa_queue *q, pa_bool_t save) {
 482     pa_source_output *o;
 483
 484     pa_source_assert_ref(s);
 485     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 486     pa_assert(q);
 487
 488     while ((o = PA_SOURCE_OUTPUT(pa_queue_pop(q)))) {
 489         if (pa_source_output_finish_move(o, s, save) < 0)
 490             pa_source_output_kill(o);
 491
 492         pa_source_output_unref(o);
 493     }
 494
 495     pa_queue_free(q, NULL, NULL);
 496 }
 497
 498 /* Called from main context */
 499 void pa_source_move_all_fail(pa_queue *q) {
 500     pa_source_output *o;
 501     pa_assert(q);
 502
 503     while ((o = PA_SOURCE_OUTPUT(pa_queue_pop(q)))) {
 504         if (pa_hook_fire(&o->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_MOVE_FAIL], o) == PA_HOOK_OK) {
 505             pa_source_output_kill(o);
 506             pa_source_output_unref(o);
 507         }
 508     }
 509
 510     pa_queue_free(q, NULL, NULL);
 511 }
 512
 513 /* Called from IO thread context */
 514 void pa_source_process_rewind(pa_source *s, size_t nbytes) {
 515     pa_source_output *o;
 516     void *state = NULL;
 517
 518     pa_source_assert_ref(s);
 519     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
 520
 521     if (s->thread_info.state == PA_SOURCE_SUSPENDED)
 522         return;
 523
 524     if (nbytes <= 0)
 525         return;
 526
 527     pa_log_debug("Processing rewind...");
 528
 529     while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL))) {
 530         pa_source_output_assert_ref(o);
 531         pa_source_output_process_rewind(o, nbytes);
 532     }
 533 }
 534
 535 /* Called from IO thread context */
 536 void pa_source_post(pa_source*s, const pa_memchunk *chunk) {
 537     pa_source_output *o;
 538     void *state = NULL;
 539
 540     pa_source_assert_ref(s);
 541     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
 542     pa_assert(chunk);
 543
 544     if (s->thread_info.state == PA_SOURCE_SUSPENDED)
 545         return;
 546
 547     if (s->thread_info.soft_muted || !pa_cvolume_is_norm(&s->thread_info.soft_volume)) {
 548         pa_memchunk vchunk = *chunk;
 549
 550         pa_memblock_ref(vchunk.memblock);
 551         pa_memchunk_make_writable(&vchunk, 0);
 552
 553         if (s->thread_info.soft_muted || pa_cvolume_is_muted(&s->thread_info.soft_volume))
 554             pa_silence_memchunk(&vchunk, &s->sample_spec);
 555         else
 556             pa_volume_memchunk(&vchunk, &s->sample_spec, &s->thread_info.soft_volume);
 557
 558         while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL))) {
 559             pa_source_output_assert_ref(o);
 560
 561             if (!o->thread_info.direct_on_input)
 562                 pa_source_output_push(o, &vchunk);
 563         }
 564
 565         pa_memblock_unref(vchunk.memblock);
 566     } else {
 567
 568         while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL))) {
 569             pa_source_output_assert_ref(o);
 570
 571             if (!o->thread_info.direct_on_input)
 572                 pa_source_output_push(o, chunk);
 573         }
 574     }
 575 }
 576
 577 /* Called from IO thread context */
 578 void pa_source_post_direct(pa_source*s, pa_source_output *o, const pa_memchunk *chunk) {
 579     pa_source_assert_ref(s);
 580     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
 581     pa_source_output_assert_ref(o);
 582     pa_assert(o->thread_info.direct_on_input);
 583     pa_assert(chunk);
 584
 585     if (s->thread_info.state == PA_SOURCE_SUSPENDED)
 586         return;
 587
 588     if (s->thread_info.soft_muted || !pa_cvolume_is_norm(&s->thread_info.soft_volume)) {
 589         pa_memchunk vchunk = *chunk;
 590
 591         pa_memblock_ref(vchunk.memblock);
 592         pa_memchunk_make_writable(&vchunk, 0);
 593
 594         if (s->thread_info.soft_muted || pa_cvolume_is_muted(&s->thread_info.soft_volume))
 595             pa_silence_memchunk(&vchunk, &s->sample_spec);
 596         else
 597             pa_volume_memchunk(&vchunk, &s->sample_spec, &s->thread_info.soft_volume);
 598
 599         pa_source_output_push(o, &vchunk);
 600
 601         pa_memblock_unref(vchunk.memblock);
 602     } else
 603         pa_source_output_push(o, chunk);
 604 }
 605
 606 /* Called from main thread */
 607 pa_usec_t pa_source_get_latency(pa_source *s) {
 608     pa_usec_t usec;
 609
 610     pa_source_assert_ref(s);
 611     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 612
 613     if (s->state == PA_SOURCE_SUSPENDED)
 614         return 0;
 615
 616     if (!(s->flags & PA_SOURCE_LATENCY))
 617         return 0;
 618
 619     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY, &usec, 0, NULL) == 0);
 620
 621     return usec;
 622 }
 623
 624 /* Called from IO thread */
 625 pa_usec_t pa_source_get_latency_within_thread(pa_source *s) {
 626     pa_usec_t usec = 0;
 627     pa_msgobject *o;
 628
 629     pa_source_assert_ref(s);
 630     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
 631
 632     /* The returned value is supposed to be in the time domain of the sound card! */
 633
 634     if (s->thread_info.state == PA_SOURCE_SUSPENDED)
 635         return 0;
 636
 637     if (!(s->flags & PA_SOURCE_LATENCY))
 638         return 0;
 639
 640     o = PA_MSGOBJECT(s);
 641
 642     /* We probably should make this a proper vtable callback instead of going through process_msg() */
 643
 644     if (o->process_msg(o, PA_SOURCE_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
 645         return -1;
 646
 647     return usec;
 648 }
 649
 650 /* Called from main thread */
 651 void pa_source_set_volume(pa_source *s, const pa_cvolume *volume) {
 652     pa_cvolume old_virtual_volume;
 653     pa_bool_t virtual_volume_changed;
 654
 655     pa_source_assert_ref(s);
 656     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 657     pa_assert(volume);
 658     pa_assert(pa_cvolume_valid(volume));
 659     pa_assert(pa_cvolume_compatible(volume, &s->sample_spec));
 660
 661     old_virtual_volume = s->virtual_volume;
 662     s->virtual_volume = *volume;
 663     virtual_volume_changed = !pa_cvolume_equal(&old_virtual_volume, &s->virtual_volume);
 664
 665     if (s->set_volume) {
 666         pa_cvolume_reset(&s->soft_volume, s->sample_spec.channels);
 667         s->set_volume(s);
 668     } else
 669         s->soft_volume = s->virtual_volume;
 670
 671     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_VOLUME, NULL, 0, NULL) == 0);
 672
 673     if (virtual_volume_changed)
 674         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 675 }
 676
 677 /* Called from main thread. Only to be called by source implementor */
 678 void pa_source_set_soft_volume(pa_source *s, const pa_cvolume *volume) {
 679     pa_source_assert_ref(s);
 680     pa_assert(volume);
 681
 682     if (PA_SOURCE_IS_LINKED(s->state))
 683         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_VOLUME, NULL, 0, NULL) == 0);
 684     else
 685         s->thread_info.soft_volume = *volume;
 686 }
 687
 688 /* Called from main thread */
 689 const pa_cvolume *pa_source_get_volume(pa_source *s, pa_bool_t force_refresh) {
 690     pa_source_assert_ref(s);
 691     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 692
 693     if (s->refresh_volume || force_refresh) {
 694         pa_cvolume old_virtual_volume = s->virtual_volume;
 695
 696         if (s->get_volume)
 697             s->get_volume(s);
 698
 699         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_VOLUME, NULL, 0, NULL) == 0);
 700
 701         if (!pa_cvolume_equal(&old_virtual_volume, &s->virtual_volume))
 702             pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 703     }
 704
 705     return &s->virtual_volume;
 706 }
 707
 708 /* Called from main thread */
 709 void pa_source_volume_changed(pa_source *s, const pa_cvolume *new_volume) {
 710     pa_source_assert_ref(s);
 711
 712     /* The source implementor may call this if the volume changed to make sure everyone is notified */
 713
 714     if (pa_cvolume_equal(&s->virtual_volume, new_volume))
 715         return;
 716
 717     s->virtual_volume = *new_volume;
 718     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 719 }
 720
 721 /* Called from main thread */
 722 void pa_source_set_mute(pa_source *s, pa_bool_t mute) {
 723     pa_bool_t old_muted;
 724
 725     pa_source_assert_ref(s);
 726     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 727
 728     old_muted = s->muted;
 729     s->muted = mute;
 730
 731     if (s->set_mute)
 732         s->set_mute(s);
 733
 734     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MUTE, NULL, 0, NULL) == 0);
 735
 736     if (old_muted != s->muted)
 737         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 738 }
 739
 740 /* Called from main thread */
 741 pa_bool_t pa_source_get_mute(pa_source *s, pa_bool_t force_refresh) {
 742     pa_source_assert_ref(s);
 743     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 744
 745     if (s->refresh_muted || force_refresh) {
 746         pa_bool_t old_muted = s->muted;
 747
 748         if (s->get_mute)
 749             s->get_mute(s);
 750
 751         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MUTE, NULL, 0, NULL) == 0);
 752
 753         if (old_muted != s->muted)
 754             pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 755     }
 756
 757     return s->muted;
 758 }
 759
 760 /* Called from main thread */
 761 void pa_source_mute_changed(pa_source *s, pa_bool_t new_muted) {
 762     pa_source_assert_ref(s);
 763
 764     /* The source implementor may call this if the mute state changed to make sure everyone is notified */
 765
 766     if (s->muted == new_muted)
 767         return;
 768
 769     s->muted = new_muted;
 770     pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 771 }
 772
 773 /* Called from main thread */
 774 pa_bool_t pa_source_update_proplist(pa_source *s, pa_update_mode_t mode, pa_proplist *p) {
 775     pa_source_assert_ref(s);
 776
 777     if (p)
 778         pa_proplist_update(s->proplist, mode, p);
 779
 780     if (PA_SOURCE_IS_LINKED(s->state)) {
 781         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_PROPLIST_CHANGED], s);
 782         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 783     }
 784
 785     return TRUE;
 786 }
 787
 788 /* Called from main thread */
 789 void pa_source_set_description(pa_source *s, const char *description) {
 790     const char *old;
 791     pa_source_assert_ref(s);
 792
 793     if (!description && !pa_proplist_contains(s->proplist, PA_PROP_DEVICE_DESCRIPTION))
 794         return;
 795
 796     old = pa_proplist_gets(s->proplist, PA_PROP_DEVICE_DESCRIPTION);
 797
 798     if (old && description && !strcmp(old, description))
 799         return;
 800
 801     if (description)
 802         pa_proplist_sets(s->proplist, PA_PROP_DEVICE_DESCRIPTION, description);
 803     else
 804         pa_proplist_unset(s->proplist, PA_PROP_DEVICE_DESCRIPTION);
 805
 806     if (PA_SOURCE_IS_LINKED(s->state)) {
 807         pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
 808         pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_PROPLIST_CHANGED], s);
 809     }
 810 }
 811
 812 /* Called from main thread */
 813 unsigned pa_source_linked_by(pa_source *s) {
 814     pa_source_assert_ref(s);
 815     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 816
 817     return pa_idxset_size(s->outputs);
 818 }
 819
 820 /* Called from main thread */
 821 unsigned pa_source_used_by(pa_source *s) {
 822     unsigned ret;
 823
 824     pa_source_assert_ref(s);
 825     pa_assert(PA_SOURCE_IS_LINKED(s->state));
 826
 827     ret = pa_idxset_size(s->outputs);
 828     pa_assert(ret >= s->n_corked);
 829
 830     return ret - s->n_corked;
 831 }
 832
 833 /* Called from main thread */
 834 unsigned pa_source_check_suspend(pa_source *s) {
 835     unsigned ret;
 836     pa_source_output *o;
 837     uint32_t idx;
 838
 839     pa_source_assert_ref(s);
 840
 841     if (!PA_SOURCE_IS_LINKED(s->state))
 842         return 0;
 843
 844     ret = 0;
 845
 846     for (o = PA_SOURCE_OUTPUT(pa_idxset_first(s->outputs, &idx)); o; o = PA_SOURCE_OUTPUT(pa_idxset_next(s->outputs, &idx))) {
 847         pa_source_output_state_t st;
 848
 849         st = pa_source_output_get_state(o);
 850         pa_assert(PA_SOURCE_OUTPUT_IS_LINKED(st));
 851
 852         if (st == PA_SOURCE_OUTPUT_CORKED)
 853             continue;
 854
 855         if (o->flags & PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND)
 856             continue;
 857
 858         ret ++;
 859     }
 860
 861     return ret;
 862 }
 863
 864 /* Called from IO thread, except when it is not */
 865 int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
 866     pa_source *s = PA_SOURCE(object);
 867     pa_source_assert_ref(s);
 868
 869     switch ((pa_source_message_t) code) {
 870
 871         case PA_SOURCE_MESSAGE_ADD_OUTPUT: {
 872             pa_source_output *o = PA_SOURCE_OUTPUT(userdata);
 873
 874             pa_hashmap_put(s->thread_info.outputs, PA_UINT32_TO_PTR(o->index), pa_source_output_ref(o));
 875
 876             if (o->direct_on_input) {
 877                 o->thread_info.direct_on_input = o->direct_on_input;
 878                 pa_hashmap_put(o->thread_info.direct_on_input->thread_info.direct_outputs, PA_UINT32_TO_PTR(o->index), o);
 879             }
 880
 881             pa_assert(!o->thread_info.attached);
 882             o->thread_info.attached = TRUE;
 883
 884             if (o->attach)
 885                 o->attach(o);
 886
 887             pa_source_output_set_state_within_thread(o, o->state);
 888
 889             if (o->thread_info.requested_source_latency != (pa_usec_t) -1)
 890                 pa_source_output_set_requested_latency_within_thread(o, o->thread_info.requested_source_latency);
 891
 892             pa_source_output_update_max_rewind(o, s->thread_info.max_rewind);
 893
 894             /* We don't just invalidate the requested latency here,
 895              * because if we are in a move we might need to fix up the
 896              * requested latency. */
 897             pa_source_output_set_requested_latency_within_thread(o, o->thread_info.requested_source_latency);
 898
 899             return 0;
 900         }
 901
 902         case PA_SOURCE_MESSAGE_REMOVE_OUTPUT: {
 903             pa_source_output *o = PA_SOURCE_OUTPUT(userdata);
 904
 905             pa_source_output_set_state_within_thread(o, o->state);
 906
 907             if (o->detach)
 908                 o->detach(o);
 909
 910             pa_assert(o->thread_info.attached);
 911             o->thread_info.attached = FALSE;
 912
 913             if (o->thread_info.direct_on_input) {
 914                 pa_hashmap_remove(o->thread_info.direct_on_input->thread_info.direct_outputs, PA_UINT32_TO_PTR(o->index));
 915                 o->thread_info.direct_on_input = NULL;
 916             }
 917
 918             if (pa_hashmap_remove(s->thread_info.outputs, PA_UINT32_TO_PTR(o->index)))
 919                 pa_source_output_unref(o);
 920
 921             pa_source_invalidate_requested_latency(s);
 922
 923             return 0;
 924         }
 925
 926         case PA_SOURCE_MESSAGE_SET_VOLUME:
 927             s->thread_info.soft_volume = s->soft_volume;
 928             return 0;
 929
 930         case PA_SOURCE_MESSAGE_GET_VOLUME:
 931             return 0;
 932
 933         case PA_SOURCE_MESSAGE_SET_MUTE:
 934             s->thread_info.soft_muted = s->muted;
 935             return 0;
 936
 937         case PA_SOURCE_MESSAGE_GET_MUTE:
 938             return 0;
 939
 940         case PA_SOURCE_MESSAGE_SET_STATE: {
 941
 942             pa_bool_t suspend_change =
 943                 (s->thread_info.state == PA_SOURCE_SUSPENDED && PA_SOURCE_IS_OPENED(PA_PTR_TO_UINT(userdata))) ||
 944                 (PA_SOURCE_IS_OPENED(s->thread_info.state) && PA_PTR_TO_UINT(userdata) == PA_SOURCE_SUSPENDED);
 945
 946             s->thread_info.state = PA_PTR_TO_UINT(userdata);
 947
 948             if (suspend_change) {
 949                 pa_source_output *o;
 950                 void *state = NULL;
 951
 952                 while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
 953                     if (o->suspend_within_thread)
 954                         o->suspend_within_thread(o, s->thread_info.state == PA_SOURCE_SUSPENDED);
 955             }
 956
 957
 958             return 0;
 959         }
 960
 961         case PA_SOURCE_MESSAGE_DETACH:
 962
 963             /* Detach all streams */
 964             pa_source_detach_within_thread(s);
 965             return 0;
 966
 967         case PA_SOURCE_MESSAGE_ATTACH:
 968
 969             /* Reattach all streams */
 970             pa_source_attach_within_thread(s);
 971             return 0;
 972
 973         case PA_SOURCE_MESSAGE_GET_REQUESTED_LATENCY: {
 974
 975             pa_usec_t *usec = userdata;
 976             *usec = pa_source_get_requested_latency_within_thread(s);
 977
 978             if (*usec == (pa_usec_t) -1)
 979                 *usec = s->thread_info.max_latency;
 980
 981             return 0;
 982         }
 983
 984         case PA_SOURCE_MESSAGE_SET_LATENCY_RANGE: {
 985             pa_usec_t *r = userdata;
 986
 987             pa_source_set_latency_range_within_thread(s, r[0], r[1]);
 988
 989             return 0;
 990         }
 991
 992         case PA_SOURCE_MESSAGE_GET_LATENCY_RANGE: {
 993             pa_usec_t *r = userdata;
 994
 995             r[0] = s->thread_info.min_latency;
 996             r[1] = s->thread_info.max_latency;
 997
 998             return 0;
 999         }
1000
1001         case PA_SOURCE_MESSAGE_GET_MAX_REWIND:
1002
1003             *((size_t*) userdata) = s->thread_info.max_rewind;
1004             return 0;
1005
1006         case PA_SOURCE_MESSAGE_SET_MAX_REWIND:
1007
1008             pa_source_set_max_rewind_within_thread(s, (size_t) offset);
1009             return 0;
1010
1011         case PA_SOURCE_MESSAGE_GET_LATENCY:
1012
1013             if (s->monitor_of) {
1014                 *((pa_usec_t*) userdata) = 0;
1015                 return 0;
1016             }
1017
1018             /* Implementors need to overwrite this implementation! */
1019             return -1;
1020
1021         case PA_SOURCE_MESSAGE_MAX:
1022             ;
1023     }
1024
1025     return -1;
1026 }
1027
1028 /* Called from main thread */
1029 int pa_source_suspend_all(pa_core *c, pa_bool_t suspend) {
1030     uint32_t idx;
1031     pa_source *source;
1032     int ret = 0;
1033
1034     pa_core_assert_ref(c);
1035
1036     for (source = PA_SOURCE(pa_idxset_first(c->sources, &idx)); source; source = PA_SOURCE(pa_idxset_next(c->sources, &idx))) {
1037         int r;
1038
1039         if (source->monitor_of)
1040             continue;
1041
1042         if ((r = pa_source_suspend(source, suspend)) < 0)
1043             ret = r;
1044     }
1045
1046     return ret;
1047 }
1048
1049 /* Called from main thread */
1050 void pa_source_detach(pa_source *s) {
1051     pa_source_assert_ref(s);
1052     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1053
1054     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_DETACH, NULL, 0, NULL) == 0);
1055 }
1056
1057 /* Called from main thread */
1058 void pa_source_attach(pa_source *s) {
1059     pa_source_assert_ref(s);
1060     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1061
1062     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_ATTACH, NULL, 0, NULL) == 0);
1063 }
1064
1065 /* Called from IO thread */
1066 void pa_source_detach_within_thread(pa_source *s) {
1067     pa_source_output *o;
1068     void *state = NULL;
1069
1070     pa_source_assert_ref(s);
1071     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
1072
1073     while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1074         if (o->detach)
1075             o->detach(o);
1076 }
1077
1078 /* Called from IO thread */
1079 void pa_source_attach_within_thread(pa_source *s) {
1080     pa_source_output *o;
1081     void *state = NULL;
1082
1083     pa_source_assert_ref(s);
1084     pa_assert(PA_SOURCE_IS_LINKED(s->thread_info.state));
1085
1086     while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1087         if (o->attach)
1088             o->attach(o);
1089 }
1090
1091 /* Called from IO thread */
1092 pa_usec_t pa_source_get_requested_latency_within_thread(pa_source *s) {
1093     pa_usec_t result = (pa_usec_t) -1;
1094     pa_source_output *o;
1095     void *state = NULL;
1096
1097     pa_source_assert_ref(s);
1098
1099     if (s->thread_info.requested_latency_valid)
1100         return s->thread_info.requested_latency;
1101
1102     while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1103
1104         if (o->thread_info.requested_source_latency != (pa_usec_t) -1 &&
1105             (result == (pa_usec_t) -1 || result > o->thread_info.requested_source_latency))
1106             result = o->thread_info.requested_source_latency;
1107
1108     if (result != (pa_usec_t) -1) {
1109         if (s->thread_info.max_latency > 0 && result > s->thread_info.max_latency)
1110             result = s->thread_info.max_latency;
1111
1112         if (s->thread_info.min_latency > 0 && result < s->thread_info.min_latency)
1113             result = s->thread_info.min_latency;
1114     }
1115
1116     s->thread_info.requested_latency = result;
1117     s->thread_info.requested_latency_valid = TRUE;
1118
1119     return result;
1120 }
1121
1122 /* Called from main thread */
1123 pa_usec_t pa_source_get_requested_latency(pa_source *s) {
1124     pa_usec_t usec;
1125
1126     pa_source_assert_ref(s);
1127     pa_assert(PA_SOURCE_IS_LINKED(s->state));
1128
1129     if (s->state == PA_SOURCE_SUSPENDED)
1130         return 0;
1131
1132     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
1133
1134     return usec;
1135 }
1136
1137 /* Called from IO thread */
1138 void pa_source_set_max_rewind_within_thread(pa_source *s, size_t max_rewind) {
1139     pa_source_output *o;
1140     void *state = NULL;
1141
1142     pa_source_assert_ref(s);
1143
1144     if (max_rewind == s->thread_info.max_rewind)
1145         return;
1146
1147     s->thread_info.max_rewind = max_rewind;
1148
1149     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1150         while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1151             pa_source_output_update_max_rewind(o, s->thread_info.max_rewind);
1152     }
1153 }
1154
1155 /* Called from main thread */
1156 void pa_source_set_max_rewind(pa_source *s, size_t max_rewind) {
1157     pa_source_assert_ref(s);
1158
1159     if (PA_SOURCE_IS_LINKED(s->state))
1160         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MAX_REWIND, NULL, max_rewind, NULL) == 0);
1161     else
1162         pa_source_set_max_rewind_within_thread(s, max_rewind);
1163 }
1164
1165 /* Called from IO thread */
1166 void pa_source_invalidate_requested_latency(pa_source *s) {
1167     pa_source_output *o;
1168     void *state = NULL;
1169
1170     pa_source_assert_ref(s);
1171
1172     s->thread_info.requested_latency_valid = FALSE;
1173
1174     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1175
1176         if (s->update_requested_latency)
1177             s->update_requested_latency(s);
1178
1179         while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1180             if (o->update_source_requested_latency)
1181                 o->update_source_requested_latency(o);
1182     }
1183
1184     if (s->monitor_of)
1185         pa_sink_invalidate_requested_latency(s->monitor_of);
1186 }
1187
1188 /* Called from main thread */
1189 void pa_source_set_latency_range(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency) {
1190     pa_source_assert_ref(s);
1191
1192     /* min_latency == 0:           no limit
1193      * min_latency anything else:  specified limit
1194      *
1195      * Similar for max_latency */
1196
1197     if (min_latency < ABSOLUTE_MIN_LATENCY)
1198         min_latency = ABSOLUTE_MIN_LATENCY;
1199
1200     if (max_latency <= 0 ||
1201         max_latency > ABSOLUTE_MAX_LATENCY)
1202         max_latency = ABSOLUTE_MAX_LATENCY;
1203
1204     pa_assert(min_latency <= max_latency);
1205
1206     /* Hmm, let's see if someone forgot to set PA_SOURCE_DYNAMIC_LATENCY here... */
1207     pa_assert((min_latency == ABSOLUTE_MIN_LATENCY &&
1208                max_latency == ABSOLUTE_MAX_LATENCY) ||
1209               (s->flags & PA_SOURCE_DYNAMIC_LATENCY));
1210
1211     if (PA_SOURCE_IS_LINKED(s->state)) {
1212         pa_usec_t r[2];
1213
1214         r[0] = min_latency;
1215         r[1] = max_latency;
1216
1217         pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_LATENCY_RANGE, r, 0, NULL) == 0);
1218     } else
1219         pa_source_set_latency_range_within_thread(s, min_latency, max_latency);
1220 }
1221
1222 /* Called from main thread */
1223 void pa_source_get_latency_range(pa_source *s, pa_usec_t *min_latency, pa_usec_t *max_latency) {
1224    pa_source_assert_ref(s);
1225    pa_assert(min_latency);
1226    pa_assert(max_latency);
1227
1228    if (PA_SOURCE_IS_LINKED(s->state)) {
1229        pa_usec_t r[2] = { 0, 0 };
1230
1231        pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY_RANGE, r, 0, NULL) == 0);
1232
1233        *min_latency = r[0];
1234        *max_latency = r[1];
1235    } else {
1236        *min_latency = s->thread_info.min_latency;
1237        *max_latency = s->thread_info.max_latency;
1238    }
1239 }
1240
1241 /* Called from IO thread, and from main thread before pa_source_put() is called */
1242 void pa_source_set_latency_range_within_thread(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency) {
1243     void *state = NULL;
1244
1245     pa_source_assert_ref(s);
1246
1247     pa_assert(min_latency >= ABSOLUTE_MIN_LATENCY);
1248     pa_assert(max_latency <= ABSOLUTE_MAX_LATENCY);
1249     pa_assert(min_latency <= max_latency);
1250
1251     /* Hmm, let's see if someone forgot to set PA_SOURCE_DYNAMIC_LATENCY here... */
1252     pa_assert((min_latency == ABSOLUTE_MIN_LATENCY &&
1253                max_latency == ABSOLUTE_MAX_LATENCY) ||
1254               (s->flags & PA_SOURCE_DYNAMIC_LATENCY) ||
1255               s->monitor_of);
1256
1257     s->thread_info.min_latency = min_latency;
1258     s->thread_info.max_latency = max_latency;
1259
1260     if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
1261         pa_source_output *o;
1262
1263         while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
1264             if (o->update_source_latency_range)
1265                 o->update_source_latency_range(o);
1266     }
1267
1268     pa_source_invalidate_requested_latency(s);
1269 }
1270
1271 /* Called from main thread */
1272 size_t pa_source_get_max_rewind(pa_source *s) {
1273     size_t r;
1274     pa_source_assert_ref(s);
1275
1276     if (!PA_SOURCE_IS_LINKED(s->state))
1277         return s->thread_info.max_rewind;
1278
1279     pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MAX_REWIND, &r, 0, NULL) == 0);
1280
1281     return r;
1282 }