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