code.delx.au - pulseaudio/blob - src/pulse/stream.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 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 <string.h>
  28 #include <stdio.h>
  29 #include <string.h>
  30
  31 #include <pulse/def.h>
  32 #include <pulse/timeval.h>
  33 #include <pulse/xmalloc.h>
  34
  35 #include <pulsecore/pstream-util.h>
  36 #include <pulsecore/log.h>
  37 #include <pulsecore/hashmap.h>
  38 #include <pulsecore/macro.h>
  39 #include <pulsecore/rtclock.h>
  40
  41 #include "internal.h"
  42
  43 #define LATENCY_IPOL_INTERVAL_USEC (333*PA_USEC_PER_MSEC)
  44
  45 #define SMOOTHER_ADJUST_TIME (1000*PA_USEC_PER_MSEC)
  46 #define SMOOTHER_HISTORY_TIME (5000*PA_USEC_PER_MSEC)
  47 #define SMOOTHER_MIN_HISTORY (4)
  48
  49 pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) {
  50     return pa_stream_new_with_proplist(c, name, ss, map, NULL);
  51 }
  52
  53 static void reset_callbacks(pa_stream *s) {
  54     s->read_callback = NULL;
  55     s->read_userdata = NULL;
  56     s->write_callback = NULL;
  57     s->write_userdata = NULL;
  58     s->state_callback = NULL;
  59     s->state_userdata = NULL;
  60     s->overflow_callback = NULL;
  61     s->overflow_userdata = NULL;
  62     s->underflow_callback = NULL;
  63     s->underflow_userdata = NULL;
  64     s->latency_update_callback = NULL;
  65     s->latency_update_userdata = NULL;
  66     s->moved_callback = NULL;
  67     s->moved_userdata = NULL;
  68     s->suspended_callback = NULL;
  69     s->suspended_userdata = NULL;
  70     s->started_callback = NULL;
  71     s->started_userdata = NULL;
  72 }
  73
  74 pa_stream *pa_stream_new_with_proplist(
  75         pa_context *c,
  76         const char *name,
  77         const pa_sample_spec *ss,
  78         const pa_channel_map *map,
  79         pa_proplist *p) {
  80
  81     pa_stream *s;
  82     int i;
  83     pa_channel_map tmap;
  84
  85     pa_assert(c);
  86     pa_assert(PA_REFCNT_VALUE(c) >= 1);
  87
  88     PA_CHECK_VALIDITY_RETURN_NULL(c, ss && pa_sample_spec_valid(ss), PA_ERR_INVALID);
  89     PA_CHECK_VALIDITY_RETURN_NULL(c, c->version >= 12 || (ss->format != PA_SAMPLE_S32LE || ss->format != PA_SAMPLE_S32NE), PA_ERR_NOTSUPPORTED);
  90     PA_CHECK_VALIDITY_RETURN_NULL(c, !map || (pa_channel_map_valid(map) && map->channels == ss->channels), PA_ERR_INVALID);
  91     PA_CHECK_VALIDITY_RETURN_NULL(c, name || (p && pa_proplist_contains(p, PA_PROP_MEDIA_NAME)), PA_ERR_INVALID);
  92
  93     if (!map)
  94         PA_CHECK_VALIDITY_RETURN_NULL(c, map = pa_channel_map_init_auto(&tmap, ss->channels, PA_CHANNEL_MAP_DEFAULT), PA_ERR_INVALID);
  95
  96     s = pa_xnew(pa_stream, 1);
  97     PA_REFCNT_INIT(s);
  98     s->context = c;
  99     s->mainloop = c->mainloop;
 100
 101     s->direction = PA_STREAM_NODIRECTION;
 102     s->state = PA_STREAM_UNCONNECTED;
 103     s->flags = 0;
 104
 105     s->sample_spec = *ss;
 106     s->channel_map = *map;
 107
 108     s->direct_on_input = PA_INVALID_INDEX;
 109
 110     s->proplist = p ? pa_proplist_copy(p) : pa_proplist_new();
 111     if (name)
 112         pa_proplist_sets(s->proplist, PA_PROP_MEDIA_NAME, name);
 113
 114     s->channel = 0;
 115     s->channel_valid = FALSE;
 116     s->syncid = c->csyncid++;
 117     s->stream_index = PA_INVALID_INDEX;
 118
 119     s->requested_bytes = 0;
 120     memset(&s->buffer_attr, 0, sizeof(s->buffer_attr));
 121
 122     /* We initialize der target length here, so that if the user
 123      * passes no explicit buffering metrics the default is similar to
 124      * what older PA versions provided. */
 125     s->buffer_attr.tlength = pa_usec_to_bytes(250*PA_USEC_PER_MSEC, ss); /* 250ms of buffering */
 126
 127     s->device_index = PA_INVALID_INDEX;
 128     s->device_name = NULL;
 129     s->suspended = FALSE;
 130
 131     pa_memchunk_reset(&s->peek_memchunk);
 132     s->peek_data = NULL;
 133
 134     s->record_memblockq = NULL;
 135
 136     s->corked = FALSE;
 137
 138     memset(&s->timing_info, 0, sizeof(s->timing_info));
 139     s->timing_info_valid = FALSE;
 140
 141     s->previous_time = 0;
 142
 143     s->read_index_not_before = 0;
 144     s->write_index_not_before = 0;
 145     for (i = 0; i < PA_MAX_WRITE_INDEX_CORRECTIONS; i++)
 146         s->write_index_corrections[i].valid = 0;
 147     s->current_write_index_correction = 0;
 148
 149     s->auto_timing_update_event = NULL;
 150     s->auto_timing_update_requested = FALSE;
 151
 152     reset_callbacks(s);
 153
 154     s->smoother = NULL;
 155
 156     /* Refcounting is strictly one-way: from the "bigger" to the "smaller" object. */
 157     PA_LLIST_PREPEND(pa_stream, c->streams, s);
 158     pa_stream_ref(s);
 159
 160     return s;
 161 }
 162
 163 static void stream_unlink(pa_stream *s) {
 164     pa_operation *o, *n;
 165     pa_assert(s);
 166
 167     if (!s->context)
 168         return;
 169
 170     /* Detach from context */
 171
 172     /* Unref all operatio object that point to us */
 173     for (o = s->context->operations; o; o = n) {
 174         n = o->next;
 175
 176         if (o->stream == s)
 177             pa_operation_cancel(o);
 178     }
 179
 180     /* Drop all outstanding replies for this stream */
 181     if (s->context->pdispatch)
 182         pa_pdispatch_unregister_reply(s->context->pdispatch, s);
 183
 184     if (s->channel_valid) {
 185         pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL);
 186         s->channel = 0;
 187         s->channel_valid = FALSE;
 188     }
 189
 190     PA_LLIST_REMOVE(pa_stream, s->context->streams, s);
 191     pa_stream_unref(s);
 192
 193     s->context = NULL;
 194
 195     if (s->auto_timing_update_event) {
 196         pa_assert(s->mainloop);
 197         s->mainloop->time_free(s->auto_timing_update_event);
 198     }
 199
 200     reset_callbacks(s);
 201 }
 202
 203 static void stream_free(pa_stream *s) {
 204     pa_assert(s);
 205
 206     stream_unlink(s);
 207
 208     if (s->peek_memchunk.memblock) {
 209         if (s->peek_data)
 210             pa_memblock_release(s->peek_memchunk.memblock);
 211         pa_memblock_unref(s->peek_memchunk.memblock);
 212     }
 213
 214     if (s->record_memblockq)
 215         pa_memblockq_free(s->record_memblockq);
 216
 217     if (s->proplist)
 218         pa_proplist_free(s->proplist);
 219
 220     if (s->smoother)
 221         pa_smoother_free(s->smoother);
 222
 223     pa_xfree(s->device_name);
 224     pa_xfree(s);
 225 }
 226
 227 void pa_stream_unref(pa_stream *s) {
 228     pa_assert(s);
 229     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 230
 231     if (PA_REFCNT_DEC(s) <= 0)
 232         stream_free(s);
 233 }
 234
 235 pa_stream* pa_stream_ref(pa_stream *s) {
 236     pa_assert(s);
 237     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 238
 239     PA_REFCNT_INC(s);
 240     return s;
 241 }
 242
 243 pa_stream_state_t pa_stream_get_state(pa_stream *s) {
 244     pa_assert(s);
 245     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 246
 247     return s->state;
 248 }
 249
 250 pa_context* pa_stream_get_context(pa_stream *s) {
 251     pa_assert(s);
 252     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 253
 254     return s->context;
 255 }
 256
 257 uint32_t pa_stream_get_index(pa_stream *s) {
 258     pa_assert(s);
 259     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 260
 261     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
 262
 263     return s->stream_index;
 264 }
 265
 266 void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
 267     pa_assert(s);
 268     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 269
 270     if (s->state == st)
 271         return;
 272
 273     pa_stream_ref(s);
 274
 275     s->state = st;
 276
 277     if (s->state_callback)
 278         s->state_callback(s, s->state_userdata);
 279
 280     if ((st == PA_STREAM_FAILED || st == PA_STREAM_TERMINATED))
 281         stream_unlink(s);
 282
 283     pa_stream_unref(s);
 284 }
 285
 286 static void request_auto_timing_update(pa_stream *s, pa_bool_t force) {
 287     pa_assert(s);
 288     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 289
 290     if (!(s->flags & PA_STREAM_AUTO_TIMING_UPDATE))
 291         return;
 292
 293     if (s->state == PA_STREAM_READY &&
 294         (force || !s->auto_timing_update_requested)) {
 295         pa_operation *o;
 296
 297 /*         pa_log("automatically requesting new timing data"); */
 298
 299         if ((o = pa_stream_update_timing_info(s, NULL, NULL))) {
 300             pa_operation_unref(o);
 301             s->auto_timing_update_requested = TRUE;
 302         }
 303     }
 304
 305     if (s->auto_timing_update_event) {
 306         struct timeval next;
 307         pa_gettimeofday(&next);
 308         pa_timeval_add(&next, LATENCY_IPOL_INTERVAL_USEC);
 309         s->mainloop->time_restart(s->auto_timing_update_event, &next);
 310     }
 311 }
 312
 313 void pa_command_stream_killed(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 314     pa_context *c = userdata;
 315     pa_stream *s;
 316     uint32_t channel;
 317
 318     pa_assert(pd);
 319     pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED);
 320     pa_assert(t);
 321     pa_assert(c);
 322     pa_assert(PA_REFCNT_VALUE(c) >= 1);
 323
 324     pa_context_ref(c);
 325
 326     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 327         !pa_tagstruct_eof(t)) {
 328         pa_context_fail(c, PA_ERR_PROTOCOL);
 329         goto finish;
 330     }
 331
 332     if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, channel)))
 333         goto finish;
 334
 335     if (s->state != PA_STREAM_READY)
 336         goto finish;
 337
 338     pa_context_set_error(c, PA_ERR_KILLED);
 339     pa_stream_set_state(s, PA_STREAM_FAILED);
 340
 341 finish:
 342     pa_context_unref(c);
 343 }
 344
 345 static void check_smoother_status(pa_stream *s, pa_bool_t aposteriori, pa_bool_t force_start, pa_bool_t force_stop) {
 346     pa_usec_t x;
 347
 348     pa_assert(s);
 349     pa_assert(!force_start || !force_stop);
 350
 351     if (!s->smoother)
 352         return;
 353
 354     x = pa_rtclock_usec();
 355
 356     if (s->timing_info_valid) {
 357         if (aposteriori)
 358             x -= s->timing_info.transport_usec;
 359         else
 360             x += s->timing_info.transport_usec;
 361
 362         if (s->direction == PA_STREAM_PLAYBACK)
 363             /* it takes a while until the pause/resume is actually
 364              * audible */
 365             x += s->timing_info.sink_usec;
 366         else
 367             /* Data froma  while back will be dropped */
 368             x -= s->timing_info.source_usec;
 369     }
 370
 371     if (s->suspended || s->corked || force_stop)
 372         pa_smoother_pause(s->smoother, x);
 373     else if (force_start || s->buffer_attr.prebuf == 0)
 374         pa_smoother_resume(s->smoother, x);
 375
 376     /* Please note that we have no idea if playback actually started
 377      * if prebuf is non-zero! */
 378 }
 379
 380 void pa_command_stream_moved(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 381     pa_context *c = userdata;
 382     pa_stream *s;
 383     uint32_t channel;
 384     const char *dn;
 385     pa_bool_t suspended;
 386     uint32_t di;
 387     pa_usec_t usec;
 388     uint32_t maxlength = 0, fragsize = 0, minreq = 0, tlength = 0, prebuf = 0;
 389
 390     pa_assert(pd);
 391     pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_MOVED || command == PA_COMMAND_RECORD_STREAM_MOVED);
 392     pa_assert(t);
 393     pa_assert(c);
 394     pa_assert(PA_REFCNT_VALUE(c) >= 1);
 395
 396     pa_context_ref(c);
 397
 398     if (c->version < 12) {
 399         pa_context_fail(c, PA_ERR_PROTOCOL);
 400         goto finish;
 401     }
 402
 403     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 404         pa_tagstruct_getu32(t, &di) < 0 ||
 405         pa_tagstruct_gets(t, &dn) < 0 ||
 406         pa_tagstruct_get_boolean(t, &suspended) < 0) {
 407         pa_context_fail(c, PA_ERR_PROTOCOL);
 408         goto finish;
 409     }
 410
 411     if (c->version >= 13) {
 412
 413         if (command == PA_COMMAND_RECORD_STREAM_MOVED) {
 414             if (pa_tagstruct_getu32(t, &maxlength) < 0 ||
 415                 pa_tagstruct_getu32(t, &fragsize) < 0 ||
 416                 pa_tagstruct_get_usec(t, &usec) < 0) {
 417                 pa_context_fail(c, PA_ERR_PROTOCOL);
 418                 goto finish;
 419             }
 420         } else {
 421             if (pa_tagstruct_getu32(t, &maxlength) < 0 ||
 422                 pa_tagstruct_getu32(t, &tlength) < 0 ||
 423                 pa_tagstruct_getu32(t, &prebuf) < 0 ||
 424                 pa_tagstruct_getu32(t, &minreq) < 0 ||
 425                 pa_tagstruct_get_usec(t, &usec) < 0) {
 426                 pa_context_fail(c, PA_ERR_PROTOCOL);
 427                 goto finish;
 428             }
 429         }
 430     }
 431
 432     if (!pa_tagstruct_eof(t)) {
 433         pa_context_fail(c, PA_ERR_PROTOCOL);
 434         goto finish;
 435     }
 436
 437     if (!dn || di == PA_INVALID_INDEX) {
 438         pa_context_fail(c, PA_ERR_PROTOCOL);
 439         goto finish;
 440     }
 441
 442     if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_MOVED ? c->playback_streams : c->record_streams, channel)))
 443         goto finish;
 444
 445     if (s->state != PA_STREAM_READY)
 446         goto finish;
 447
 448     if (c->version >= 13) {
 449         if (s->direction == PA_STREAM_RECORD)
 450             s->timing_info.configured_source_usec = usec;
 451         else
 452             s->timing_info.configured_sink_usec = usec;
 453
 454         s->buffer_attr.maxlength = maxlength;
 455         s->buffer_attr.fragsize = fragsize;
 456         s->buffer_attr.tlength = tlength;
 457         s->buffer_attr.prebuf = prebuf;
 458         s->buffer_attr.minreq = minreq;
 459     }
 460
 461     pa_xfree(s->device_name);
 462     s->device_name = pa_xstrdup(dn);
 463     s->device_index = di;
 464
 465     s->suspended = suspended;
 466
 467     check_smoother_status(s, TRUE, FALSE, FALSE);
 468     request_auto_timing_update(s, TRUE);
 469
 470     if (s->moved_callback)
 471         s->moved_callback(s, s->moved_userdata);
 472
 473 finish:
 474     pa_context_unref(c);
 475 }
 476
 477 void pa_command_stream_suspended(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 478     pa_context *c = userdata;
 479     pa_stream *s;
 480     uint32_t channel;
 481     pa_bool_t suspended;
 482
 483     pa_assert(pd);
 484     pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_SUSPENDED || command == PA_COMMAND_RECORD_STREAM_SUSPENDED);
 485     pa_assert(t);
 486     pa_assert(c);
 487     pa_assert(PA_REFCNT_VALUE(c) >= 1);
 488
 489     pa_context_ref(c);
 490
 491     if (c->version < 12) {
 492         pa_context_fail(c, PA_ERR_PROTOCOL);
 493         goto finish;
 494     }
 495
 496     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 497         pa_tagstruct_get_boolean(t, &suspended) < 0 ||
 498         !pa_tagstruct_eof(t)) {
 499         pa_context_fail(c, PA_ERR_PROTOCOL);
 500         goto finish;
 501     }
 502
 503     if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_SUSPENDED ? c->playback_streams : c->record_streams, channel)))
 504         goto finish;
 505
 506     if (s->state != PA_STREAM_READY)
 507         goto finish;
 508
 509     s->suspended = suspended;
 510
 511     check_smoother_status(s, TRUE, FALSE, FALSE);
 512     request_auto_timing_update(s, TRUE);
 513
 514     if (s->suspended_callback)
 515         s->suspended_callback(s, s->suspended_userdata);
 516
 517 finish:
 518     pa_context_unref(c);
 519 }
 520
 521 void pa_command_stream_started(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
 522     pa_context *c = userdata;
 523     pa_stream *s;
 524     uint32_t channel;
 525
 526     pa_assert(pd);
 527     pa_assert(command == PA_COMMAND_STARTED);
 528     pa_assert(t);
 529     pa_assert(c);
 530     pa_assert(PA_REFCNT_VALUE(c) >= 1);
 531
 532     pa_context_ref(c);
 533
 534     if (c->version < 13) {
 535         pa_context_fail(c, PA_ERR_PROTOCOL);
 536         goto finish;
 537     }
 538
 539     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 540         !pa_tagstruct_eof(t)) {
 541         pa_context_fail(c, PA_ERR_PROTOCOL);
 542         goto finish;
 543     }
 544
 545     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
 546         goto finish;
 547
 548     if (s->state != PA_STREAM_READY)
 549         goto finish;
 550
 551     check_smoother_status(s, TRUE, TRUE, FALSE);
 552     request_auto_timing_update(s, TRUE);
 553
 554     if (s->started_callback)
 555         s->started_callback(s, s->suspended_userdata);
 556
 557 finish:
 558     pa_context_unref(c);
 559 }
 560
 561 void pa_command_request(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 562     pa_stream *s;
 563     pa_context *c = userdata;
 564     uint32_t bytes, channel;
 565
 566     pa_assert(pd);
 567     pa_assert(command == PA_COMMAND_REQUEST);
 568     pa_assert(t);
 569     pa_assert(c);
 570     pa_assert(PA_REFCNT_VALUE(c) >= 1);
 571
 572     pa_context_ref(c);
 573
 574     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 575         pa_tagstruct_getu32(t, &bytes) < 0 ||
 576         !pa_tagstruct_eof(t)) {
 577         pa_context_fail(c, PA_ERR_PROTOCOL);
 578         goto finish;
 579     }
 580
 581     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
 582         goto finish;
 583
 584     if (s->state != PA_STREAM_READY)
 585         goto finish;
 586
 587     s->requested_bytes += bytes;
 588
 589     if (s->requested_bytes > 0 && s->write_callback)
 590         s->write_callback(s, s->requested_bytes, s->write_userdata);
 591
 592 finish:
 593     pa_context_unref(c);
 594 }
 595
 596 void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 597     pa_stream *s;
 598     pa_context *c = userdata;
 599     uint32_t channel;
 600
 601     pa_assert(pd);
 602     pa_assert(command == PA_COMMAND_OVERFLOW || command == PA_COMMAND_UNDERFLOW);
 603     pa_assert(t);
 604     pa_assert(c);
 605     pa_assert(PA_REFCNT_VALUE(c) >= 1);
 606
 607     pa_context_ref(c);
 608
 609     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 610         !pa_tagstruct_eof(t)) {
 611         pa_context_fail(c, PA_ERR_PROTOCOL);
 612         goto finish;
 613     }
 614
 615     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
 616         goto finish;
 617
 618     if (s->state != PA_STREAM_READY)
 619         goto finish;
 620
 621     if (s->buffer_attr.prebuf > 0)
 622         check_smoother_status(s, TRUE, FALSE, TRUE);
 623
 624     request_auto_timing_update(s, TRUE);
 625
 626     if (command == PA_COMMAND_OVERFLOW) {
 627         if (s->overflow_callback)
 628             s->overflow_callback(s, s->overflow_userdata);
 629     } else if (command == PA_COMMAND_UNDERFLOW) {
 630         if (s->underflow_callback)
 631             s->underflow_callback(s, s->underflow_userdata);
 632     }
 633
 634  finish:
 635     pa_context_unref(c);
 636 }
 637
 638 static void invalidate_indexes(pa_stream *s, pa_bool_t r, pa_bool_t w) {
 639     pa_assert(s);
 640     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 641
 642 /*     pa_log("invalidate r:%u w:%u tag:%u", r, w, s->context->ctag); */
 643
 644     if (s->state != PA_STREAM_READY)
 645         return;
 646
 647     if (w) {
 648         s->write_index_not_before = s->context->ctag;
 649
 650         if (s->timing_info_valid)
 651             s->timing_info.write_index_corrupt = TRUE;
 652
 653 /*         pa_log("write_index invalidated"); */
 654     }
 655
 656     if (r) {
 657         s->read_index_not_before = s->context->ctag;
 658
 659         if (s->timing_info_valid)
 660             s->timing_info.read_index_corrupt = TRUE;
 661
 662 /*         pa_log("read_index invalidated"); */
 663     }
 664
 665     request_auto_timing_update(s, TRUE);
 666 }
 667
 668 static void auto_timing_update_callback(PA_GCC_UNUSED pa_mainloop_api *m, PA_GCC_UNUSED pa_time_event *e, PA_GCC_UNUSED const struct timeval *tv, void *userdata) {
 669     pa_stream *s = userdata;
 670
 671     pa_assert(s);
 672     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 673
 674     pa_stream_ref(s);
 675     request_auto_timing_update(s, FALSE);
 676     pa_stream_unref(s);
 677 }
 678
 679 static void create_stream_complete(pa_stream *s) {
 680     pa_assert(s);
 681     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 682     pa_assert(s->state == PA_STREAM_CREATING);
 683
 684     pa_stream_set_state(s, PA_STREAM_READY);
 685
 686     if (s->requested_bytes > 0 && s->write_callback)
 687         s->write_callback(s, s->requested_bytes, s->write_userdata);
 688
 689     if (s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
 690         struct timeval tv;
 691         pa_gettimeofday(&tv);
 692         tv.tv_usec += LATENCY_IPOL_INTERVAL_USEC; /* every 100 ms */
 693         pa_assert(!s->auto_timing_update_event);
 694         s->auto_timing_update_event = s->mainloop->time_new(s->mainloop, &tv, &auto_timing_update_callback, s);
 695
 696         request_auto_timing_update(s, TRUE);
 697     }
 698
 699     check_smoother_status(s, TRUE, FALSE, FALSE);
 700 }
 701
 702 static void automatic_buffer_attr(pa_stream *s, pa_buffer_attr *attr, const pa_sample_spec *ss) {
 703     pa_assert(s);
 704     pa_assert(attr);
 705     pa_assert(ss);
 706
 707     if (s->context->version >= 13)
 708         return;
 709
 710     /* Version older than 0.9.10 didn't do server side buffer_attr
 711      * selection, hence we have to fake it on the client side. */
 712
 713     /* We choose fairly conservative values here, to not confuse
 714      * old clients with extremely large playback buffers */
 715
 716     if (!attr->maxlength <= 0)
 717         attr->maxlength = 4*1024*1024; /* 4MB is the maximum queue length PulseAudio <= 0.9.9 supported. */
 718
 719     if (!attr->tlength <= 0)
 720         attr->tlength = pa_usec_to_bytes(250*PA_USEC_PER_MSEC, ss); /* 250ms of buffering */
 721
 722     if (!attr->minreq <= 0)
 723         attr->minreq = (attr->tlength)/5; /* Ask for more data when there are only 200ms left in the playback buffer */
 724
 725     if (!attr->prebuf)
 726         attr->prebuf = attr->tlength; /* Start to play only when the playback is fully filled up once */
 727
 728     if (!attr->fragsize)
 729         attr->fragsize  = attr->tlength; /* Pass data to the app only when the buffer is filled up once */
 730 }
 731
 732 void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 733     pa_stream *s = userdata;
 734
 735     pa_assert(pd);
 736     pa_assert(s);
 737     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 738     pa_assert(s->state == PA_STREAM_CREATING);
 739
 740     pa_stream_ref(s);
 741
 742     if (command != PA_COMMAND_REPLY) {
 743         if (pa_context_handle_error(s->context, command, t, FALSE) < 0)
 744             goto finish;
 745
 746         pa_stream_set_state(s, PA_STREAM_FAILED);
 747         goto finish;
 748     }
 749
 750     if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
 751         s->channel == PA_INVALID_INDEX ||
 752         ((s->direction != PA_STREAM_UPLOAD) && (pa_tagstruct_getu32(t, &s->stream_index) < 0 ||  s->stream_index == PA_INVALID_INDEX)) ||
 753         ((s->direction != PA_STREAM_RECORD) && pa_tagstruct_getu32(t, &s->requested_bytes) < 0)) {
 754         pa_context_fail(s->context, PA_ERR_PROTOCOL);
 755         goto finish;
 756     }
 757
 758     if (s->context->version >= 9) {
 759         if (s->direction == PA_STREAM_PLAYBACK) {
 760             if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
 761                 pa_tagstruct_getu32(t, &s->buffer_attr.tlength) < 0 ||
 762                 pa_tagstruct_getu32(t, &s->buffer_attr.prebuf) < 0 ||
 763                 pa_tagstruct_getu32(t, &s->buffer_attr.minreq) < 0) {
 764                 pa_context_fail(s->context, PA_ERR_PROTOCOL);
 765                 goto finish;
 766             }
 767         } else if (s->direction == PA_STREAM_RECORD) {
 768             if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
 769                 pa_tagstruct_getu32(t, &s->buffer_attr.fragsize) < 0) {
 770                 pa_context_fail(s->context, PA_ERR_PROTOCOL);
 771                 goto finish;
 772             }
 773         }
 774     }
 775
 776     if (s->context->version >= 12 && s->direction != PA_STREAM_UPLOAD) {
 777         pa_sample_spec ss;
 778         pa_channel_map cm;
 779         const char *dn = NULL;
 780         pa_bool_t suspended;
 781
 782         if (pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 783             pa_tagstruct_get_channel_map(t, &cm) < 0 ||
 784             pa_tagstruct_getu32(t, &s->device_index) < 0 ||
 785             pa_tagstruct_gets(t, &dn) < 0 ||
 786             pa_tagstruct_get_boolean(t, &suspended) < 0) {
 787             pa_context_fail(s->context, PA_ERR_PROTOCOL);
 788             goto finish;
 789         }
 790
 791         if (!dn || s->device_index == PA_INVALID_INDEX ||
 792             ss.channels != cm.channels ||
 793             !pa_channel_map_valid(&cm) ||
 794             !pa_sample_spec_valid(&ss) ||
 795             (!(s->flags & PA_STREAM_FIX_FORMAT) && ss.format != s->sample_spec.format) ||
 796             (!(s->flags & PA_STREAM_FIX_RATE) && ss.rate != s->sample_spec.rate) ||
 797             (!(s->flags & PA_STREAM_FIX_CHANNELS) && !pa_channel_map_equal(&cm, &s->channel_map))) {
 798             pa_context_fail(s->context, PA_ERR_PROTOCOL);
 799             goto finish;
 800         }
 801
 802         pa_xfree(s->device_name);
 803         s->device_name = pa_xstrdup(dn);
 804         s->suspended = suspended;
 805
 806         s->channel_map = cm;
 807         s->sample_spec = ss;
 808     }
 809
 810     if (s->context->version >= 13 && s->direction != PA_STREAM_UPLOAD) {
 811         pa_usec_t usec;
 812
 813         if (pa_tagstruct_get_usec(t, &usec) < 0) {
 814             pa_context_fail(s->context, PA_ERR_PROTOCOL);
 815             goto finish;
 816         }
 817
 818         if (s->direction == PA_STREAM_RECORD)
 819             s->timing_info.configured_source_usec = usec;
 820         else
 821             s->timing_info.configured_sink_usec = usec;
 822     }
 823
 824     if (!pa_tagstruct_eof(t)) {
 825         pa_context_fail(s->context, PA_ERR_PROTOCOL);
 826         goto finish;
 827     }
 828
 829     if (s->direction == PA_STREAM_RECORD) {
 830         pa_assert(!s->record_memblockq);
 831
 832         s->record_memblockq = pa_memblockq_new(
 833                 0,
 834                 s->buffer_attr.maxlength,
 835                 0,
 836                 pa_frame_size(&s->sample_spec),
 837                 1,
 838                 0,
 839                 0,
 840                 NULL);
 841     }
 842
 843     s->channel_valid = TRUE;
 844     pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
 845
 846     create_stream_complete(s);
 847
 848 finish:
 849     pa_stream_unref(s);
 850 }
 851
 852 static int create_stream(
 853         pa_stream_direction_t direction,
 854         pa_stream *s,
 855         const char *dev,
 856         const pa_buffer_attr *attr,
 857         pa_stream_flags_t flags,
 858         const pa_cvolume *volume,
 859         pa_stream *sync_stream) {
 860
 861     pa_tagstruct *t;
 862     uint32_t tag;
 863
 864     pa_assert(s);
 865     pa_assert(PA_REFCNT_VALUE(s) >= 1);
 866     pa_assert(direction == PA_STREAM_PLAYBACK || direction == PA_STREAM_RECORD);
 867
 868     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
 869     PA_CHECK_VALIDITY(s->context, s->direct_on_input == PA_INVALID_INDEX || direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
 870     PA_CHECK_VALIDITY(s->context, !(flags & ~(PA_STREAM_START_CORKED|
 871                                               PA_STREAM_INTERPOLATE_TIMING|
 872                                               PA_STREAM_NOT_MONOTONOUS|
 873                                               PA_STREAM_AUTO_TIMING_UPDATE|
 874                                               PA_STREAM_NO_REMAP_CHANNELS|
 875                                               PA_STREAM_NO_REMIX_CHANNELS|
 876                                               PA_STREAM_FIX_FORMAT|
 877                                               PA_STREAM_FIX_RATE|
 878                                               PA_STREAM_FIX_CHANNELS|
 879                                               PA_STREAM_DONT_MOVE|
 880                                               PA_STREAM_VARIABLE_RATE|
 881                                               PA_STREAM_PEAK_DETECT|
 882                                               PA_STREAM_START_MUTED|
 883                                               PA_STREAM_ADJUST_LATENCY)), PA_ERR_INVALID);
 884
 885     PA_CHECK_VALIDITY(s->context, s->context->version >= 12 || !(flags & PA_STREAM_VARIABLE_RATE), PA_ERR_NOTSUPPORTED);
 886     PA_CHECK_VALIDITY(s->context, s->context->version >= 13 || !(flags & PA_STREAM_PEAK_DETECT), PA_ERR_NOTSUPPORTED);
 887     /* Althought some of the other flags are not supported on older
 888      * version, we don't check for them here, because it doesn't hurt
 889      * when they are passed but actually not supported. This makes
 890      * client development easier */
 891
 892     PA_CHECK_VALIDITY(s->context, direction == PA_STREAM_PLAYBACK || !(flags & (PA_STREAM_START_MUTED)), PA_ERR_INVALID);
 893     PA_CHECK_VALIDITY(s->context, direction == PA_STREAM_RECORD || !(flags & (PA_STREAM_PEAK_DETECT)), PA_ERR_INVALID);
 894     PA_CHECK_VALIDITY(s->context, !volume || volume->channels == s->sample_spec.channels, PA_ERR_INVALID);
 895     PA_CHECK_VALIDITY(s->context, !sync_stream || (direction == PA_STREAM_PLAYBACK && sync_stream->direction == PA_STREAM_PLAYBACK), PA_ERR_INVALID);
 896
 897     pa_stream_ref(s);
 898
 899     s->direction = direction;
 900     s->flags = flags;
 901     s->corked = !!(flags & PA_STREAM_START_CORKED);
 902
 903     if (sync_stream)
 904         s->syncid = sync_stream->syncid;
 905
 906     if (attr)
 907         s->buffer_attr = *attr;
 908     automatic_buffer_attr(s, &s->buffer_attr, &s->sample_spec);
 909
 910     if (flags & PA_STREAM_INTERPOLATE_TIMING) {
 911         pa_usec_t x;
 912
 913         if (s->smoother)
 914             pa_smoother_free(s->smoother);
 915
 916         s->smoother = pa_smoother_new(SMOOTHER_ADJUST_TIME, SMOOTHER_HISTORY_TIME, !(flags & PA_STREAM_NOT_MONOTONIC), SMOOTHER_MIN_HISTORY);
 917
 918         x = pa_rtclock_usec();
 919         pa_smoother_set_time_offset(s->smoother, x);
 920         pa_smoother_pause(s->smoother, x);
 921     }
 922
 923     if (!dev)
 924         dev = s->direction == PA_STREAM_PLAYBACK ? s->context->conf->default_sink : s->context->conf->default_source;
 925
 926     t = pa_tagstruct_command(
 927             s->context,
 928             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM,
 929             &tag);
 930
 931     if (s->context->version < 13)
 932         pa_tagstruct_puts(t, pa_proplist_gets(s->proplist, PA_PROP_MEDIA_NAME));
 933
 934     pa_tagstruct_put(
 935             t,
 936             PA_TAG_SAMPLE_SPEC, &s->sample_spec,
 937             PA_TAG_CHANNEL_MAP, &s->channel_map,
 938             PA_TAG_U32, PA_INVALID_INDEX,
 939             PA_TAG_STRING, dev,
 940             PA_TAG_U32, s->buffer_attr.maxlength,
 941             PA_TAG_BOOLEAN, s->corked,
 942             PA_TAG_INVALID);
 943
 944     if (s->direction == PA_STREAM_PLAYBACK) {
 945         pa_cvolume cv;
 946
 947         pa_tagstruct_put(
 948                 t,
 949                 PA_TAG_U32, s->buffer_attr.tlength,
 950                 PA_TAG_U32, s->buffer_attr.prebuf,
 951                 PA_TAG_U32, s->buffer_attr.minreq,
 952                 PA_TAG_U32, s->syncid,
 953                 PA_TAG_INVALID);
 954
 955         if (!volume)
 956             volume = pa_cvolume_reset(&cv, s->sample_spec.channels);
 957
 958         pa_tagstruct_put_cvolume(t, volume);
 959     } else
 960         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
 961
 962     if (s->context->version >= 12) {
 963         pa_tagstruct_put(
 964                 t,
 965                 PA_TAG_BOOLEAN, flags & PA_STREAM_NO_REMAP_CHANNELS,
 966                 PA_TAG_BOOLEAN, flags & PA_STREAM_NO_REMIX_CHANNELS,
 967                 PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_FORMAT,
 968                 PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_RATE,
 969                 PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_CHANNELS,
 970                 PA_TAG_BOOLEAN, flags & PA_STREAM_DONT_MOVE,
 971                 PA_TAG_BOOLEAN, flags & PA_STREAM_VARIABLE_RATE,
 972                 PA_TAG_INVALID);
 973     }
 974
 975     if (s->context->version >= 13) {
 976
 977         if (s->direction == PA_STREAM_PLAYBACK)
 978             pa_tagstruct_put_boolean(t, flags & PA_STREAM_START_MUTED);
 979         else
 980             pa_tagstruct_put_boolean(t, flags & PA_STREAM_PEAK_DETECT);
 981
 982         pa_tagstruct_put(
 983                 t,
 984                 PA_TAG_BOOLEAN, flags & PA_STREAM_ADJUST_LATENCY,
 985                 PA_TAG_PROPLIST, s->proplist,
 986                 PA_TAG_INVALID);
 987
 988         if (s->direction == PA_STREAM_RECORD)
 989             pa_tagstruct_putu32(t, s->direct_on_input);
 990     }
 991
 992     pa_pstream_send_tagstruct(s->context->pstream, t);
 993     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s, NULL);
 994
 995     pa_stream_set_state(s, PA_STREAM_CREATING);
 996
 997     pa_stream_unref(s);
 998     return 0;
 999 }
