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