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