code.delx.au - pulseaudio/blob - src/polyp/stream.c

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