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