src/pulse/stream.c

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