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     }
 442
 443     s->channel_valid = 1;
 444     pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
 445
 446     pa_stream_set_state(s, PA_STREAM_READY);
 447
 448     if (s->direction != PA_STREAM_UPLOAD &&
 449         s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
 450         struct timeval tv;
 451
 452         pa_gettimeofday(&tv);
 453         tv.tv_usec += LATENCY_IPOL_INTERVAL_USEC; /* every 100 ms */
 454
 455         assert(!s->auto_timing_update_event);
 456         s->auto_timing_update_event = s->mainloop->time_new(s->mainloop, &tv, &auto_timing_update_callback, s);
 457
 458         request_auto_timing_update(s, 1);
 459     }
 460
 461     if (s->requested_bytes > 0 && s->ref > 1 && s->write_callback)
 462         s->write_callback(s, s->requested_bytes, s->write_userdata);
 463
 464 finish:
 465     pa_stream_unref(s);
 466 }
 467
 468 static int create_stream(
 469         pa_stream_direction_t direction,
 470         pa_stream *s,
 471         const char *dev,
 472         const pa_buffer_attr *attr,
 473         pa_stream_flags_t flags,
 474         const pa_cvolume *volume,
 475         pa_stream *sync_stream) {
 476
 477     pa_tagstruct *t;
 478     uint32_t tag;
 479
 480     assert(s);
 481     assert(s->ref >= 1);
 482
 483     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
 484     PA_CHECK_VALIDITY(s->context, !(flags & ~((direction != PA_STREAM_UPLOAD ?
 485                                                PA_STREAM_START_CORKED|
 486                                                PA_STREAM_INTERPOLATE_TIMING|
 487                                                PA_STREAM_NOT_MONOTONOUS|
 488                                                PA_STREAM_AUTO_TIMING_UPDATE : 0))), PA_ERR_INVALID);
 489     PA_CHECK_VALIDITY(s->context, !volume || volume->channels == s->sample_spec.channels, PA_ERR_INVALID);
 490     PA_CHECK_VALIDITY(s->context, !sync_stream || (direction == PA_STREAM_PLAYBACK && sync_stream->direction == PA_STREAM_PLAYBACK), PA_ERR_INVALID);
 491
 492     pa_stream_ref(s);
 493
 494     s->direction = direction;
 495     s->flags = flags;
 496
 497     if (sync_stream)
 498         s->syncid = sync_stream->syncid;
 499
 500     if (attr)
 501         s->buffer_attr = *attr;
 502     else {
 503         /* half a second */
 504         s->buffer_attr.tlength = pa_bytes_per_second(&s->sample_spec)/2;
 505         s->buffer_attr.maxlength = (s->buffer_attr.tlength*3)/2;
 506         s->buffer_attr.minreq = s->buffer_attr.tlength/100;
 507         s->buffer_attr.prebuf = s->buffer_attr.tlength - s->buffer_attr.minreq;
 508         s->buffer_attr.fragsize = s->buffer_attr.tlength/100;
 509     }
 510
 511     if (!dev)
 512         dev = s->direction == PA_STREAM_PLAYBACK ? s->context->conf->default_sink : s->context->conf->default_source;
 513
 514     t = pa_tagstruct_command(
 515             s->context,
 516             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM,
 517             &tag);
 518
 519     pa_tagstruct_put(
 520             t,
 521             PA_TAG_STRING, s->name,
 522             PA_TAG_SAMPLE_SPEC, &s->sample_spec,
 523             PA_TAG_CHANNEL_MAP, &s->channel_map,
 524             PA_TAG_U32, PA_INVALID_INDEX,
 525             PA_TAG_STRING, dev,
 526             PA_TAG_U32, s->buffer_attr.maxlength,
 527             PA_TAG_BOOLEAN, !!(flags & PA_STREAM_START_CORKED),
 528             PA_TAG_INVALID);
 529
 530     if (s->direction == PA_STREAM_PLAYBACK) {
 531         pa_cvolume cv;
 532
 533         pa_tagstruct_put(
 534                 t,
 535                 PA_TAG_U32, s->buffer_attr.tlength,
 536                 PA_TAG_U32, s->buffer_attr.prebuf,
 537                 PA_TAG_U32, s->buffer_attr.minreq,
 538                 PA_TAG_U32, s->syncid,
 539                 PA_TAG_INVALID);
 540
 541         if (!volume)
 542             volume = pa_cvolume_reset(&cv, s->sample_spec.channels);
 543
 544         pa_tagstruct_put_cvolume(t, volume);
 545     } else
 546         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
 547
 548     pa_pstream_send_tagstruct(s->context->pstream, t);
 549     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s, NULL);
 550
 551     pa_stream_set_state(s, PA_STREAM_CREATING);
 552
 553     pa_stream_unref(s);
 554     return 0;
 555 }
 556
 557 int pa_stream_connect_playback(
 558         pa_stream *s,
 559         const char *dev,
 560         const pa_buffer_attr *attr,
 561         pa_stream_flags_t flags,
 562         pa_cvolume *volume,
 563         pa_stream *sync_stream) {
 564
 565     assert(s);
 566     assert(s->ref >= 1);
 567
 568     return create_stream(PA_STREAM_PLAYBACK, s, dev, attr, flags, volume, sync_stream);
 569 }
 570
 571 int pa_stream_connect_record(
 572         pa_stream *s,
 573         const char *dev,
 574         const pa_buffer_attr *attr,
 575         pa_stream_flags_t flags) {
 576
 577     assert(s);
 578     assert(s->ref >= 1);
 579
 580     return create_stream(PA_STREAM_RECORD, s, dev, attr, flags, NULL, NULL);
 581 }
 582
 583 int pa_stream_write(
 584         pa_stream *s,
 585         const void *data,
 586         size_t length,
 587         void (*free_cb)(void *p),
 588         int64_t offset,
 589         pa_seek_mode_t seek) {
 590
 591     pa_memchunk chunk;
 592
 593     assert(s);
 594     assert(s->ref >= 1);
 595     assert(data);
 596
 597     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 598     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
 599     PA_CHECK_VALIDITY(s->context, seek <= PA_SEEK_RELATIVE_END, PA_ERR_INVALID);
 600     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || (seek == PA_SEEK_RELATIVE && offset == 0), PA_ERR_INVALID);
 601
 602     if (length <= 0)
 603         return 0;
 604
 605     if (free_cb)
 606         chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) data, length, free_cb, 1);
 607     else {
 608         chunk.memblock = pa_memblock_new(s->context->mempool, length);
 609         memcpy(chunk.memblock->data, data, length);
 610     }
 611
 612     chunk.index = 0;
 613     chunk.length = length;
 614
 615     pa_pstream_send_memblock(s->context->pstream, s->channel, offset, seek, &chunk);
 616     pa_memblock_unref(chunk.memblock);
 617
 618     if (length < s->requested_bytes)
 619         s->requested_bytes -= length;
 620     else
 621         s->requested_bytes = 0;
 622
 623     if (s->direction == PA_STREAM_PLAYBACK) {
 624
 625         /* Update latency request correction */
 626         if (s->write_index_corrections[s->current_write_index_correction].valid) {
 627
 628             if (seek == PA_SEEK_ABSOLUTE) {
 629                 s->write_index_corrections[s->current_write_index_correction].corrupt = 0;
 630                 s->write_index_corrections[s->current_write_index_correction].absolute = 1;
 631                 s->write_index_corrections[s->current_write_index_correction].value = offset + length;
 632             } else if (seek == PA_SEEK_RELATIVE) {
 633                 if (!s->write_index_corrections[s->current_write_index_correction].corrupt)
 634                     s->write_index_corrections[s->current_write_index_correction].value += offset + length;
 635             } else
 636                 s->write_index_corrections[s->current_write_index_correction].corrupt = 1;
 637         }
 638
 639         /* Update the write index in the already available latency data */
 640         if (s->timing_info_valid) {
 641
 642             if (seek == PA_SEEK_ABSOLUTE) {
 643                 s->timing_info.write_index_corrupt = 0;
 644                 s->timing_info.write_index = offset + length;
 645             } else if (seek == PA_SEEK_RELATIVE) {
 646                 if (!s->timing_info.write_index_corrupt)
 647                     s->timing_info.write_index += offset + length;
 648             } else
 649                 s->timing_info.write_index_corrupt = 1;
 650         }
 651
 652         if (!s->timing_info_valid || s->timing_info.write_index_corrupt)
 653             request_auto_timing_update(s, 1);
 654     }
 655
 656     return 0;
 657 }
 658
 659 int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
 660     assert(s);
 661     assert(s->ref >= 1);
 662     assert(data);
 663     assert(length);
 664
 665     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 666     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
 667
 668     if (!s->peek_memchunk.memblock) {
 669
 670         if (pa_memblockq_peek(s->record_memblockq, &s->peek_memchunk) < 0) {
 671             *data = NULL;
 672             *length = 0;
 673             return 0;
 674         }
 675     }
 676
 677     *data = (const char*) s->peek_memchunk.memblock->data + s->peek_memchunk.index;
 678     *length = s->peek_memchunk.length;
 679     return 0;
 680 }
 681
 682 int pa_stream_drop(pa_stream *s) {
 683     assert(s);
 684     assert(s->ref >= 1);
 685
 686     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 687     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
 688     PA_CHECK_VALIDITY(s->context, s->peek_memchunk.memblock, PA_ERR_BADSTATE);
 689
 690     pa_memblockq_drop(s->record_memblockq, &s->peek_memchunk, s->peek_memchunk.length);
 691
 692     /* Fix the simulated local read index */
 693     if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
 694         s->timing_info.read_index += s->peek_memchunk.length;
 695
 696     pa_memblock_unref(s->peek_memchunk.memblock);
 697     s->peek_memchunk.length = 0;
 698     s->peek_memchunk.index = 0;
 699     s->peek_memchunk.memblock = NULL;
 700
 701     return 0;
 702 }
 703
 704 size_t pa_stream_writable_size(pa_stream *s) {
 705     assert(s);
 706     assert(s->ref >= 1);
 707
 708     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
 709     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
 710
 711     return s->requested_bytes;
 712 }
 713
 714 size_t pa_stream_readable_size(pa_stream *s) {
 715     assert(s);
 716     assert(s->ref >= 1);
 717
 718     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
 719     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
 720
 721     return pa_memblockq_get_length(s->record_memblockq);
 722 }
 723
 724 pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
 725     pa_operation *o;
 726     pa_tagstruct *t;
 727     uint32_t tag;
 728
 729     assert(s);
 730     assert(s->ref >= 1);
 731
 732     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 733     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
 734
 735     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
 736
 737     t = pa_tagstruct_command(s->context, PA_COMMAND_DRAIN_PLAYBACK_STREAM, &tag);
 738     pa_tagstruct_putu32(t, s->channel);
 739     pa_pstream_send_tagstruct(s->context->pstream, t);
 740     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);
 741
 742     return o;
 743 }
 744
 745 static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
 746     pa_operation *o = userdata;
 747     struct timeval local, remote, now;
 748     pa_timing_info *i;
 749
 750     assert(pd);
 751     assert(o);
 752
 753     if (!o->context || !o->stream)
 754         goto finish;
 755
 756     i = &o->stream->timing_info;
 757
 758 /*     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); */
 759
 760     o->stream->timing_info_valid = 0;
 761     i->write_index_corrupt = 0;
 762     i->read_index_corrupt = 0;
 763
 764 /*     pa_log("timing update %u\n", tag); */
 765
 766     if (command != PA_COMMAND_REPLY) {
 767         if (pa_context_handle_error(o->context, command, t) < 0)
 768             goto finish;
 769
 770     } else if (pa_tagstruct_get_usec(t, &i->sink_usec) < 0 ||
 771                pa_tagstruct_get_usec(t, &i->source_usec) < 0 ||
 772                pa_tagstruct_get_boolean(t, &i->playing) < 0 ||
 773                pa_tagstruct_get_timeval(t, &local) < 0 ||
 774                pa_tagstruct_get_timeval(t, &remote) < 0 ||
 775                pa_tagstruct_gets64(t, &i->write_index) < 0 ||
 776                pa_tagstruct_gets64(t, &i->read_index) < 0 ||
 777                !pa_tagstruct_eof(t)) {
 778         pa_context_fail(o->context, PA_ERR_PROTOCOL);
 779         goto finish;
 780
 781     } else {
 782         o->stream->timing_info_valid = 1;
 783
 784         pa_gettimeofday(&now);
 785
 786         /* Calculcate timestamps */
 787         if (pa_timeval_cmp(&local, &remote) <= 0 && pa_timeval_cmp(&remote, &now) <= 0) {
 788             /* local and remote seem to have synchronized clocks */
 789
 790             if (o->stream->direction == PA_STREAM_PLAYBACK)
 791                 i->transport_usec = pa_timeval_diff(&remote, &local);
 792             else
 793                 i->transport_usec = pa_timeval_diff(&now, &remote);
 794
 795             i->synchronized_clocks = 1;
 796             i->timestamp = remote;
 797         } else {
 798             /* clocks are not synchronized, let's estimate latency then */
 799             i->transport_usec = pa_timeval_diff(&now, &local)/2;
 800             i->synchronized_clocks = 0;
 801             i->timestamp = local;
 802             pa_timeval_add(&i->timestamp, i->transport_usec);
 803         }
 804
 805         /* Invalidate read and write indexes if necessary */
 806         if (tag < o->stream->read_index_not_before)
 807             i->read_index_corrupt = 1;
 808
 809         if (tag < o->stream->write_index_not_before)
 810             i->write_index_corrupt = 1;
 811
 812         if (o->stream->direction == PA_STREAM_PLAYBACK) {
 813             /* Write index correction */
 814
 815             int n, j;
 816             uint32_t ctag = tag;
 817
 818             /* Go through the saved correction values and add up the total correction.*/
 819
 820             for (n = 0, j = o->stream->current_write_index_correction+1;
 821                  n < PA_MAX_WRITE_INDEX_CORRECTIONS;
 822                  n++, j = (j + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS) {
 823
 824                 /* Step over invalid data or out-of-date data */
 825                 if (!o->stream->write_index_corrections[j].valid ||
 826                     o->stream->write_index_corrections[j].tag < ctag)
 827                     continue;
 828
 829                 /* Make sure that everything is in order */
 830                 ctag = o->stream->write_index_corrections[j].tag+1;
 831
 832                 /* Now fix the write index */
 833                 if (o->stream->write_index_corrections[j].corrupt) {
 834                     /* A corrupting seek was made */
 835                     i->write_index = 0;
 836                     i->write_index_corrupt = 1;
 837                 } else if (o->stream->write_index_corrections[j].absolute) {
 838                     /* An absolute seek was made */
 839                     i->write_index = o->stream->write_index_corrections[j].value;
 840                     i->write_index_corrupt = 0;
 841                 } else if (!i->write_index_corrupt) {
 842                     /* A relative seek was made */
 843                     i->write_index += o->stream->write_index_corrections[j].value;
 844                 }
 845             }
 846         }
 847
 848         if (o->stream->direction == PA_STREAM_RECORD) {
 849             /* Read index correction */
 850
 851             if (!i->read_index_corrupt)
 852                 i->read_index -= pa_memblockq_get_length(o->stream->record_memblockq);
 853         }
 854
 855         o->stream->cached_time_valid = 0;
 856     }
 857
 858     o->stream->auto_timing_update_requested = 0;
 859 /*     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); */
 860
 861     /* Clear old correction entries */
 862     if (o->stream->direction == PA_STREAM_PLAYBACK) {
 863         int n;
 864
 865         for (n = 0; n < PA_MAX_WRITE_INDEX_CORRECTIONS; n++) {
 866             if (!o->stream->write_index_corrections[n].valid)
 867                 continue;
 868
 869             if (o->stream->write_index_corrections[n].tag <= tag)
 870                 o->stream->write_index_corrections[n].valid = 0;
 871         }
 872     }
 873
 874     if (o->stream->latency_update_callback)
 875         o->stream->latency_update_callback(o->stream, o->stream->latency_update_userdata);
 876
 877     if (o->callback && o->stream && o->stream->state == PA_STREAM_READY) {
 878         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
 879         cb(o->stream, o->stream->timing_info_valid, o->userdata);
 880     }
 881
 882 finish:
 883
 884     pa_operation_done(o);
 885     pa_operation_unref(o);
 886 }
 887
 888 pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
 889     uint32_t tag;
 890     pa_operation *o;
 891     pa_tagstruct *t;
 892     struct timeval now;
 893     int cidx = 0;
 894
 895     assert(s);
 896     assert(s->ref >= 1);
 897
 898     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 899     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
 900
 901     if (s->direction == PA_STREAM_PLAYBACK) {
 902         /* Find a place to store the write_index correction data for this entry */
 903         cidx = (s->current_write_index_correction + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS;
 904
 905         /* Check if we could allocate a correction slot. If not, there are too many outstanding queries */
 906         PA_CHECK_VALIDITY_RETURN_NULL(s->context, !s->write_index_corrections[cidx].valid, PA_ERR_INTERNAL);
 907     }
 908     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
 909
 910     t = pa_tagstruct_command(
 911             s->context,
 912             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY,
 913             &tag);
 914     pa_tagstruct_putu32(t, s->channel);
 915     pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
 916
 917     pa_pstream_send_tagstruct(s->context->pstream, t);
 918     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);
 919
 920     if (s->direction == PA_STREAM_PLAYBACK) {
 921         /* Fill in initial correction data */
 922         o->stream->current_write_index_correction = cidx;
 923         o->stream->write_index_corrections[cidx].valid = 1;
 924         o->stream->write_index_corrections[cidx].tag = tag;
 925         o->stream->write_index_corrections[cidx].absolute = 0;
 926         o->stream->write_index_corrections[cidx].value = 0;
 927         o->stream->write_index_corrections[cidx].corrupt = 0;
 928     }
 929
 930 /*     pa_log("requesting update %u\n", tag); */
 931
 932     return o;
 933 }
 934
 935 void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 936     pa_stream *s = userdata;
 937
 938     assert(pd);
 939     assert(s);
 940     assert(s->ref >= 1);
 941
 942     pa_stream_ref(s);
 943
 944     if (command != PA_COMMAND_REPLY) {
 945         if (pa_context_handle_error(s->context, command, t) < 0)
 946             goto finish;
 947
 948         pa_stream_set_state(s, PA_STREAM_FAILED);
 949         goto finish;
 950     } else if (!pa_tagstruct_eof(t)) {
 951         pa_context_fail(s->context, PA_ERR_PROTOCOL);
 952         goto finish;
 953     }
 954
 955     pa_stream_set_state(s, PA_STREAM_TERMINATED);
 956
 957 finish:
 958     pa_stream_unref(s);
 959 }
 960
 961 int pa_stream_disconnect(pa_stream *s) {
 962     pa_tagstruct *t;
 963     uint32_t tag;
 964
 965     assert(s);
 966     assert(s->ref >= 1);
 967
 968     PA_CHECK_VALIDITY(s->context, s->channel_valid, PA_ERR_BADSTATE);
 969     PA_CHECK_VALIDITY(s->context, s->context->state == PA_CONTEXT_READY, PA_ERR_BADSTATE);
 970
 971     pa_stream_ref(s);
 972
 973     t = pa_tagstruct_command(
 974             s->context,
 975             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM :
 976             (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM),
 977             &tag);
 978     pa_tagstruct_putu32(t, s->channel);
 979     pa_pstream_send_tagstruct(s->context->pstream, t);
 980     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s, NULL);
 981
 982     pa_stream_unref(s);
 983     return 0;
 984 }
 985
 986 void pa_stream_set_read_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
 987     assert(s);
 988     assert(s->ref >= 1);
 989
 990     s->read_callback = cb;
 991     s->read_userdata = userdata;
 992 }
 993
 994 void pa_stream_set_write_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
 995     assert(s);
 996     assert(s->ref >= 1);
 997
 998     s->write_callback = cb;
 999     s->write_userdata = userdata;