1000
1001 int pa_stream_connect_playback(
1002         pa_stream *s,
1003         const char *dev,
1004         const pa_buffer_attr *attr,
1005         pa_stream_flags_t flags,
1006         pa_cvolume *volume,
1007         pa_stream *sync_stream) {
1008
1009     pa_assert(s);
1010     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1011
1012     return create_stream(PA_STREAM_PLAYBACK, s, dev, attr, flags, volume, sync_stream);
1013 }
1014
1015 int pa_stream_connect_record(
1016         pa_stream *s,
1017         const char *dev,
1018         const pa_buffer_attr *attr,
1019         pa_stream_flags_t flags) {
1020
1021     pa_assert(s);
1022     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1023
1024     return create_stream(PA_STREAM_RECORD, s, dev, attr, flags, NULL, NULL);
1025 }
1026
1027 int pa_stream_write(
1028         pa_stream *s,
1029         const void *data,
1030         size_t length,
1031         void (*free_cb)(void *p),
1032         int64_t offset,
1033         pa_seek_mode_t seek) {
1034
1035     pa_memchunk chunk;
1036     pa_seek_mode_t t_seek;
1037     int64_t t_offset;
1038     size_t t_length;
1039     const void *t_data;
1040
1041     pa_assert(s);
1042     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1043     pa_assert(data);
1044
1045     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1046     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1047     PA_CHECK_VALIDITY(s->context, seek <= PA_SEEK_RELATIVE_END, PA_ERR_INVALID);
1048     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || (seek == PA_SEEK_RELATIVE && offset == 0), PA_ERR_INVALID);
1049
1050     if (length <= 0)
1051         return 0;
1052
1053     t_seek = seek;
1054     t_offset = offset;
1055     t_length = length;
1056     t_data = data;
1057
1058     while (t_length > 0) {
1059
1060         chunk.index = 0;
1061
1062         if (free_cb && !pa_pstream_get_shm(s->context->pstream)) {
1063             chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) t_data, t_length, free_cb, 1);
1064             chunk.length = t_length;
1065         } else {
1066             void *d;
1067
1068             chunk.length = PA_MIN(t_length, pa_mempool_block_size_max(s->context->mempool));
1069             chunk.memblock = pa_memblock_new(s->context->mempool, chunk.length);
1070
1071             d = pa_memblock_acquire(chunk.memblock);
1072             memcpy(d, t_data, chunk.length);
1073             pa_memblock_release(chunk.memblock);
1074         }
1075
1076         pa_pstream_send_memblock(s->context->pstream, s->channel, t_offset, t_seek, &chunk);
1077
1078         t_offset = 0;
1079         t_seek = PA_SEEK_RELATIVE;
1080
1081         t_data = (const uint8_t*) t_data + chunk.length;
1082         t_length -= chunk.length;
1083
1084         pa_memblock_unref(chunk.memblock);
1085     }
1086
1087     if (free_cb && pa_pstream_get_shm(s->context->pstream))
1088         free_cb((void*) data);
1089
1090     if (length < s->requested_bytes)
1091         s->requested_bytes -= length;
1092     else
1093         s->requested_bytes = 0;
1094
1095     /* FIXME!!! ^^^ will break when offset is != 0 and mode is not RELATIVE*/
1096
1097     if (s->direction == PA_STREAM_PLAYBACK) {
1098
1099         /* Update latency request correction */
1100         if (s->write_index_corrections[s->current_write_index_correction].valid) {
1101
1102             if (seek == PA_SEEK_ABSOLUTE) {
1103                 s->write_index_corrections[s->current_write_index_correction].corrupt = FALSE;
1104                 s->write_index_corrections[s->current_write_index_correction].absolute = TRUE;
1105                 s->write_index_corrections[s->current_write_index_correction].value = offset + length;
1106             } else if (seek == PA_SEEK_RELATIVE) {
1107                 if (!s->write_index_corrections[s->current_write_index_correction].corrupt)
1108                     s->write_index_corrections[s->current_write_index_correction].value += offset + length;
1109             } else
1110                 s->write_index_corrections[s->current_write_index_correction].corrupt = TRUE;
1111         }
1112
1113         /* Update the write index in the already available latency data */
1114         if (s->timing_info_valid) {
1115
1116             if (seek == PA_SEEK_ABSOLUTE) {
1117                 s->timing_info.write_index_corrupt = FALSE;
1118                 s->timing_info.write_index = offset + length;
1119             } else if (seek == PA_SEEK_RELATIVE) {
1120                 if (!s->timing_info.write_index_corrupt)
1121                     s->timing_info.write_index += offset + length;
1122             } else
1123                 s->timing_info.write_index_corrupt = TRUE;
1124         }
1125
1126         if (!s->timing_info_valid || s->timing_info.write_index_corrupt)
1127             request_auto_timing_update(s, TRUE);
1128     }
1129
1130     return 0;
1131 }
1132
1133 int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
1134     pa_assert(s);
1135     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1136     pa_assert(data);
1137     pa_assert(length);
1138
1139     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1140     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
1141
1142     if (!s->peek_memchunk.memblock) {
1143
1144         if (pa_memblockq_peek(s->record_memblockq, &s->peek_memchunk) < 0) {
1145             *data = NULL;
1146             *length = 0;
1147             return 0;
1148         }
1149
1150         s->peek_data = pa_memblock_acquire(s->peek_memchunk.memblock);
1151     }
1152
1153     pa_assert(s->peek_data);
1154     *data = (uint8_t*) s->peek_data + s->peek_memchunk.index;
1155     *length = s->peek_memchunk.length;
1156     return 0;
1157 }
1158
1159 int pa_stream_drop(pa_stream *s) {
1160     pa_assert(s);
1161     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1162
1163     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1164     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
1165     PA_CHECK_VALIDITY(s->context, s->peek_memchunk.memblock, PA_ERR_BADSTATE);
1166
1167     pa_memblockq_drop(s->record_memblockq, s->peek_memchunk.length);
1168
1169     /* Fix the simulated local read index */
1170     if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
1171         s->timing_info.read_index += s->peek_memchunk.length;
1172
1173     pa_assert(s->peek_data);
1174     pa_memblock_release(s->peek_memchunk.memblock);
1175     pa_memblock_unref(s->peek_memchunk.memblock);
1176     pa_memchunk_reset(&s->peek_memchunk);
1177
1178     return 0;
1179 }
1180
1181 size_t pa_stream_writable_size(pa_stream *s) {
1182     pa_assert(s);
1183     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1184
1185     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
1186     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
1187
1188     return s->requested_bytes;
1189 }
1190
1191 size_t pa_stream_readable_size(pa_stream *s) {
1192     pa_assert(s);
1193     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1194
1195     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
1196     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
1197
1198     return pa_memblockq_get_length(s->record_memblockq);
1199 }
1200
1201 pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1202     pa_operation *o;
1203     pa_tagstruct *t;
1204     uint32_t tag;
1205
1206     pa_assert(s);
1207     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1208
1209     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1210     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1211
1212     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1213
1214     t = pa_tagstruct_command(s->context, PA_COMMAND_DRAIN_PLAYBACK_STREAM, &tag);
1215     pa_tagstruct_putu32(t, s->channel);
1216     pa_pstream_send_tagstruct(s->context->pstream, t);
1217     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
1218
1219     return o;
1220 }
1221
1222 static pa_usec_t calc_time(pa_stream *s, pa_bool_t ignore_transport) {
1223     pa_usec_t usec;
1224
1225     pa_assert(s);
1226     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1227     pa_assert(s->state == PA_STREAM_READY);
1228     pa_assert(s->direction != PA_STREAM_UPLOAD);
1229     pa_assert(s->timing_info_valid);
1230     pa_assert(s->direction != PA_STREAM_PLAYBACK || !s->timing_info.read_index_corrupt);
1231     pa_assert(s->direction != PA_STREAM_RECORD || !s->timing_info.write_index_corrupt);
1232
1233     if (s->direction == PA_STREAM_PLAYBACK) {
1234         /* The last byte that was written into the output device
1235          * had this time value associated */
1236         usec = pa_bytes_to_usec(s->timing_info.read_index < 0 ? 0 : (uint64_t) s->timing_info.read_index, &s->sample_spec);
1237
1238         if (!s->corked && !s->suspended) {
1239
1240             if (!ignore_transport)
1241                 /* Because the latency info took a little time to come
1242                  * to us, we assume that the real output time is actually
1243                  * a little ahead */
1244                 usec += s->timing_info.transport_usec;
1245
1246             /* However, the output device usually maintains a buffer
1247                too, hence the real sample currently played is a little
1248                back  */
1249             if (s->timing_info.sink_usec >= usec)
1250                 usec = 0;
1251             else
1252                 usec -= s->timing_info.sink_usec;
1253         }
1254
1255     } else if (s->direction == PA_STREAM_RECORD) {
1256         /* The last byte written into the server side queue had
1257          * this time value associated */
1258         usec = pa_bytes_to_usec(s->timing_info.write_index < 0 ? 0 : (uint64_t) s->timing_info.write_index, &s->sample_spec);
1259
1260         if (!s->corked && !s->suspended) {
1261
1262             if (!ignore_transport)
1263                 /* Add transport latency */
1264                 usec += s->timing_info.transport_usec;
1265
1266             /* Add latency of data in device buffer */
1267             usec += s->timing_info.source_usec;
1268
1269             /* If this is a monitor source, we need to correct the
1270              * time by the playback device buffer */
1271             if (s->timing_info.sink_usec >= usec)
1272                 usec = 0;
1273             else
1274                 usec -= s->timing_info.sink_usec;
1275         }
1276     }
1277
1278     return usec;
1279 }
1280
1281 static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1282     pa_operation *o = userdata;
1283     struct timeval local, remote, now;
1284     pa_timing_info *i;
1285     pa_bool_t playing = FALSE;
1286     uint64_t underrun_for = 0, playing_for = 0;
1287
1288     pa_assert(pd);
1289     pa_assert(o);
1290     pa_assert(PA_REFCNT_VALUE(o) >= 1);
1291
1292     if (!o->context || !o->stream)
1293         goto finish;
1294
1295     i = &o->stream->timing_info;
1296
1297     o->stream->timing_info_valid = FALSE;
1298     i->write_index_corrupt = TRUE;
1299     i->read_index_corrupt = TRUE;
1300
1301     if (command != PA_COMMAND_REPLY) {
1302         if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
1303             goto finish;
1304
1305     } else {
1306
1307         if (pa_tagstruct_get_usec(t, &i->sink_usec) < 0 ||
1308             pa_tagstruct_get_usec(t, &i->source_usec) < 0 ||
1309             pa_tagstruct_get_boolean(t, &playing) < 0 ||
1310             pa_tagstruct_get_timeval(t, &local) < 0 ||
1311             pa_tagstruct_get_timeval(t, &remote) < 0 ||
1312             pa_tagstruct_gets64(t, &i->write_index) < 0 ||
1313             pa_tagstruct_gets64(t, &i->read_index) < 0) {
1314
1315             pa_context_fail(o->context, PA_ERR_PROTOCOL);
1316             goto finish;
1317         }
1318
1319         if (o->context->version >= 13 &&
1320             o->stream->direction == PA_STREAM_PLAYBACK)
1321             if (pa_tagstruct_getu64(t, &underrun_for) < 0 ||
1322                 pa_tagstruct_getu64(t, &playing_for) < 0) {
1323
1324                 pa_context_fail(o->context, PA_ERR_PROTOCOL);
1325                 goto finish;
1326             }
1327
1328
1329         if (!pa_tagstruct_eof(t)) {
1330             pa_context_fail(o->context, PA_ERR_PROTOCOL);
1331             goto finish;
1332         }
1333         o->stream->timing_info_valid = TRUE;
1334         i->write_index_corrupt = FALSE;
1335         i->read_index_corrupt = FALSE;
1336
1337         i->playing = (int) playing;
1338         i->since_underrun = playing ? playing_for : underrun_for;
1339
1340         pa_gettimeofday(&now);
1341
1342         /* Calculcate timestamps */
1343         if (pa_timeval_cmp(&local, &remote) <= 0 && pa_timeval_cmp(&remote, &now) <= 0) {
1344             /* local and remote seem to have synchronized clocks */
1345
1346             if (o->stream->direction == PA_STREAM_PLAYBACK)
1347                 i->transport_usec = pa_timeval_diff(&remote, &local);
1348             else
1349                 i->transport_usec = pa_timeval_diff(&now, &remote);
1350
1351             i->synchronized_clocks = TRUE;
1352             i->timestamp = remote;
1353         } else {
1354             /* clocks are not synchronized, let's estimate latency then */
1355             i->transport_usec = pa_timeval_diff(&now, &local)/2;
1356             i->synchronized_clocks = FALSE;
1357             i->timestamp = local;
1358             pa_timeval_add(&i->timestamp, i->transport_usec);
1359         }
1360
1361         /* Invalidate read and write indexes if necessary */
1362         if (tag < o->stream->read_index_not_before)
1363             i->read_index_corrupt = TRUE;
1364
1365         if (tag < o->stream->write_index_not_before)
1366             i->write_index_corrupt = TRUE;
1367
1368         if (o->stream->direction == PA_STREAM_PLAYBACK) {
1369             /* Write index correction */
1370
1371             int n, j;
1372             uint32_t ctag = tag;
1373
1374             /* Go through the saved correction values and add up the
1375              * total correction.*/
1376             for (n = 0, j = o->stream->current_write_index_correction+1;
1377                  n < PA_MAX_WRITE_INDEX_CORRECTIONS;
1378                  n++, j = (j + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS) {
1379
1380                 /* Step over invalid data or out-of-date data */
1381                 if (!o->stream->write_index_corrections[j].valid ||
1382                     o->stream->write_index_corrections[j].tag < ctag)
1383                     continue;
1384
1385                 /* Make sure that everything is in order */
1386                 ctag = o->stream->write_index_corrections[j].tag+1;
1387
1388                 /* Now fix the write index */
1389                 if (o->stream->write_index_corrections[j].corrupt) {
1390                     /* A corrupting seek was made */
1391                     i->write_index_corrupt = TRUE;
1392                 } else if (o->stream->write_index_corrections[j].absolute) {
1393                     /* An absolute seek was made */
1394                     i->write_index = o->stream->write_index_corrections[j].value;
1395                     i->write_index_corrupt = FALSE;
1396                 } else if (!i->write_index_corrupt) {
1397                     /* A relative seek was made */
1398                     i->write_index += o->stream->write_index_corrections[j].value;
1399                 }
1400             }
1401
1402             /* Clear old correction entries */
1403             for (n = 0; n < PA_MAX_WRITE_INDEX_CORRECTIONS; n++) {
1404                 if (!o->stream->write_index_corrections[n].valid)
1405                     continue;
1406
1407                 if (o->stream->write_index_corrections[n].tag <= tag)
1408                     o->stream->write_index_corrections[n].valid = FALSE;
1409             }
1410         }
1411
1412         if (o->stream->direction == PA_STREAM_RECORD) {
1413             /* Read index correction */
1414
1415             if (!i->read_index_corrupt)
1416                 i->read_index -= pa_memblockq_get_length(o->stream->record_memblockq);
1417         }
1418
1419         /* Update smoother */
1420         if (o->stream->smoother) {
1421             pa_usec_t u, x;
1422
1423             u = x = pa_rtclock_usec() - i->transport_usec;
1424
1425             if (o->stream->direction == PA_STREAM_PLAYBACK && o->context->version >= 13) {
1426                 pa_usec_t su;
1427
1428                 /* If we weren't playing then it will take some time
1429                  * until the audio will actually come out through the
1430                  * speakers. Since we follow that timing here, we need
1431                  * to try to fix this up */
1432
1433                 su = pa_bytes_to_usec(i->since_underrun, &o->stream->sample_spec);
1434
1435                 if (su < i->sink_usec)
1436                     x += i->sink_usec - su;
1437             }
1438
1439             if (!i->playing)
1440                 pa_smoother_pause(o->stream->smoother, x);
1441
1442             /* Update the smoother */
1443             if ((o->stream->direction == PA_STREAM_PLAYBACK && !i->read_index_corrupt) ||
1444                 (o->stream->direction == PA_STREAM_RECORD && !i->write_index_corrupt))
1445                 pa_smoother_put(o->stream->smoother, u, calc_time(o->stream, TRUE));
1446
1447             if (i->playing)
1448                 pa_smoother_resume(o->stream->smoother, x);
1449         }
1450     }
1451
1452     o->stream->auto_timing_update_requested = FALSE;
1453
1454     if (o->stream->latency_update_callback)
1455         o->stream->latency_update_callback(o->stream, o->stream->latency_update_userdata);
1456
1457     if (o->callback && o->stream && o->stream->state == PA_STREAM_READY) {
1458         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
1459         cb(o->stream, o->stream->timing_info_valid, o->userdata);
1460     }
1461
1462 finish:
1463
1464     pa_operation_done(o);
1465     pa_operation_unref(o);
1466 }
1467
1468 pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1469     uint32_t tag;
1470     pa_operation *o;
1471     pa_tagstruct *t;
1472     struct timeval now;
1473     int cidx = 0;
1474
1475     pa_assert(s);
1476     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1477
1478     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1479     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1480
1481     if (s->direction == PA_STREAM_PLAYBACK) {
1482         /* Find a place to store the write_index correction data for this entry */
1483         cidx = (s->current_write_index_correction + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS;
1484
1485         /* Check if we could allocate a correction slot. If not, there are too many outstanding queries */
1486         PA_CHECK_VALIDITY_RETURN_NULL(s->context, !s->write_index_corrections[cidx].valid, PA_ERR_INTERNAL);
1487     }
1488     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1489
1490     t = pa_tagstruct_command(
1491             s->context,
1492             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY,
1493             &tag);
1494     pa_tagstruct_putu32(t, s->channel);
1495     pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
1496
1497     pa_pstream_send_tagstruct(s->context->pstream, t);
1498     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_timing_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
1499
1500     if (s->direction == PA_STREAM_PLAYBACK) {
1501         /* Fill in initial correction data */
1502
1503         s->current_write_index_correction = cidx;
1504
1505         s->write_index_corrections[cidx].valid = TRUE;
1506         s->write_index_corrections[cidx].absolute = FALSE;
1507         s->write_index_corrections[cidx].corrupt = FALSE;
1508         s->write_index_corrections[cidx].tag = tag;
1509         s->write_index_corrections[cidx].value = 0;
1510     }
1511
1512     return o;
1513 }
1514
1515 void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
1516     pa_stream *s = userdata;
1517
1518     pa_assert(pd);
1519     pa_assert(s);
1520     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1521
1522     pa_stream_ref(s);
1523
1524     if (command != PA_COMMAND_REPLY) {
1525         if (pa_context_handle_error(s->context, command, t, FALSE) < 0)
1526             goto finish;
1527
1528         pa_stream_set_state(s, PA_STREAM_FAILED);
1529         goto finish;
1530     } else if (!pa_tagstruct_eof(t)) {
1531         pa_context_fail(s->context, PA_ERR_PROTOCOL);
1532         goto finish;
1533     }
1534
1535     pa_stream_set_state(s, PA_STREAM_TERMINATED);
1536
1537 finish:
1538     pa_stream_unref(s);
1539 }
1540
1541 int pa_stream_disconnect(pa_stream *s) {
1542     pa_tagstruct *t;
1543     uint32_t tag;
1544
1545     pa_assert(s);
1546     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1547
1548     PA_CHECK_VALIDITY(s->context, s->channel_valid, PA_ERR_BADSTATE);
1549     PA_CHECK_VALIDITY(s->context, s->context->state == PA_CONTEXT_READY, PA_ERR_BADSTATE);
1550
1551     pa_stream_ref(s);
1552
1553     t = pa_tagstruct_command(
1554             s->context,
1555             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM :
1556             (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM),
1557             &tag);
1558     pa_tagstruct_putu32(t, s->channel);
1559     pa_pstream_send_tagstruct(s->context->pstream, t);
1560     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s, NULL);
1561
1562     pa_stream_unref(s);
1563     return 0;
1564 }
1565
1566 void pa_stream_set_read_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
1567     pa_assert(s);
1568     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1569
1570     if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1571         return;
1572
1573     s->read_callback = cb;
1574     s->read_userdata = userdata;
1575 }
1576
1577 void pa_stream_set_write_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
1578     pa_assert(s);
1579     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1580
1581     if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1582         return;
1583
1584     s->write_callback = cb;
1585     s->write_userdata = userdata;
1586 }
1587
1588 void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1589     pa_assert(s);
1590     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1591
1592     if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1593         return;
1594
1595     s->state_callback = cb;
1596     s->state_userdata = userdata;
1597 }
1598
1599 void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1600     pa_assert(s);
1601     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1602
1603     if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1604         return;
1605
1606     s->overflow_callback = cb;
1607     s->overflow_userdata = userdata;
1608 }
1609
1610 void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1611     pa_assert(s);
1612     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1613
1614     if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1615         return;
1616
1617     s->underflow_callback = cb;
1618     s->underflow_userdata = userdata;
1619 }
1620
1621 void pa_stream_set_latency_update_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1622     pa_assert(s);
1623     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1624
1625     if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1626         return;
1627
1628     s->latency_update_callback = cb;
1629     s->latency_update_userdata = userdata;
1630 }
1631
1632 void pa_stream_set_moved_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1633     pa_assert(s);
1634     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1635
1636     if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1637         return;
1638
1639     s->moved_callback = cb;
1640     s->moved_userdata = userdata;
1641 }
1642
1643 void pa_stream_set_suspended_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1644     pa_assert(s);
1645     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1646
1647     if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1648         return;
1649
1650     s->suspended_callback = cb;
1651     s->suspended_userdata = userdata;
1652 }
1653
1654 void pa_stream_set_started_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1655     pa_assert(s);
1656     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1657
1658     if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1659         return;
1660
1661     s->started_callback = cb;
1662     s->started_userdata = userdata;
1663 }
1664
1665 void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
1666     pa_operation *o = userdata;
1667     int success = 1;
1668
1669     pa_assert(pd);
1670     pa_assert(o);
1671     pa_assert(PA_REFCNT_VALUE(o) >= 1);
1672
1673     if (!o->context)
1674         goto finish;
1675
1676     if (command != PA_COMMAND_REPLY) {
1677         if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
1678             goto finish;
1679
1680         success = 0;
1681     } else if (!pa_tagstruct_eof(t)) {
1682         pa_context_fail(o->context, PA_ERR_PROTOCOL);
1683         goto finish;
1684     }
1685
1686     if (o->callback) {
1687         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
1688         cb(o->stream, success, o->userdata);
1689     }
1690
1691 finish:
1692     pa_operation_done(o);
1693     pa_operation_unref(o);
1694 }
1695
1696 pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, void *userdata) {
1697     pa_operation *o;
1698     pa_tagstruct *t;
1699     uint32_t tag;
1700
1701     pa_assert(s);
1702     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1703
1704     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1705     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1706
1707     s->corked = b;
1708
1709     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1710
1711     t = pa_tagstruct_command(
1712             s->context,
1713             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CORK_PLAYBACK_STREAM : PA_COMMAND_CORK_RECORD_STREAM,
1714             &tag);
1715     pa_tagstruct_putu32(t, s->channel);
1716     pa_tagstruct_put_boolean(t, !!b);
1717     pa_pstream_send_tagstruct(s->context->pstream, t);
1718     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
1719
1720     check_smoother_status(s, FALSE, FALSE, FALSE);
1721
1722     /* This might cause the indexes to hang/start again, hence
1723      * let's request a timing update */
1724     request_auto_timing_update(s, TRUE);
1725
1726     return o;
1727 }
1728
1729 static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command, pa_stream_success_cb_t cb, void *userdata) {
1730     pa_tagstruct *t;
1731     pa_operation *o;
1732     uint32_t tag;
1733
1734     pa_assert(s);
1735     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1736
1737     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1738
1739     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1740
1741     t = pa_tagstruct_command(s->context, command, &tag);
1742     pa_tagstruct_putu32(t, s->channel);
1743     pa_pstream_send_tagstruct(s->context->pstream, t);
1744     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
1745
1746     return o;
1747 }
1748
1749 pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1750     pa_operation *o;
1751
1752     pa_assert(s);
1753     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1754
1755     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1756     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1757
1758     if (!(o = stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata)))
1759         return NULL;
1760
1761     if (s->direction == PA_STREAM_PLAYBACK) {
1762
1763         if (s->write_index_corrections[s->current_write_index_correction].valid)
1764             s->write_index_corrections[s->current_write_index_correction].corrupt = TRUE;
1765
1766         if (s->buffer_attr.prebuf > 0)
1767             check_smoother_status(s, FALSE, FALSE, TRUE);
1768
1769         /* This will change the write index, but leave the
1770          * read index untouched. */
1771         invalidate_indexes(s, FALSE, TRUE);
1772
1773     } else
1774         /* For record streams this has no influence on the write
1775          * index, but the read index might jump. */
1776         invalidate_indexes(s, TRUE, FALSE);
1777
1778     return o;
1779 }
1780
1781 pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1782     pa_operation *o;
1783
1784     pa_assert(s);
1785     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1786
1787     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1788     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1789     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
1790
1791     if (!(o = stream_send_simple_command(s, PA_COMMAND_PREBUF_PLAYBACK_STREAM, cb, userdata)))
1792         return NULL;
1793
1794     /* This might cause the read index to hang again, hence
1795      * let's request a timing update */
1796     request_auto_timing_update(s, TRUE);
1797
1798     return o;
1799 }
1800
1801 pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1802     pa_operation *o;
1803
1804     pa_assert(s);
1805     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1806
1807     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1808     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1809     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
1810
1811     if (!(o = stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata)))
1812         return NULL;
1813
1814     /* This might cause the read index to start moving again, hence
1815      * let's request a timing update */
1816     request_auto_timing_update(s, TRUE);
1817
1818     return o;
1819 }
1820
1821 pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_success_cb_t cb, void *userdata) {
1822     pa_operation *o;
1823
1824     pa_assert(s);
1825     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1826     pa_assert(name);
1827
1828     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1829     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1830
1831     if (s->context->version >= 13) {
1832         pa_proplist *p = pa_proplist_new();
1833
1834         pa_proplist_sets(p, PA_PROP_APPLICATION_NAME, name);
1835         o = pa_stream_proplist_update(s, PA_UPDATE_REPLACE, p, cb, userdata);
1836         pa_proplist_free(p);
1837     } else {
1838         pa_tagstruct *t;
1839         uint32_t tag;
1840
1841         o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1842         t = pa_tagstruct_command(
1843                 s->context,
1844                 s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_NAME : PA_COMMAND_SET_PLAYBACK_STREAM_NAME,
1845                 &tag);
1846         pa_tagstruct_putu32(t, s->channel);
1847         pa_tagstruct_puts(t, name);
1848         pa_pstream_send_tagstruct(s->context->pstream, t);
1849         pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
1850     }
1851
1852     return o;
1853 }
1854
1855 int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
1856     pa_usec_t usec;
1857
1858     pa_assert(s);
1859     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1860
1861     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1862     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1863     PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
1864     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
1865     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
1866
1867     if (s->smoother)
1868         usec = pa_smoother_get(s->smoother, pa_rtclock_usec());
1869     else
1870         usec = calc_time(s, FALSE);
1871
1872     /* Make sure the time runs monotonically */
1873     if (!(s->flags & PA_STREAM_NOT_MONOTONOUS)) {
1874         if (usec < s->previous_time)
1875             usec = s->previous_time;
1876         else
1877             s->previous_time = usec;
1878     }
1879
1880     if (r_usec)
1881         *r_usec = usec;
1882
1883     return 0;
1884 }
1885
1886 static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t a, pa_usec_t b, int *negative) {
1887     pa_assert(s);
1888     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1889
1890     if (negative)
1891         *negative = 0;
1892
1893     if (a >= b)
1894         return a-b;
1895     else {
1896         if (negative && s->direction == PA_STREAM_RECORD) {
1897             *negative = 1;
1898             return b-a;
1899         } else
1900             return 0;
1901     }
1902 }
1903
1904 int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
1905     pa_usec_t t, c;
1906     int r;
1907     int64_t cindex;
1908
1909     pa_assert(s);
1910     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1911     pa_assert(r_usec);
1912
1913     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1914     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1915     PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
1916     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
1917     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
1918
1919     if ((r = pa_stream_get_time(s, &t)) < 0)
1920         return r;
1921
1922     if (s->direction == PA_STREAM_PLAYBACK)
1923         cindex = s->timing_info.write_index;
1924     else
1925         cindex = s->timing_info.read_index;
1926
1927     if (cindex < 0)
1928         cindex = 0;
1929
1930     c = pa_bytes_to_usec(cindex, &s->sample_spec);
1931
1932     if (s->direction == PA_STREAM_PLAYBACK)
1933         *r_usec = time_counter_diff(s, c, t, negative);
1934     else
1935         *r_usec = time_counter_diff(s, t, c, negative);
1936
1937     return 0;
1938 }
1939
1940 const pa_timing_info* pa_stream_get_timing_info(pa_stream *s) {
1941     pa_assert(s);
1942     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1943
1944     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1945     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1946     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->timing_info_valid, PA_ERR_BADSTATE);
1947
1948     return &s->timing_info;
1949 }
1950
1951 const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s) {
1952     pa_assert(s);
1953     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1954
1955     return &s->sample_spec;
1956 }
1957
1958 const pa_channel_map* pa_stream_get_channel_map(pa_stream *s) {
1959     pa_assert(s);
1960     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1961
1962     return &s->channel_map;
1963 }
1964
1965 const pa_buffer_attr* pa_stream_get_buffer_attr(pa_stream *s) {
1966     pa_assert(s);
1967     pa_assert(PA_REFCNT_VALUE(s) >= 1);
1968
1969     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1970     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1971     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 9, PA_ERR_NOTSUPPORTED);
1972
1973     return &s->buffer_attr;
1974 }
1975
1976 static void stream_set_buffer_attr_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
1977     pa_operation *o = userdata;
1978     int success = 1;
1979
1980     pa_assert(pd);
1981     pa_assert(o);
1982     pa_assert(PA_REFCNT_VALUE(o) >= 1);
1983
1984     if (!o->context)
1985         goto finish;
1986
1987     if (command != PA_COMMAND_REPLY) {
1988         if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
1989             goto finish;
1990
1991         success = 0;
1992     } else {
1993
1994         if (o->stream->direction == PA_STREAM_PLAYBACK) {
1995             if (pa_tagstruct_getu32(t, &o->stream->buffer_attr.maxlength) < 0 ||
1996                 pa_tagstruct_getu32(t, &o->stream->buffer_attr.tlength) < 0 ||
1997                 pa_tagstruct_getu32(t, &o->stream->buffer_attr.prebuf) < 0 ||
1998                 pa_tagstruct_getu32(t, &o->stream->buffer_attr.minreq) < 0) {
1999                 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2000                 goto finish;
2001             }
2002         } else if (o->stream->direction == PA_STREAM_RECORD) {
2003             if (pa_tagstruct_getu32(t, &o->stream->buffer_attr.maxlength) < 0 ||
2004                 pa_tagstruct_getu32(t, &o->stream->buffer_attr.fragsize) < 0) {
2005                 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2006                 goto finish;
2007             }
2008         }
2009
2010         if (!pa_tagstruct_eof(t)) {
2011             pa_context_fail(o->context, PA_ERR_PROTOCOL);
2012             goto finish;
2013         }
2014     }
2015
2016     if (o->callback) {
2017         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
2018         cb(o->stream, success, o->userdata);
2019     }
2020
2021 finish:
2022     pa_operation_done(o);
2023     pa_operation_unref(o);
2024 }
2025
2026
2027 pa_operation* pa_stream_set_buffer_attr(pa_stream *s, const pa_buffer_attr *attr, pa_stream_success_cb_t cb, void *userdata) {
2028     pa_operation *o;
2029     pa_tagstruct *t;
2030     uint32_t tag;
2031
2032     pa_assert(s);
2033     pa_assert(PA_REFCNT_VALUE(s) >= 1);
2034     pa_assert(attr);
2035
2036     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2037     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2038     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2039
2040     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2041
2042     t = pa_tagstruct_command(
2043             s->context,
2044             s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_BUFFER_ATTR : PA_COMMAND_SET_PLAYBACK_STREAM_BUFFER_ATTR,
2045             &tag);
2046     pa_tagstruct_putu32(t, s->channel);
2047
2048     pa_tagstruct_putu32(t, attr->maxlength);
2049
2050     if (s->direction == PA_STREAM_PLAYBACK)
2051         pa_tagstruct_put(
2052                 t,
2053                 PA_TAG_U32, attr->tlength,
2054                 PA_TAG_U32, attr->prebuf,
2055                 PA_TAG_U32, attr->minreq,
2056                 PA_TAG_INVALID);
2057     else
2058         pa_tagstruct_putu32(t, attr->fragsize);
2059
2060     if (s->context->version >= 13)
2061         pa_tagstruct_put_boolean(t, !!(s->flags & PA_STREAM_ADJUST_LATENCY));
2062
2063     pa_pstream_send_tagstruct(s->context->pstream, t);
2064     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_set_buffer_attr_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
2065
2066     /* This might cause changes in the read/write indexex, hence let's
2067      * request a timing update */
2068     request_auto_timing_update(s, TRUE);
2069
2070     return o;
2071 }
2072
2073 uint32_t pa_stream_get_device_index(pa_stream *s) {
2074     pa_assert(s);
2075     pa_assert(PA_REFCNT_VALUE(s) >= 1);
2076
2077     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2078     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2079     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED, PA_INVALID_INDEX);
2080     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->device_index != PA_INVALID_INDEX, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2081
2082     return s->device_index;
2083 }
2084
2085 const char *pa_stream_get_device_name(pa_stream *s) {
2086     pa_assert(s);
2087     pa_assert(PA_REFCNT_VALUE(s) >= 1);
2088
2089     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2090     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2091     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2092     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->device_name, PA_ERR_BADSTATE);
2093
2094     return s->device_name;
2095 }
2096
2097 int pa_stream_is_suspended(pa_stream *s) {
2098     pa_assert(s);
2099     pa_assert(PA_REFCNT_VALUE(s) >= 1);
2100
2101     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2102     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2103     PA_CHECK_VALIDITY(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2104
2105     return s->suspended;
2106 }
2107
2108 int pa_stream_is_corked(pa_stream *s) {
2109     pa_assert(s);
2110     pa_assert(PA_REFCNT_VALUE(s) >= 1);
2111
2112     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2113     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2114
2115     return s->corked;
2116 }
2117
2118 static void stream_update_sample_rate_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
2119     pa_operation *o = userdata;
2120     int success = 1;
2121
2122     pa_assert(pd);
2123     pa_assert(o);
2124     pa_assert(PA_REFCNT_VALUE(o) >= 1);
2125
2126     if (!o->context)
2127         goto finish;
2128
2129     if (command != PA_COMMAND_REPLY) {
2130         if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
2131             goto finish;
2132
2133         success = 0;
2134     } else {
2135
2136         if (!pa_tagstruct_eof(t)) {
2137             pa_context_fail(o->context, PA_ERR_PROTOCOL);
2138             goto finish;
2139         }
2140     }
2141
2142     o->stream->sample_spec.rate = PA_PTR_TO_UINT(o->private);
2143     pa_assert(pa_sample_spec_valid(&o->stream->sample_spec));
2144
2145     if (o->callback) {
2146         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
2147         cb(o->stream, success, o->userdata);
2148     }
2149
2150 finish:
2151     pa_operation_done(o);
2152     pa_operation_unref(o);
2153 }
2154
2155
2156 pa_operation *pa_stream_update_sample_rate(pa_stream *s, uint32_t rate, pa_stream_success_cb_t cb, void *userdata) {
2157     pa_operation *o;
2158     pa_tagstruct *t;
2159     uint32_t tag;
2160
2161     pa_assert(s);
2162     pa_assert(PA_REFCNT_VALUE(s) >= 1);
2163
2164     PA_CHECK_VALIDITY_RETURN_NULL(s->context, rate > 0 && rate <= PA_RATE_MAX, PA_ERR_INVALID);
2165     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2166     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2167     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->flags & PA_STREAM_VARIABLE_RATE, PA_ERR_BADSTATE);
2168     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2169
2170     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2171     o->private = PA_UINT_TO_PTR(rate);
2172
2173     t = pa_tagstruct_command(
2174             s->context,
2175             s->direction == PA_STREAM_RECORD ? PA_COMMAND_UPDATE_RECORD_STREAM_SAMPLE_RATE : PA_COMMAND_UPDATE_PLAYBACK_STREAM_SAMPLE_RATE,
2176             &tag);
2177     pa_tagstruct_putu32(t, s->channel);
2178     pa_tagstruct_putu32(t, rate);
2179
2180     pa_pstream_send_tagstruct(s->context->pstream, t);
2181     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_update_sample_rate_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
2182
2183     return o;
2184 }
2185
2186 pa_operation *pa_stream_proplist_update(pa_stream *s, pa_update_mode_t mode, pa_proplist *p, pa_stream_success_cb_t cb, void *userdata) {
2187     pa_operation *o;
2188     pa_tagstruct *t;
2189     uint32_t tag;
2190
2191     pa_assert(s);
2192     pa_assert(PA_REFCNT_VALUE(s) >= 1);
2193
2194     PA_CHECK_VALIDITY_RETURN_NULL(s->context, mode == PA_UPDATE_SET || mode == PA_UPDATE_MERGE || mode == PA_UPDATE_REPLACE, PA_ERR_INVALID);
2195     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2196     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2197     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED);
2198
2199     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2200
2201     t = pa_tagstruct_command(
2202             s->context,
2203             s->direction == PA_STREAM_RECORD ? PA_COMMAND_UPDATE_RECORD_STREAM_PROPLIST : PA_COMMAND_UPDATE_PLAYBACK_STREAM_PROPLIST,
2204             &tag);
2205     pa_tagstruct_putu32(t, s->channel);
2206     pa_tagstruct_putu32(t, (uint32_t) mode);
2207     pa_tagstruct_put_proplist(t, p);
2208
2209     pa_pstream_send_tagstruct(s->context->pstream, t);
2210     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
2211
2212     /* Please note that we don't update s->proplist here, because we
2213      * don't export that field */
2214
2215     return o;
2216 }
2217
2218 pa_operation *pa_stream_proplist_remove(pa_stream *s, const char *const keys[], pa_stream_success_cb_t cb, void *userdata) {
2219     pa_operation *o;
2220     pa_tagstruct *t;
2221     uint32_t tag;
2222     const char * const*k;
2223
2224     pa_assert(s);
2225     pa_assert(PA_REFCNT_VALUE(s) >= 1);
2226
2227     PA_CHECK_VALIDITY_RETURN_NULL(s->context, keys && keys[0], PA_ERR_INVALID);
2228     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2229     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2230     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED);
2231
2232     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2233
2234     t = pa_tagstruct_command(
2235             s->context,
2236             s->direction == PA_STREAM_RECORD ? PA_COMMAND_REMOVE_RECORD_STREAM_PROPLIST : PA_COMMAND_REMOVE_PLAYBACK_STREAM_PROPLIST,
2237             &tag);
2238     pa_tagstruct_putu32(t, s->channel);
2239
2240     for (k = keys; *k; k++)
2241         pa_tagstruct_puts(t, *k);
2242
2243     pa_tagstruct_puts(t, NULL);
2244
2245     pa_pstream_send_tagstruct(s->context->pstream, t);
2246     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
2247
2248     /* Please note that we don't update s->proplist here, because we
2249      * don't export that field */
2250
2251     return o;
2252 }
2253
2254 int pa_stream_set_monitor_stream(pa_stream *s, uint32_t sink_input_idx) {
2255     pa_assert(s);
2256     pa_assert(PA_REFCNT_VALUE(s) >= 1);
2257
2258     PA_CHECK_VALIDITY(s->context, sink_input_idx != PA_INVALID_INDEX, PA_ERR_INVALID);
2259     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
2260     PA_CHECK_VALIDITY(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED);
2261
2262     s->direct_on_input = sink_input_idx;
2263
2264     return 0;
2265 }
2266
2267 uint32_t pa_stream_get_monitor_stream(pa_stream *s) {
2268     pa_assert(s);
2269     pa_assert(PA_REFCNT_VALUE(s) >= 1);
2270
2271     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direct_on_input != PA_INVALID_INDEX, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2272     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED, PA_INVALID_INDEX);
2273
2274     return s->direct_on_input;
2275 }