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