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