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