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