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