src/pulsecore/source.c

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