src/pulse/stream.c

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of PulseAudio.
   5
   6   Copyright 2004-2006 Lennart Poettering
   7   Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
   8
   9   PulseAudio is free software; you can redistribute it and/or modify
  10   it under the terms of the GNU Lesser General Public License as published
  11   by the Free Software Foundation; either version 2 of the License,
  12   or (at your option) any later version.
  13
  14   PulseAudio is distributed in the hope that it will be useful, but
  15   WITHOUT ANY WARRANTY; without even the implied warranty of
  16   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  17   General Public License for more details.
  18
  19   You should have received a copy of the GNU Lesser General Public License
  20   along with PulseAudio; if not, write to the Free Software
  21   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  22   USA.
  23 ***/
  24
  25 #ifdef HAVE_CONFIG_H
  26 #include <config.h>
  27 #endif
  28
  29 #include <assert.h>
  30 #include <string.h>
  31 #include <stdio.h>
  32 #include <string.h>
  33
  34 #include <pulse/def.h>
  35 #include <pulse/timeval.h>
  36 #include <pulse/xmalloc.h>
  37
  38 #include <pulsecore/pstream-util.h>
  39 #include <pulsecore/log.h>
  40 #include <pulsecore/hashmap.h>
  41
  42 #include "internal.h"
  43
  44 #define LATENCY_IPOL_INTERVAL_USEC (100000L)
  45
  46 pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) {
  47     pa_stream *s;
  48     int i;
  49
  50     assert(c);
  51
  52     PA_CHECK_VALIDITY_RETURN_NULL(c, ss && pa_sample_spec_valid(ss), PA_ERR_INVALID);
  53     PA_CHECK_VALIDITY_RETURN_NULL(c, !map || (pa_channel_map_valid(map) && map->channels == ss->channels), PA_ERR_INVALID);
  54
  55     s = pa_xnew(pa_stream, 1);
  56     s->ref = 1;
  57     s->context = c;
  58     s->mainloop = c->mainloop;
  59
  60     s->read_callback = NULL;
  61     s->read_userdata = NULL;
  62     s->write_callback = NULL;
  63     s->write_userdata = NULL;
  64     s->state_callback = NULL;
  65     s->state_userdata = NULL;
  66     s->overflow_callback = NULL;
  67     s->overflow_userdata = NULL;
  68     s->underflow_callback = NULL;
  69     s->underflow_userdata = NULL;
  70     s->latency_update_callback = NULL;
  71     s->latency_update_userdata = NULL;
  72
  73     s->direction = PA_STREAM_NODIRECTION;
  74     s->name = pa_xstrdup(name);
  75     s->sample_spec = *ss;
  76     s->flags = 0;
  77
  78     if (map)
  79         s->channel_map = *map;
  80     else
  81         pa_channel_map_init_auto(&s->channel_map, ss->channels, PA_CHANNEL_MAP_DEFAULT);
  82
  83     s->channel = 0;
  84     s->channel_valid = 0;
  85     s->syncid = c->csyncid++;
  86     s->device_index = PA_INVALID_INDEX;
  87     s->requested_bytes = 0;
  88     s->state = PA_STREAM_UNCONNECTED;
  89     memset(&s->buffer_attr, 0, sizeof(s->buffer_attr));
  90
  91     s->peek_memchunk.index = 0;
  92     s->peek_memchunk.length = 0;
  93     s->peek_memchunk.memblock = NULL;
  94
  95     s->record_memblockq = NULL;
  96
  97     s->previous_time = 0;
  98     s->timing_info_valid = 0;
  99     s->read_index_not_before = 0;
 100     s->write_index_not_before = 0;
 101
 102     for (i = 0; i < PA_MAX_WRITE_INDEX_CORRECTIONS; i++)
 103         s->write_index_corrections[i].valid = 0;
 104     s->current_write_index_correction = 0;
 105
 106     s->corked = 0;
 107
 108     s->cached_time_valid = 0;
 109
 110     s->auto_timing_update_event = NULL;
 111     s->auto_timing_update_requested = 0;
 112
 113     /* Refcounting is strictly one-way: from the "bigger" to the "smaller" object. */
 114     PA_LLIST_PREPEND(pa_stream, c->streams, s);
 115     pa_stream_ref(s);
 116
 117     return s;
 118 }
 119
 120 static void stream_free(pa_stream *s) {
 121     assert(s && !s->context && !s->channel_valid);
 122
 123     if (s->auto_timing_update_event) {
 124         assert(s->mainloop);
 125         s->mainloop->time_free(s->auto_timing_update_event);
 126     }
 127
 128     if (s->peek_memchunk.memblock)
 129         pa_memblock_unref(s->peek_memchunk.memblock);
 130
 131     if (s->record_memblockq)
 132         pa_memblockq_free(s->record_memblockq);
 133
 134     pa_xfree(s->name);
 135     pa_xfree(s);
 136 }
 137
 138 void pa_stream_unref(pa_stream *s) {
 139     assert(s);
 140     assert(s->ref >= 1);
 141
 142     if (--(s->ref) == 0)
 143         stream_free(s);
 144 }
 145
 146 pa_stream* pa_stream_ref(pa_stream *s) {
 147     assert(s);
 148     assert(s->ref >= 1);
 149
 150     s->ref++;
 151     return s;
 152 }
 153
 154 pa_stream_state_t pa_stream_get_state(pa_stream *s) {
 155     assert(s);
 156     assert(s->ref >= 1);
 157
 158     return s->state;
 159 }
 160
 161 pa_context* pa_stream_get_context(pa_stream *s) {
 162     assert(s);
 163     assert(s->ref >= 1);
 164
 165     return s->context;
 166 }
 167
 168 uint32_t pa_stream_get_index(pa_stream *s) {
 169     assert(s);
 170     assert(s->ref >= 1);
 171
 172     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
 173
 174     return s->device_index;
 175 }
 176
 177 void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
 178     assert(s);
 179     assert(s->ref >= 1);
 180
 181     if (s->state == st)
 182         return;
 183
 184     pa_stream_ref(s);
 185
 186     s->state = st;
 187     if (s->state_callback)
 188         s->state_callback(s, s->state_userdata);
 189
 190     if ((st == PA_STREAM_FAILED || st == PA_STREAM_TERMINATED) && s->context) {
 191
 192         /* Detach from context */
 193         pa_operation *o, *n;
 194
 195         /* Unref all operatio object that point to us */
 196         for (o = s->context->operations; o; o = n) {
 197             n = o->next;
 198
 199             if (o->stream == s)
 200                 pa_operation_cancel(o);
 201         }
 202
 203         /* Drop all outstanding replies for this stream */
 204         if (s->context->pdispatch)
 205             pa_pdispatch_unregister_reply(s->context->pdispatch, s);
 206
 207         if (s->channel_valid)
 208             pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL);
 209
 210         PA_LLIST_REMOVE(pa_stream, s->context->streams, s);
 211         pa_stream_unref(s);
 212
 213         s->channel = 0;
 214         s->channel_valid = 0;
 215
 216         s->context = NULL;
 217     }
 218
 219     pa_stream_unref(s);
 220 }
 221
 222 void pa_command_stream_killed(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 223     pa_context *c = userdata;
 224     pa_stream *s;
 225     uint32_t channel;
 226
 227     assert(pd);
 228     assert(command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED);
 229     assert(t);
 230     assert(c);
 231
 232     pa_context_ref(c);
 233
 234     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 235         !pa_tagstruct_eof(t)) {
 236         pa_context_fail(c, PA_ERR_PROTOCOL);
 237         goto finish;
 238     }
 239
 240     if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, channel)))
 241         goto finish;
 242
 243     pa_context_set_error(c, PA_ERR_KILLED);
 244     pa_stream_set_state(s, PA_STREAM_FAILED);
 245
 246 finish:
 247     pa_context_unref(c);
 248 }
 249
 250 void pa_command_request(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 251     pa_stream *s;
 252     pa_context *c = userdata;
 253     uint32_t bytes, channel;
 254
 255     assert(pd);
 256     assert(command == PA_COMMAND_REQUEST);
 257     assert(t);
 258     assert(c);
 259
 260     pa_context_ref(c);
 261
 262     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 263         pa_tagstruct_getu32(t, &bytes) < 0 ||
 264         !pa_tagstruct_eof(t)) {
 265         pa_context_fail(c, PA_ERR_PROTOCOL);
 266         goto finish;
 267     }
 268
 269     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
 270         goto finish;
 271
 272     if (s->state == PA_STREAM_READY) {
 273         s->requested_bytes += bytes;
 274
 275         if (s->requested_bytes > 0 && s->write_callback)
 276             s->write_callback(s, s->requested_bytes, s->write_userdata);
 277     }
 278
 279 finish:
 280     pa_context_unref(c);
 281 }
 282
 283 void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 284     pa_stream *s;
 285     pa_context *c = userdata;
 286     uint32_t channel;
 287
 288     assert(pd);
 289     assert(command == PA_COMMAND_OVERFLOW || command == PA_COMMAND_UNDERFLOW);
 290     assert(t);
 291     assert(c);
 292
 293     pa_context_ref(c);
 294
 295     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 296         !pa_tagstruct_eof(t)) {
 297         pa_context_fail(c, PA_ERR_PROTOCOL);
 298         goto finish;
 299     }
 300
 301     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
 302         goto finish;
 303
 304     if (s->state == PA_STREAM_READY) {
 305
 306         if (command == PA_COMMAND_OVERFLOW) {
 307             if (s->overflow_callback)
 308                 s->overflow_callback(s, s->overflow_userdata);
 309         } else if (command == PA_COMMAND_UNDERFLOW) {
 310             if (s->underflow_callback)
 311                 s->underflow_callback(s, s->underflow_userdata);
 312         }
 313     }
 314
 315  finish:
 316     pa_context_unref(c);
 317 }
 318
 319 static void request_auto_timing_update(pa_stream *s, int force) {
 320     struct timeval next;
 321     assert(s);
 322
 323     if (!(s->flags & PA_STREAM_AUTO_TIMING_UPDATE))
 324         return;
 325
 326     if (s->state == PA_STREAM_READY &&
 327         (force || !s->auto_timing_update_requested)) {
 328         pa_operation *o;
 329
 330 /*         pa_log("automatically requesting new timing data");   */
 331
 332         if ((o = pa_stream_update_timing_info(s, NULL, NULL))) {
 333             pa_operation_unref(o);
 334             s->auto_timing_update_requested = 1;
 335         }
 336     }
 337
 338     pa_gettimeofday(&next);
 339     pa_timeval_add(&next, LATENCY_IPOL_INTERVAL_USEC);
 340     s->mainloop->time_restart(s->auto_timing_update_event, &next);
 341 }
 342
 343 static void invalidate_indexes(pa_stream *s, int r, int w) {
 344     assert(s);
 345
 346 /*     pa_log("invalidate r:%u w:%u tag:%u", r, w, s->context->ctag); */
 347
 348     if (s->state != PA_STREAM_READY)
 349         return;
 350
 351     if (w) {
 352         s->write_index_not_before = s->context->ctag;
 353
 354         if (s->timing_info_valid)
 355             s->timing_info.write_index_corrupt = 1;
 356
 357 /*         pa_log("write_index invalidated"); */
 358     }
 359
 360     if (r) {
 361         s->read_index_not_before = s->context->ctag;
 362
 363         if (s->timing_info_valid)
 364             s->timing_info.read_index_corrupt = 1;
 365
 366 /*         pa_log("read_index invalidated"); */
 367     }
 368
 369     if ((s->direction == PA_STREAM_PLAYBACK && r) ||
 370         (s->direction == PA_STREAM_RECORD && w))
 371         s->cached_time_valid = 0;
 372
 373     request_auto_timing_update(s, 1);
 374 }
 375
 376 static void auto_timing_update_callback(PA_GCC_UNUSED pa_mainloop_api *m, PA_GCC_UNUSED pa_time_event *e, PA_GCC_UNUSED const struct timeval *tv, void *userdata) {
 377     pa_stream *s = userdata;
 378
 379 /*     pa_log("time event");    */
 380
 381     pa_stream_ref(s);
 382     request_auto_timing_update(s, 0);
 383     pa_stream_unref(s);
 384 }
 385
 386 void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 387     pa_stream *s = userdata;
 388
 389     assert(pd);
 390     assert(s);
 391     assert(s->state == PA_STREAM_CREATING);
 392
 393     pa_stream_ref(s);
 394
 395     if (command != PA_COMMAND_REPLY) {
 396         if (pa_context_handle_error(s->context, command, t) < 0)
 397             goto finish;
 398
 399         pa_stream_set_state(s, PA_STREAM_FAILED);
 400         goto finish;
 401     }
 402
 403     if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
 404         ((s->direction != PA_STREAM_UPLOAD) && pa_tagstruct_getu32(t, &s->device_index) < 0) ||
 405         ((s->direction != PA_STREAM_RECORD) && pa_tagstruct_getu32(t, &s->requested_bytes) < 0)) {
 406         pa_context_fail(s->context, PA_ERR_PROTOCOL);
 407         goto finish;
 408     }
 409
 410     if (pa_context_get_server_protocol_version(s->context) >= 9) {
 411         if (s->direction == PA_STREAM_PLAYBACK) {
 412             if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
 413                 pa_tagstruct_getu32(t, &s->buffer_attr.tlength) < 0 ||
 414                 pa_tagstruct_getu32(t, &s->buffer_attr.prebuf) < 0 ||
 415                 pa_tagstruct_getu32(t, &s->buffer_attr.minreq) < 0) {
 416                 pa_context_fail(s->context, PA_ERR_PROTOCOL);
 417                 goto finish;
 418             }
 419         } else if (s->direction == PA_STREAM_RECORD) {
 420             if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
 421                 pa_tagstruct_getu32(t, &s->buffer_attr.fragsize) < 0) {
 422                 pa_context_fail(s->context, PA_ERR_PROTOCOL);
 423                 goto finish;
 424             }
 425         }
 426     }
 427
 428     if (!pa_tagstruct_eof(t)) {
 429         pa_context_fail(s->context, PA_ERR_PROTOCOL);
 430         goto finish;
 431     }
 432
 433     if (s->direction == PA_STREAM_RECORD) {
 434         assert(!s->record_memblockq);
 435
 436         s->record_memblockq = pa_memblockq_new(
 437                 0,
 438                 s->buffer_attr.maxlength,
 439                 0,
 440                 pa_frame_size(&s->sample_spec),
 441                 1,
 442                 0,
 443                 NULL);
 444     }
 445
 446     s->channel_valid = 1;
 447     pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
 448
 449     pa_stream_set_state(s, PA_STREAM_READY);
 450
 451     if (s->direction != PA_STREAM_UPLOAD &&
 452         s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
 453         struct timeval tv;
 454
 455         pa_gettimeofday(&tv);
 456         tv.tv_usec += LATENCY_IPOL_INTERVAL_USEC; /* every 100 ms */
 457
 458         assert(!s->auto_timing_update_event);
 459         s->auto_timing_update_event = s->mainloop->time_new(s->mainloop, &tv, &auto_timing_update_callback, s);
 460
 461         request_auto_timing_update(s, 1);
 462     }
 463
 464     if (s->requested_bytes > 0 && s->ref > 1 && s->write_callback)
 465         s->write_callback(s, s->requested_bytes, s->write_userdata);
 466
 467 finish:
 468     pa_stream_unref(s);
 469 }
 470
 471 static int create_stream(
 472         pa_stream_direction_t direction,
 473         pa_stream *s,
 474         const char *dev,
 475         const pa_buffer_attr *attr,
 476         pa_stream_flags_t flags,
 477         const pa_cvolume *volume,
 478         pa_stream *sync_stream) {
 479
 480     pa_tagstruct *t;
 481     uint32_t tag;
 482
 483     assert(s);
 484     assert(s->ref >= 1);
 485
 486     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
 487     PA_CHECK_VALIDITY(s->context, !(flags & ~((direction != PA_STREAM_UPLOAD ?
 488                                                PA_STREAM_START_CORKED|
 489                                                PA_STREAM_INTERPOLATE_TIMING|
 490                                                PA_STREAM_NOT_MONOTONOUS|
 491                                                PA_STREAM_AUTO_TIMING_UPDATE : 0))), PA_ERR_INVALID);
 492     PA_CHECK_VALIDITY(s->context, !volume || volume->channels == s->sample_spec.channels, PA_ERR_INVALID);
 493     PA_CHECK_VALIDITY(s->context, !sync_stream || (direction == PA_STREAM_PLAYBACK && sync_stream->direction == PA_STREAM_PLAYBACK), PA_ERR_INVALID);
 494
 495     pa_stream_ref(s);
 496
 497     s->direction = direction;
 498     s->flags = flags;
 499
 500     if (sync_stream)
 501         s->syncid = sync_stream->syncid;
 502
 503     if (attr)
 504         s->buffer_attr = *attr;
 505     else {
 506         /* half a second */
 507         s->buffer_attr.tlength = pa_bytes_per_second(&s->sample_spec)/2;
 508         s->buffer_attr.maxlength = (s->buffer_attr.tlength*3)/2;
 509         s->buffer_attr.minreq = s->buffer_attr.tlength/100;
 510         s->buffer_attr.prebuf = s->buffer_attr.tlength - s->buffer_attr.minreq;
 511         s->buffer_attr.fragsize = s->buffer_attr.tlength/100;
 512     }
 513
 514     if (!dev)
 515         dev = s->direction == PA_STREAM_PLAYBACK ? s->context->conf->default_sink : s->context->conf->default_source;
 516
 517     t = pa_tagstruct_command(
 518             s->context,
 519             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM,
 520             &tag);
 521
 522     pa_tagstruct_put(
 523             t,
 524             PA_TAG_STRING, s->name,
 525             PA_TAG_SAMPLE_SPEC, &s->sample_spec,
 526             PA_TAG_CHANNEL_MAP, &s->channel_map,
 527             PA_TAG_U32, PA_INVALID_INDEX,
 528             PA_TAG_STRING, dev,
 529             PA_TAG_U32, s->buffer_attr.maxlength,
 530             PA_TAG_BOOLEAN, !!(flags & PA_STREAM_START_CORKED),
 531             PA_TAG_INVALID);
 532
 533     if (s->direction == PA_STREAM_PLAYBACK) {
 534         pa_cvolume cv;
 535
 536         pa_tagstruct_put(
 537                 t,
 538                 PA_TAG_U32, s->buffer_attr.tlength,
 539                 PA_TAG_U32, s->buffer_attr.prebuf,
 540                 PA_TAG_U32, s->buffer_attr.minreq,
 541                 PA_TAG_U32, s->syncid,
 542                 PA_TAG_INVALID);
 543
 544         if (!volume)
 545             volume = pa_cvolume_reset(&cv, s->sample_spec.channels);
 546
 547         pa_tagstruct_put_cvolume(t, volume);
 548     } else
 549         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
 550
 551     pa_pstream_send_tagstruct(s->context->pstream, t);
 552     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s, NULL);
 553
 554     pa_stream_set_state(s, PA_STREAM_CREATING);
 555
 556     pa_stream_unref(s);
 557     return 0;
 558 }
 559
 560 int pa_stream_connect_playback(
 561         pa_stream *s,
 562         const char *dev,
 563         const pa_buffer_attr *attr,
 564         pa_stream_flags_t flags,
 565         pa_cvolume *volume,
 566         pa_stream *sync_stream) {
 567
 568     assert(s);
 569     assert(s->ref >= 1);
 570
 571     return create_stream(PA_STREAM_PLAYBACK, s, dev, attr, flags, volume, sync_stream);
 572 }
 573
 574 int pa_stream_connect_record(
 575         pa_stream *s,
 576         const char *dev,
 577         const pa_buffer_attr *attr,
 578         pa_stream_flags_t flags) {
 579
 580     assert(s);
 581     assert(s->ref >= 1);
 582
 583     return create_stream(PA_STREAM_RECORD, s, dev, attr, flags, NULL, NULL);
 584 }
 585
 586 int pa_stream_write(
 587         pa_stream *s,
 588         const void *data,
 589         size_t length,
 590         void (*free_cb)(void *p),
 591         int64_t offset,
 592         pa_seek_mode_t seek) {
 593
 594     pa_memchunk chunk;
 595
 596     assert(s);
 597     assert(s->ref >= 1);
 598     assert(data);
 599
 600     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 601     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
 602     PA_CHECK_VALIDITY(s->context, seek <= PA_SEEK_RELATIVE_END, PA_ERR_INVALID);
 603     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || (seek == PA_SEEK_RELATIVE && offset == 0), PA_ERR_INVALID);
 604
 605     if (length <= 0)
 606         return 0;
 607
 608     if (free_cb)
 609         chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) data, length, free_cb, 1);
 610     else {
 611         chunk.memblock = pa_memblock_new(s->context->mempool, length);
 612         memcpy(chunk.memblock->data, data, length);
 613     }
 614
 615     chunk.index = 0;
 616     chunk.length = length;
 617
 618     pa_pstream_send_memblock(s->context->pstream, s->channel, offset, seek, &chunk);
 619     pa_memblock_unref(chunk.memblock);
 620
 621     if (length < s->requested_bytes)
 622         s->requested_bytes -= length;
 623     else
 624         s->requested_bytes = 0;
 625
 626     if (s->direction == PA_STREAM_PLAYBACK) {
 627
 628         /* Update latency request correction */
 629         if (s->write_index_corrections[s->current_write_index_correction].valid) {
 630
 631             if (seek == PA_SEEK_ABSOLUTE) {
 632                 s->write_index_corrections[s->current_write_index_correction].corrupt = 0;
 633                 s->write_index_corrections[s->current_write_index_correction].absolute = 1;
 634                 s->write_index_corrections[s->current_write_index_correction].value = offset + length;
 635             } else if (seek == PA_SEEK_RELATIVE) {
 636                 if (!s->write_index_corrections[s->current_write_index_correction].corrupt)
 637                     s->write_index_corrections[s->current_write_index_correction].value += offset + length;
 638             } else
 639                 s->write_index_corrections[s->current_write_index_correction].corrupt = 1;
 640         }
 641
 642         /* Update the write index in the already available latency data */
 643         if (s->timing_info_valid) {
 644
 645             if (seek == PA_SEEK_ABSOLUTE) {
 646                 s->timing_info.write_index_corrupt = 0;
 647                 s->timing_info.write_index = offset + length;
 648             } else if (seek == PA_SEEK_RELATIVE) {
 649                 if (!s->timing_info.write_index_corrupt)
 650                     s->timing_info.write_index += offset + length;
 651             } else
 652                 s->timing_info.write_index_corrupt = 1;
 653         }
 654
 655         if (!s->timing_info_valid || s->timing_info.write_index_corrupt)
 656             request_auto_timing_update(s, 1);
 657     }
 658
 659     return 0;
 660 }
 661
 662 int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
 663     assert(s);
 664     assert(s->ref >= 1);
 665     assert(data);
 666     assert(length);
 667
 668     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 669     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
 670
 671     if (!s->peek_memchunk.memblock) {
 672
 673         if (pa_memblockq_peek(s->record_memblockq, &s->peek_memchunk) < 0) {
 674             *data = NULL;
 675             *length = 0;
 676             return 0;
 677         }
 678     }
 679
 680     *data = (const char*) s->peek_memchunk.memblock->data + s->peek_memchunk.index;
 681     *length = s->peek_memchunk.length;
 682     return 0;
 683 }
 684
 685 int pa_stream_drop(pa_stream *s) {
 686     assert(s);
 687     assert(s->ref >= 1);
 688
 689     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 690     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
 691     PA_CHECK_VALIDITY(s->context, s->peek_memchunk.memblock, PA_ERR_BADSTATE);
 692
 693     pa_memblockq_drop(s->record_memblockq, &s->peek_memchunk, s->peek_memchunk.length);
 694
 695     /* Fix the simulated local read index */
 696     if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
 697         s->timing_info.read_index += s->peek_memchunk.length;
 698
 699     pa_memblock_unref(s->peek_memchunk.memblock);
 700     s->peek_memchunk.length = 0;
 701     s->peek_memchunk.index = 0;
 702     s->peek_memchunk.memblock = NULL;
 703
 704     return 0;
 705 }
 706
 707 size_t pa_stream_writable_size(pa_stream *s) {
 708     assert(s);
 709     assert(s->ref >= 1);
 710
 711     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
 712     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
 713
 714     return s->requested_bytes;
 715 }
 716
 717 size_t pa_stream_readable_size(pa_stream *s) {
 718     assert(s);
 719     assert(s->ref >= 1);
 720
 721     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
 722     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
 723
 724     return pa_memblockq_get_length(s->record_memblockq);
 725 }
 726
 727 pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
 728     pa_operation *o;
 729     pa_tagstruct *t;
 730     uint32_t tag;
 731
 732     assert(s);
 733     assert(s->ref >= 1);
 734
 735     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 736     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
 737
 738     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
 739
 740     t = pa_tagstruct_command(s->context, PA_COMMAND_DRAIN_PLAYBACK_STREAM, &tag);
 741     pa_tagstruct_putu32(t, s->channel);
 742     pa_pstream_send_tagstruct(s->context->pstream, t);
 743     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);
 744
 745     return o;
 746 }
 747
 748 static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
 749     pa_operation *o = userdata;
 750     struct timeval local, remote, now;
 751     pa_timing_info *i;
 752
 753     assert(pd);
 754     assert(o);
 755
 756     if (!o->context || !o->stream)
 757         goto finish;
 758
 759     i = &o->stream->timing_info;
 760
 761 /*     pa_log("pre corrupt w:%u r:%u\n", !o->stream->timing_info_valid || i->write_index_corrupt,!o->stream->timing_info_valid || i->read_index_corrupt); */
 762
 763     o->stream->timing_info_valid = 0;
 764     i->write_index_corrupt = 0;
 765     i->read_index_corrupt = 0;
 766
 767 /*     pa_log("timing update %u\n", tag); */
 768
 769     if (command != PA_COMMAND_REPLY) {
 770         if (pa_context_handle_error(o->context, command, t) < 0)
 771             goto finish;
 772
 773     } else if (pa_tagstruct_get_usec(t, &i->sink_usec) < 0 ||
 774                pa_tagstruct_get_usec(t, &i->source_usec) < 0 ||
 775                pa_tagstruct_get_boolean(t, &i->playing) < 0 ||
 776                pa_tagstruct_get_timeval(t, &local) < 0 ||
 777                pa_tagstruct_get_timeval(t, &remote) < 0 ||
 778                pa_tagstruct_gets64(t, &i->write_index) < 0 ||
 779                pa_tagstruct_gets64(t, &i->read_index) < 0 ||
 780                !pa_tagstruct_eof(t)) {
 781         pa_context_fail(o->context, PA_ERR_PROTOCOL);
 782         goto finish;
 783
 784     } else {
 785         o->stream->timing_info_valid = 1;
 786
 787         pa_gettimeofday(&now);
 788
 789         /* Calculcate timestamps */
 790         if (pa_timeval_cmp(&local, &remote) <= 0 && pa_timeval_cmp(&remote, &now) <= 0) {
 791             /* local and remote seem to have synchronized clocks */
 792
 793             if (o->stream->direction == PA_STREAM_PLAYBACK)
 794                 i->transport_usec = pa_timeval_diff(&remote, &local);
 795             else
 796                 i->transport_usec = pa_timeval_diff(&now, &remote);
 797
 798             i->synchronized_clocks = 1;
 799             i->timestamp = remote;
 800         } else {
 801             /* clocks are not synchronized, let's estimate latency then */
 802             i->transport_usec = pa_timeval_diff(&now, &local)/2;
 803             i->synchronized_clocks = 0;
 804             i->timestamp = local;
 805             pa_timeval_add(&i->timestamp, i->transport_usec);
 806         }
 807
 808         /* Invalidate read and write indexes if necessary */
 809         if (tag < o->stream->read_index_not_before)
 810             i->read_index_corrupt = 1;
 811
 812         if (tag < o->stream->write_index_not_before)
 813             i->write_index_corrupt = 1;
 814
 815         if (o->stream->direction == PA_STREAM_PLAYBACK) {
 816             /* Write index correction */
 817
 818             int n, j;
 819             uint32_t ctag = tag;
 820
 821             /* Go through the saved correction values and add up the total correction.*/
 822
 823             for (n = 0, j = o->stream->current_write_index_correction+1;
 824                  n < PA_MAX_WRITE_INDEX_CORRECTIONS;
 825                  n++, j = (j + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS) {
 826
 827                 /* Step over invalid data or out-of-date data */
 828                 if (!o->stream->write_index_corrections[j].valid ||
 829                     o->stream->write_index_corrections[j].tag < ctag)
 830                     continue;
 831
 832                 /* Make sure that everything is in order */
 833                 ctag = o->stream->write_index_corrections[j].tag+1;
 834
 835                 /* Now fix the write index */
 836                 if (o->stream->write_index_corrections[j].corrupt) {
 837                     /* A corrupting seek was made */
 838                     i->write_index = 0;
 839                     i->write_index_corrupt = 1;
 840                 } else if (o->stream->write_index_corrections[j].absolute) {
 841                     /* An absolute seek was made */
 842                     i->write_index = o->stream->write_index_corrections[j].value;
 843                     i->write_index_corrupt = 0;
 844                 } else if (!i->write_index_corrupt) {
 845                     /* A relative seek was made */
 846                     i->write_index += o->stream->write_index_corrections[j].value;
 847                 }
 848             }
 849         }
 850
 851         if (o->stream->direction == PA_STREAM_RECORD) {
 852             /* Read index correction */
 853
 854             if (!i->read_index_corrupt)
 855                 i->read_index -= pa_memblockq_get_length(o->stream->record_memblockq);
 856         }
 857
 858         o->stream->cached_time_valid = 0;
 859     }
 860
 861     o->stream->auto_timing_update_requested = 0;
 862 /*     pa_log("post corrupt w:%u r:%u\n", i->write_index_corrupt || !o->stream->timing_info_valid, i->read_index_corrupt || !o->stream->timing_info_valid); */
 863
 864     /* Clear old correction entries */
 865     if (o->stream->direction == PA_STREAM_PLAYBACK) {
 866         int n;
 867
 868         for (n = 0; n < PA_MAX_WRITE_INDEX_CORRECTIONS; n++) {
 869             if (!o->stream->write_index_corrections[n].valid)
 870                 continue;
 871
 872             if (o->stream->write_index_corrections[n].tag <= tag)
 873                 o->stream->write_index_corrections[n].valid = 0;
 874         }
 875     }
 876
 877     if (o->stream->latency_update_callback)
 878         o->stream->latency_update_callback(o->stream, o->stream->latency_update_userdata);
 879
 880     if (o->callback && o->stream && o->stream->state == PA_STREAM_READY) {
 881         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
 882         cb(o->stream, o->stream->timing_info_valid, o->userdata);
 883     }
 884
 885 finish:
 886
 887     pa_operation_done(o);
 888     pa_operation_unref(o);
 889 }
 890
 891 pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
 892     uint32_t tag;
 893     pa_operation *o;
 894     pa_tagstruct *t;
 895     struct timeval now;
 896     int cidx = 0;
 897
 898     assert(s);
 899     assert(s->ref >= 1);
 900
 901     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 902     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
 903
 904     if (s->direction == PA_STREAM_PLAYBACK) {
 905         /* Find a place to store the write_index correction data for this entry */
 906         cidx = (s->current_write_index_correction + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS;
 907
 908         /* Check if we could allocate a correction slot. If not, there are too many outstanding queries */
 909         PA_CHECK_VALIDITY_RETURN_NULL(s->context, !s->write_index_corrections[cidx].valid, PA_ERR_INTERNAL);
 910     }
 911     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
 912
 913     t = pa_tagstruct_command(
 914             s->context,
 915             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY,
 916             &tag);
 917     pa_tagstruct_putu32(t, s->channel);
 918     pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
 919
 920     pa_pstream_send_tagstruct(s->context->pstream, t);
 921     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);
 922
 923     if (s->direction == PA_STREAM_PLAYBACK) {
 924         /* Fill in initial correction data */
 925         o->stream->current_write_index_correction = cidx;
 926         o->stream->write_index_corrections[cidx].valid = 1;
 927         o->stream->write_index_corrections[cidx].tag = tag;
 928         o->stream->write_index_corrections[cidx].absolute = 0;
 929         o->stream->write_index_corrections[cidx].value = 0;
 930         o->stream->write_index_corrections[cidx].corrupt = 0;
 931     }
 932
 933 /*     pa_log("requesting update %u\n", tag); */
 934
 935     return o;
 936 }
 937
 938 void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 939     pa_stream *s = userdata;
 940
 941     assert(pd);
 942     assert(s);
 943     assert(s->ref >= 1);
 944
 945     pa_stream_ref(s);
 946
 947     if (command != PA_COMMAND_REPLY) {
 948         if (pa_context_handle_error(s->context, command, t) < 0)
 949             goto finish;
 950
 951         pa_stream_set_state(s, PA_STREAM_FAILED);
 952         goto finish;
 953     } else if (!pa_tagstruct_eof(t)) {
 954         pa_context_fail(s->context, PA_ERR_PROTOCOL);
 955         goto finish;
 956     }
 957
 958     pa_stream_set_state(s, PA_STREAM_TERMINATED);
 959
 960 finish:
 961     pa_stream_unref(s);
 962 }
 963
 964 int pa_stream_disconnect(pa_stream *s) {
 965     pa_tagstruct *t;
 966     uint32_t tag;
 967
 968     assert(s);
 969     assert(s->ref >= 1);
 970
 971     PA_CHECK_VALIDITY(s->context, s->channel_valid, PA_ERR_BADSTATE);
 972     PA_CHECK_VALIDITY(s->context, s->context->state == PA_CONTEXT_READY, PA_ERR_BADSTATE);
 973
 974     pa_stream_ref(s);
 975
 976     t = pa_tagstruct_command(
 977             s->context,
 978             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM :
 979             (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM),
 980             &tag);
 981     pa_tagstruct_putu32(t, s->channel);
 982     pa_pstream_send_tagstruct(s->context->pstream, t);
 983     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s, NULL);
 984
 985     pa_stream_unref(s);
 986     return 0;
 987 }
 988
 989 void pa_stream_set_read_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
 990     assert(s);
 991     assert(s->ref >= 1);
 992
 993     s->read_callback = cb;
 994     s->read_userdata = userdata;
 995 }
 996
 997 void pa_stream_set_write_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
 998     assert(s);
 999     assert(s->ref >= 1);
