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