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

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