1000
1001     s->write_callback = cb;
1002     s->write_userdata = userdata;
1003 }
1004
1005 void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1006     assert(s);
1007     assert(s->ref >= 1);
1008
1009     s->state_callback = cb;
1010     s->state_userdata = userdata;
1011 }
1012
1013 void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1014     assert(s);
1015     assert(s->ref >= 1);
1016
1017     s->overflow_callback = cb;
1018     s->overflow_userdata = userdata;
1019 }
1020
1021 void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1022     assert(s);
1023     assert(s->ref >= 1);
1024
1025     s->underflow_callback = cb;
1026     s->underflow_userdata = userdata;
1027 }
1028
1029 void pa_stream_set_latency_update_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1030     assert(s);
1031     assert(s->ref >= 1);
1032
1033     s->latency_update_callback = cb;
1034     s->latency_update_userdata = userdata;
1035 }
1036
1037 void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
1038     pa_operation *o = userdata;
1039     int success = 1;
1040
1041     assert(pd);
1042     assert(o);
1043     assert(o->ref >= 1);
1044
1045     if (!o->context)
1046         goto finish;
1047
1048     if (command != PA_COMMAND_REPLY) {
1049         if (pa_context_handle_error(o->context, command, t) < 0)
1050             goto finish;
1051
1052         success = 0;
1053     } else if (!pa_tagstruct_eof(t)) {
1054         pa_context_fail(o->context, PA_ERR_PROTOCOL);
1055         goto finish;
1056     }
1057
1058     if (o->callback) {
1059         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
1060         cb(o->stream, success, o->userdata);
1061     }
1062
1063 finish:
1064     pa_operation_done(o);
1065     pa_operation_unref(o);
1066 }
1067
1068 pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, void *userdata) {
1069     pa_operation *o;
1070     pa_tagstruct *t;
1071     uint32_t tag;
1072
1073     assert(s);
1074     assert(s->ref >= 1);
1075
1076     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1077     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1078
1079     s->corked = b;
1080
1081     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1082
1083     t = pa_tagstruct_command(
1084             s->context,
1085             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CORK_PLAYBACK_STREAM : PA_COMMAND_CORK_RECORD_STREAM,
1086             &tag);
1087     pa_tagstruct_putu32(t, s->channel);
1088     pa_tagstruct_put_boolean(t, !!b);
1089     pa_pstream_send_tagstruct(s->context->pstream, t);
1090     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);
1091
1092     if (s->direction == PA_STREAM_PLAYBACK)
1093         invalidate_indexes(s, 1, 0);
1094
1095     return o;
1096 }
1097
1098 static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command, pa_stream_success_cb_t cb, void *userdata) {
1099     pa_tagstruct *t;
1100     pa_operation *o;
1101     uint32_t tag;
1102
1103     assert(s);
1104     assert(s->ref >= 1);
1105
1106     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1107
1108     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1109
1110     t = pa_tagstruct_command(s->context, command, &tag);
1111     pa_tagstruct_putu32(t, s->channel);
1112     pa_pstream_send_tagstruct(s->context->pstream, t);
1113     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);
1114
1115     return o;
1116 }
1117
1118 pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1119     pa_operation *o;
1120
1121     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1122
1123     if ((o = stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata))) {
1124
1125         if (s->direction == PA_STREAM_PLAYBACK) {
1126             if (s->write_index_corrections[s->current_write_index_correction].valid)
1127                 s->write_index_corrections[s->current_write_index_correction].corrupt = 1;
1128
1129             if (s->timing_info_valid)
1130                 s->timing_info.write_index_corrupt = 1;
1131
1132             if (s->buffer_attr.prebuf > 0)
1133                 invalidate_indexes(s, 1, 0);
1134             else
1135                 request_auto_timing_update(s, 1);
1136         } else
1137             invalidate_indexes(s, 0, 1);
1138     }
1139
1140     return o;
1141 }
1142
1143 pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1144     pa_operation *o;
1145
1146     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1147     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
1148
1149     if ((o = stream_send_simple_command(s, PA_COMMAND_PREBUF_PLAYBACK_STREAM, cb, userdata)))
1150         invalidate_indexes(s, 1, 0);
1151
1152     return o;
1153 }
1154
1155 pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1156     pa_operation *o;
1157
1158     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1159     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
1160
1161     if ((o = stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata)))
1162         invalidate_indexes(s, 1, 0);
1163
1164     return o;
1165 }
1166
1167 pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_success_cb_t cb, void *userdata) {
1168     pa_operation *o;
1169     pa_tagstruct *t;
1170     uint32_t tag;
1171
1172     assert(s);
1173     assert(s->ref >= 1);
1174     assert(name);
1175
1176     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1177     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1178
1179     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1180
1181     t = pa_tagstruct_command(
1182             s->context,
1183             s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_NAME : PA_COMMAND_SET_PLAYBACK_STREAM_NAME,
1184             &tag);
1185     pa_tagstruct_putu32(t, s->channel);
1186     pa_tagstruct_puts(t, name);
1187     pa_pstream_send_tagstruct(s->context->pstream, t);
1188     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);
1189
1190     return o;
1191 }
1192
1193 int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
1194     pa_usec_t usec = 0;
1195
1196     assert(s);
1197     assert(s->ref >= 1);
1198
1199     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1200     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1201     PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
1202     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
1203     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
1204
1205     if (s->cached_time_valid)
1206         /* We alredy calculated the time value for this timing info, so let's reuse it */
1207         usec = s->cached_time;
1208     else {
1209         if (s->direction == PA_STREAM_PLAYBACK) {
1210             /* The last byte that was written into the output device
1211              * had this time value associated */
1212             usec = pa_bytes_to_usec(s->timing_info.read_index < 0 ? 0 : (uint64_t) s->timing_info.read_index, &s->sample_spec);
1213
1214             if (!s->corked) {
1215                 /* Because the latency info took a little time to come
1216                  * to us, we assume that the real output time is actually
1217                  * a little ahead */
1218                 usec += s->timing_info.transport_usec;
1219
1220                 /* However, the output device usually maintains a buffer
1221                    too, hence the real sample currently played is a little
1222                    back  */
1223                 if (s->timing_info.sink_usec >= usec)
1224                     usec = 0;
1225                 else
1226                     usec -= s->timing_info.sink_usec;
1227             }
1228
1229         } else if (s->direction == PA_STREAM_RECORD) {
1230             /* The last byte written into the server side queue had
1231              * this time value associated */
1232             usec = pa_bytes_to_usec(s->timing_info.write_index < 0 ? 0 : (uint64_t) s->timing_info.write_index, &s->sample_spec);
1233
1234             if (!s->corked) {
1235                 /* Add transport latency */
1236                 usec += s->timing_info.transport_usec;
1237
1238                 /* Add latency of data in device buffer */
1239                 usec += s->timing_info.source_usec;
1240
1241                 /* If this is a monitor source, we need to correct the
1242                  * time by the playback device buffer */
1243                 if (s->timing_info.sink_usec >= usec)
1244                     usec = 0;
1245                 else
1246                     usec -= s->timing_info.sink_usec;
1247             }
1248         }
1249
1250         s->cached_time = usec;
1251         s->cached_time_valid = 1;
1252     }
1253
1254     /* Interpolate if requested */
1255     if (s->flags & PA_STREAM_INTERPOLATE_TIMING) {
1256
1257         /* We just add the time that passed since the latency info was
1258          * current */
1259         if (!s->corked && s->timing_info.playing) {
1260             struct timeval now;
1261             usec += pa_timeval_diff(pa_gettimeofday(&now), &s->timing_info.timestamp);
1262         }
1263     }
1264
1265     /* Make sure the time runs monotonically */
1266     if (!(s->flags & PA_STREAM_NOT_MONOTONOUS)) {
1267         if (usec < s->previous_time)
1268             usec = s->previous_time;
1269         else
1270             s->previous_time = usec;
1271     }
1272
1273     if (r_usec)
1274         *r_usec = usec;
1275
1276     return 0;
1277 }
1278
1279 static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t a, pa_usec_t b, int *negative) {
1280     assert(s);
1281     assert(s->ref >= 1);
1282
1283     if (negative)
1284         *negative = 0;
1285
1286     if (a >= b)
1287         return a-b;
1288     else {
1289         if (negative && s->direction == PA_STREAM_RECORD) {
1290             *negative = 1;
1291             return b-a;
1292         } else
1293             return 0;
1294     }
1295 }
1296
1297 int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
1298     pa_usec_t t, c;
1299     int r;
1300     int64_t cindex;
1301
1302     assert(s);
1303     assert(s->ref >= 1);
1304     assert(r_usec);
1305
1306     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1307     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1308     PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
1309     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
1310     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
1311
1312     if ((r = pa_stream_get_time(s, &t)) < 0)
1313         return r;
1314
1315     if (s->direction == PA_STREAM_PLAYBACK)
1316         cindex = s->timing_info.write_index;
1317     else
1318         cindex = s->timing_info.read_index;
1319
1320     if (cindex < 0)
1321         cindex = 0;
1322
1323     c = pa_bytes_to_usec(cindex, &s->sample_spec);
1324
1325     if (s->direction == PA_STREAM_PLAYBACK)
1326         *r_usec = time_counter_diff(s, c, t, negative);
1327     else
1328         *r_usec = time_counter_diff(s, t, c, negative);
1329
1330     return 0;
1331 }
1332
1333 const pa_timing_info* pa_stream_get_timing_info(pa_stream *s) {
1334     assert(s);
1335     assert(s->ref >= 1);
1336
1337     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1338     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1339     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->timing_info_valid, PA_ERR_BADSTATE);
1340
1341     return &s->timing_info;
1342 }
1343
1344 const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s) {
1345     assert(s);
1346     assert(s->ref >= 1);
1347
1348     return &s->sample_spec;
1349 }
1350
1351 const pa_channel_map* pa_stream_get_channel_map(pa_stream *s) {
1352     assert(s);
1353     assert(s->ref >= 1);
1354
1355     return &s->channel_map;
1356 }
1357
1358 const pa_buffer_attr* pa_stream_get_buffer_attr(pa_stream *s) {
1359     assert(s);
1360     assert(s->ref >= 1);
1361
1362     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1363     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1364     PA_CHECK_VALIDITY_RETURN_NULL(s->context,
1365         pa_context_get_server_protocol_version(s->context) >= 9, PA_ERR_NODATA);
1366
1367     return &s->buffer_attr;
1368 }