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