1000 }
1001
1002 void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1003     assert(s);
1004     assert(s->ref >= 1);
1005
1006     s->state_callback = cb;
1007     s->state_userdata = userdata;
1008 }
1009
1010 void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1011     assert(s);
1012     assert(s->ref >= 1);
1013
1014     s->overflow_callback = cb;
1015     s->overflow_userdata = userdata;
1016 }
1017
1018 void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1019     assert(s);
1020     assert(s->ref >= 1);
1021
1022     s->underflow_callback = cb;
1023     s->underflow_userdata = userdata;
1024 }
1025
1026 void pa_stream_set_latency_update_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1027     assert(s);
1028     assert(s->ref >= 1);
1029
1030     s->latency_update_callback = cb;
1031     s->latency_update_userdata = userdata;
1032 }
1033
1034 void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
1035     pa_operation *o = userdata;
1036     int success = 1;
1037
1038     assert(pd);
1039     assert(o);
1040     assert(o->ref >= 1);
1041
1042     if (!o->context)
1043         goto finish;
1044
1045     if (command != PA_COMMAND_REPLY) {
1046         if (pa_context_handle_error(o->context, command, t) < 0)
1047             goto finish;
1048
1049         success = 0;
1050     } else if (!pa_tagstruct_eof(t)) {
1051         pa_context_fail(o->context, PA_ERR_PROTOCOL);
1052         goto finish;
1053     }
1054
1055     if (o->callback) {
1056         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
1057         cb(o->stream, success, o->userdata);
1058     }
1059
1060 finish:
1061     pa_operation_done(o);
1062     pa_operation_unref(o);
1063 }
1064
1065 pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, void *userdata) {
1066     pa_operation *o;
1067     pa_tagstruct *t;
1068     uint32_t tag;
1069
1070     assert(s);
1071     assert(s->ref >= 1);
1072
1073     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1074     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1075
1076     s->corked = b;
1077
1078     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1079
1080     t = pa_tagstruct_command(
1081             s->context,
1082             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CORK_PLAYBACK_STREAM : PA_COMMAND_CORK_RECORD_STREAM,
1083             &tag);
1084     pa_tagstruct_putu32(t, s->channel);
1085     pa_tagstruct_put_boolean(t, !!b);
1086     pa_pstream_send_tagstruct(s->context->pstream, t);
1087     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);
1088
1089     if (s->direction == PA_STREAM_PLAYBACK)
1090         invalidate_indexes(s, 1, 0);
1091
1092     return o;
1093 }
1094
1095 static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command, pa_stream_success_cb_t cb, void *userdata) {
1096     pa_tagstruct *t;
1097     pa_operation *o;
1098     uint32_t tag;
1099
1100     assert(s);
1101     assert(s->ref >= 1);
1102
1103     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1104
1105     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1106
1107     t = pa_tagstruct_command(s->context, command, &tag);
1108     pa_tagstruct_putu32(t, s->channel);
1109     pa_pstream_send_tagstruct(s->context->pstream, t);
1110     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);
1111
1112     return o;
1113 }
1114
1115 pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1116     pa_operation *o;
1117
1118     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1119
1120     if ((o = stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata))) {
1121
1122         if (s->direction == PA_STREAM_PLAYBACK) {
1123             if (s->write_index_corrections[s->current_write_index_correction].valid)
1124                 s->write_index_corrections[s->current_write_index_correction].corrupt = 1;
1125
1126             if (s->timing_info_valid)
1127                 s->timing_info.write_index_corrupt = 1;
1128
1129             if (s->buffer_attr.prebuf > 0)
1130                 invalidate_indexes(s, 1, 0);
1131             else
1132                 request_auto_timing_update(s, 1);
1133         } else
1134             invalidate_indexes(s, 0, 1);
1135     }
1136
1137     return o;
1138 }
1139
1140 pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1141     pa_operation *o;
1142
1143     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1144     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
1145
1146     if ((o = stream_send_simple_command(s, PA_COMMAND_PREBUF_PLAYBACK_STREAM, cb, userdata)))
1147         invalidate_indexes(s, 1, 0);
1148
1149     return o;
1150 }
1151
1152 pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1153     pa_operation *o;
1154
1155     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1156     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
1157
1158     if ((o = stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata)))
1159         invalidate_indexes(s, 1, 0);
1160
1161     return o;
1162 }
1163
1164 pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_success_cb_t cb, void *userdata) {
1165     pa_operation *o;
1166     pa_tagstruct *t;
1167     uint32_t tag;
1168
1169     assert(s);
1170     assert(s->ref >= 1);
1171     assert(name);
1172
1173     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1174     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1175
1176     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1177
1178     t = pa_tagstruct_command(
1179             s->context,
1180             s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_NAME : PA_COMMAND_SET_PLAYBACK_STREAM_NAME,
1181             &tag);
1182     pa_tagstruct_putu32(t, s->channel);
1183     pa_tagstruct_puts(t, name);
1184     pa_pstream_send_tagstruct(s->context->pstream, t);
1185     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);
1186
1187     return o;
1188 }
1189
1190 int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
1191     pa_usec_t usec = 0;
1192
1193     assert(s);
1194     assert(s->ref >= 1);
1195
1196     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1197     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1198     PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
1199     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
1200     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
1201
1202     if (s->cached_time_valid)
1203         /* We alredy calculated the time value for this timing info, so let's reuse it */
1204         usec = s->cached_time;
1205     else {
1206         if (s->direction == PA_STREAM_PLAYBACK) {
1207             /* The last byte that was written into the output device
1208              * had this time value associated */
1209             usec = pa_bytes_to_usec(s->timing_info.read_index < 0 ? 0 : (uint64_t) s->timing_info.read_index, &s->sample_spec);
1210
1211             if (!s->corked) {
1212                 /* Because the latency info took a little time to come
1213                  * to us, we assume that the real output time is actually
1214                  * a little ahead */
1215                 usec += s->timing_info.transport_usec;
1216
1217                 /* However, the output device usually maintains a buffer
1218                    too, hence the real sample currently played is a little
1219                    back  */
1220                 if (s->timing_info.sink_usec >= usec)
1221                     usec = 0;
1222                 else
1223                     usec -= s->timing_info.sink_usec;
1224             }
1225
1226         } else if (s->direction == PA_STREAM_RECORD) {
1227             /* The last byte written into the server side queue had
1228              * this time value associated */
1229             usec = pa_bytes_to_usec(s->timing_info.write_index < 0 ? 0 : (uint64_t) s->timing_info.write_index, &s->sample_spec);
1230
1231             if (!s->corked) {
1232                 /* Add transport latency */
1233                 usec += s->timing_info.transport_usec;
1234
1235                 /* Add latency of data in device buffer */
1236                 usec += s->timing_info.source_usec;
1237
1238                 /* If this is a monitor source, we need to correct the
1239                  * time by the playback device buffer */
1240                 if (s->timing_info.sink_usec >= usec)
1241                     usec = 0;
1242                 else
1243                     usec -= s->timing_info.sink_usec;
1244             }
1245         }
1246
1247         s->cached_time = usec;
1248         s->cached_time_valid = 1;
1249     }
1250
1251     /* Interpolate if requested */
1252     if (s->flags & PA_STREAM_INTERPOLATE_TIMING) {
1253
1254         /* We just add the time that passed since the latency info was
1255          * current */
1256         if (!s->corked && s->timing_info.playing) {
1257             struct timeval now;
1258             usec += pa_timeval_diff(pa_gettimeofday(&now), &s->timing_info.timestamp);
1259         }
1260     }
1261
1262     /* Make sure the time runs monotonically */
1263     if (!(s->flags & PA_STREAM_NOT_MONOTONOUS)) {
1264         if (usec < s->previous_time)
1265             usec = s->previous_time;
1266         else
1267             s->previous_time = usec;
1268     }
1269
1270     if (r_usec)
1271         *r_usec = usec;
1272
1273     return 0;
1274 }
1275
1276 static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t a, pa_usec_t b, int *negative) {
1277     assert(s);
1278     assert(s->ref >= 1);
1279
1280     if (negative)
1281         *negative = 0;
1282
1283     if (a >= b)
1284         return a-b;
1285     else {
1286         if (negative && s->direction == PA_STREAM_RECORD) {
1287             *negative = 1;
1288             return b-a;
1289         } else
1290             return 0;
1291     }
1292 }
1293
1294 int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
1295     pa_usec_t t, c;
1296     int r;
1297     int64_t cindex;
1298
1299     assert(s);
1300     assert(s->ref >= 1);
1301     assert(r_usec);
1302
1303     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1304     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1305     PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
1306     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
1307     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
1308
1309     if ((r = pa_stream_get_time(s, &t)) < 0)
1310         return r;
1311
1312     if (s->direction == PA_STREAM_PLAYBACK)
1313         cindex = s->timing_info.write_index;
1314     else
1315         cindex = s->timing_info.read_index;
1316
1317     if (cindex < 0)
1318         cindex = 0;
1319
1320     c = pa_bytes_to_usec(cindex, &s->sample_spec);
1321
1322     if (s->direction == PA_STREAM_PLAYBACK)
1323         *r_usec = time_counter_diff(s, c, t, negative);
1324     else
1325         *r_usec = time_counter_diff(s, t, c, negative);
1326
1327     return 0;
1328 }
1329
1330 const pa_timing_info* pa_stream_get_timing_info(pa_stream *s) {
1331     assert(s);
1332     assert(s->ref >= 1);
1333
1334     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1335     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1336     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->timing_info_valid, PA_ERR_BADSTATE);
1337
1338     return &s->timing_info;
1339 }
1340
1341 const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s) {
1342     assert(s);
1343     assert(s->ref >= 1);
1344
1345     return &s->sample_spec;
1346 }
1347
1348 const pa_channel_map* pa_stream_get_channel_map(pa_stream *s) {
1349     assert(s);
1350     assert(s->ref >= 1);
1351
1352     return &s->channel_map;
1353 }
1354
1355 const pa_buffer_attr* pa_stream_get_buffer_attr(pa_stream *s) {
1356     assert(s);
1357     assert(s->ref >= 1);
1358
1359     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1360     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1361     PA_CHECK_VALIDITY_RETURN_NULL(s->context,
1362         pa_context_get_server_protocol_version(s->context) >= 9, PA_ERR_NODATA);
1363
1364     return &s->buffer_attr;
1365 }